fix: debt batch j05 (D-0760..)

internal/bundle/installer_appready.go

@@ -14,6 +14,7 @@ package bundle
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -72,6 +73,10 @@ func finishInstall(ctx context.Context, rn *appready.Runner, store *ledger.Store
 	if out == nil {
 		out = os.Stderr
 	}
+	prior, _, err := store.Load()
+	if err != nil {
+		return err
+	}
 	var names []string
 	_ = rn.Step(appready.StepLedger, func() (string, error) {
 		members, err := installedMembers(pluginDir, planned)
@@ -114,6 +119,9 @@ func finishInstall(ctx context.Context, rn *appready.Runner, store *ledger.Store
 		appready.Render(out, rep)
 	}
 	if f := rn.Failure(); f != nil {
+		if f.Step == appready.StepReconcile {
+			return errors.Join(appReadyVerdict(out, result.Bundle, f), rollbackReconcile(ctx, store, pluginDir, before, prior, opts, result, out))
+		}
 		return appReadyVerdict(out, result.Bundle, f)
 	}
 	return nil

internal/bundle/installer_reconcile_rollback.go

@@ -0,0 +1,54 @@
+package bundle
+
+// Purpose: undo fresh members after a refused reconcile without dropping kept schemas.
+// Inputs: pre-install membership and ledger, completed install result.
+// Outputs: rollback result and restored ledger records. Failed removals stay recorded.
+import (
+	"context"
+	"io"
+	"strings"
+
+	"github.com/nself-org/cli/internal/bundle/ledger"
+	"github.com/nself-org/cli/internal/plugin"
+)
+
+func rollbackReconcile(ctx context.Context, store *ledger.Store, dir string, before map[string]string, prior ledger.Ledger, opts InstallOpts, result *InstallResult, out io.Writer) error {
+	var fresh []string
+	for _, name := range result.Installed {
+		if _, existed := before[strings.ToLower(name)]; !existed {
+			fresh = append(fresh, name)
+		}
+	}
+	result.RolledBack = rollbackInstalled(ctx, opts.Config, plugin.UndoInstall, dir, fresh, out)
+	removed := make(map[string]bool)
+	for _, name := range result.RolledBack {
+		removed[strings.ToLower(name)] = true
+	}
+	return store.Update(func(l *ledger.Ledger) error {
+		// Restore the pre-install ledger, retaining any fresh member whose cleanup
+		// failed so that it remains discoverable for a later remove.
+		current := l.Clone()
+		*l = prior.Clone()
+		// Existing members may have been upgraded; retain their current facts
+		// while restoring the ownership references from before this install.
+		for key, record := range l.Plugins {
+			if now, ok := current.Plugins[key]; ok {
+				record.Tier, record.Version, record.Checksum = now.Tier, now.Version, now.Checksum
+				l.Plugins[key] = record
+			}
+		}
+		for _, name := range fresh {
+			key := strings.ToLower(name)
+			if removed[key] {
+				continue
+			}
+			if record, ok := current.Plugins[key]; ok {
+				l.Plugins[key] = record
+				for _, slug := range record.InstalledBy {
+					l.Bundles[slug] = current.Bundles[slug]
+				}
+			}
+		}
+		return nil
+	})
+}

internal/bundle/installer_reconcile_rollback_test.go

@@ -0,0 +1,57 @@
+package bundle
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/nself-org/cli/internal/bundle/appready"
+)
+
+func TestReconcileFailureRollsBackFreshMembers(t *testing.T) {
+	for _, refusal := range []string{"refused", "regeneration failed"} {
+		for _, keep := range []bool{false, true} {
+			t.Run(refusal+map[bool]string{false: "/fresh", true: "/existing"}[keep], func(t *testing.T) {
+				dir := isolate(t)
+				registry := mockNotificationsOnly()
+				if keep {
+					registry = mockTaskRegistry()
+				}
+				setupOfflineRegistry(t, registry)
+				if keep {
+					writeMember(t, dir, "jobs", 1, nil)
+				}
+				var out bytes.Buffer
+				opts := freeOpts(t, dir, &out, func(name string) error { writeMember(t, dir, name, 1, nil); return nil })
+				boom := errors.New(refusal)
+				opts.Hooks = &appready.Hooks{Reconcile: func(context.Context) error { return boom }}
+				res, err := Install(opCtx(t), "task", opts)
+				if !errors.Is(err, boom) {
+					t.Fatalf("error = %v", err)
+				}
+				_, statErr := os.Stat(filepath.Join(dir, "notifications"))
+				if !os.IsNotExist(statErr) {
+					t.Fatalf("fresh member remains: %v", statErr)
+				}
+				if keep {
+					if _, err := os.Stat(filepath.Join(dir, "jobs")); err != nil {
+						t.Fatal(err)
+					}
+				}
+				l, _, err := newLedgerStore(dir).Load()
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, ok := l.Plugins["notifications"]; ok {
+					t.Fatalf("ledger member present = %v, want %v", ok, false)
+				}
+				if len(res.RolledBack) != 1 {
+					t.Fatalf("rollback = %v", res.RolledBack)
+				}
+			})
+		}
+	}
+}

internal/importer/importer.go

@@ -14,9 +14,10 @@ type FlagSpec struct {
 	EnvVarName  bool
 }
 type Input struct {
-	Values map[string]string
-	Env    func(string) string
-	Target *config.Config
+	SkipStorage bool
+	Values      map[string]string
+	Env         func(string) string
+	Target      *config.Config
 }
 type Detection struct {
 	Kind    string   `json:"kind"`

internal/importer/run.go

@@ -114,6 +114,7 @@ func Run(ctx context.Context, cfg *config.Config, s Source, in Input, opts RunOp
 		defer release()
 	}
 	in.Target = cfg
+	in.SkipStorage = opts.SkipStorage
 	detection, err := s.Detect(ctx, in)
 	if err != nil {
 		return report, publicError(err, "E517", "source detection failed")

internal/importer/supabase/extract.go

@@ -24,6 +24,7 @@ import (
 
 // run carries what the snapshot hooks learn for the later steps.
 type run struct {
+	skipStorage  bool
 	api          *api
 	catalog      map[string]int64 // objects per bucket in storage.objects
 	total        int64
@@ -35,6 +36,9 @@ var projectRef = regexp.MustCompile(`^[a-z0-9]{1,40}$`)
 
 // apiFor builds the storage client from --api-url or --project-ref; nil when neither is set.
 func apiFor(in importer.Input) (*api, error) {
+	if in.SkipStorage {
+		return nil, nil
+	}
 	raw := in.Values["api-url"]
 	if raw == "" && in.Values["project-ref"] != "" {
 		if !projectRef.MatchString(in.Values["project-ref"]) {
@@ -67,14 +71,14 @@ func (Source) Extract(ctx context.Context, in importer.Input, stage string) (imp
 		return importer.Extracted{}, err
 	}
 	// Fail before the dump when storage objects exist but no storage API was configured.
-	if n, qerr := storageObjects(ctx, db); qerr != nil || (n > 0 && client == nil) {
+	if n, qerr := storageObjects(ctx, db); qerr != nil || (n > 0 && client == nil && !in.SkipStorage) {
 		_ = db.Close(ctx)
 		if qerr != nil {
 			return importer.Extracted{}, qerr
 		}
 		return importer.Extracted{}, errs.New("E520", "source has storage objects; --api-url (or --project-ref) and --service-key-env are required")
 	}
-	r := &run{api: client}
+	r := &run{api: client, skipStorage: in.SkipStorage}
 	ext, err := postgres.Snapshot(ctx, db, d, in.Target, stage, postgres.Options{Source: "supabase", InTx: r.inTx, Finish: r.finish})
 	ext.UntranslatedPolicies = r.untranslated
 	return ext, err

internal/importer/supabase/objects.go

@@ -21,7 +21,12 @@ const maxObjectBytes = 5 << 30
 
 var s3Bucket = regexp.MustCompile(`^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$`)
 
-func (r *run) catalogCounts(ctx context.Context, tx pgx.Tx, m *portable.Manifest) error {
+type storageCatalog interface {
+	QueryRow(context.Context, string, ...any) pgx.Row
+	Query(context.Context, string, ...any) (pgx.Rows, error)
+}
+
+func (r *run) catalogCounts(ctx context.Context, tx storageCatalog, m *portable.Manifest) error {
 	fail := errs.New("E520", "source storage lookup failed")
 	var has bool
 	if tx.QueryRow(ctx, `SELECT to_regclass('storage.objects') IS NOT NULL`).Scan(&has) != nil {
@@ -32,6 +37,24 @@ func (r *run) catalogCounts(ctx context.Context, tx pgx.Tx, m *portable.Manifest
 	if !has {
 		return nil
 	}
+	if r.skipStorage {
+		rows, err := tx.Query(ctx, `SELECT bucket_id::text, name::text FROM storage.objects ORDER BY 1, 2`)
+		if err != nil {
+			return fail
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var bucket, key string
+			if rows.Scan(&bucket, &key) != nil {
+				return fail
+			}
+			m.Exemptions = append(m.Exemptions, portable.Exemption{Kind: "object", ID: bucket + "/" + key, Reason: "--skip-storage"})
+		}
+		if rows.Err() != nil {
+			return fail
+		}
+		return nil
+	}
 	rows, err := tx.Query(ctx, `SELECT bucket_id::text, count(*) FROM storage.objects GROUP BY 1 ORDER BY 1`)
 	if err != nil {
 		return fail

internal/importer/supabase/objects_skip_test.go

@@ -0,0 +1,40 @@
+package supabase
+
+import (
+	"context"
+	"testing"
+
+	"github.com/nself-org/cli/internal/importer"
+	"github.com/nself-org/cli/internal/portable"
+)
+
+func TestSkipStorageExemptsSourceObjects(t *testing.T) {
+	for _, bucket := range []string{"avatars", "Invalid bucket!", "a"} {
+		t.Run(bucket, func(t *testing.T) {
+			db := detectionDB(t, true, []portable.Object{{Bucket: bucket, Key: "folder/photo.png"}})
+			r := &run{skipStorage: true}
+			m := portable.Manifest{}
+			if err := r.catalogCounts(context.Background(), db, &m); err != nil {
+				t.Fatal(err)
+			}
+			if len(m.Exemptions) != 1 || m.Exemptions[0].ID != bucket+"/folder/photo.png" || m.Exemptions[0].Reason != "--skip-storage" {
+				t.Fatalf("exemptions = %+v", m.Exemptions)
+			}
+			if err := r.finish(context.Background(), nil, &m); err != nil {
+				t.Fatal(err)
+			}
+			if len(m.Storage.Objects) != 0 {
+				t.Fatal("skipped objects were exported")
+			}
+		})
+	}
+}
+
+func TestSkipStorageNeedsNoAPICredentials(t *testing.T) {
+	for _, values := range []map[string]string{{}, {"api-url": "invalid"}, {"project-ref": "invalid!"}} {
+		a, err := apiFor(importer.Input{Values: values, SkipStorage: true})
+		if err != nil || a != nil {
+			t.Fatal("skipped storage requested API credentials")
+		}
+	}
+}

internal/importer/verify/auth_state_test.go

@@ -27,6 +27,14 @@ func TestCompareStateUserAuth(t *testing.T) {
 		miss bool
 	}{
 		{"identical", base, ok, false},
+		{"metadata changed", func() User {
+			b := base
+			b.Metadata = map[string]any{"preferences": map[string]any{"theme": "dark"}}
+			return b
+		}(), with(func(u *UserAuth) { u.Metadata = map[string]any{"preferences": map[string]any{"theme": "light"}} }), true},
+		{"metadata missing", func() User { b := base; b.Metadata = map[string]any{"a": true}; return b }(), ok, true},
+		{"metadata added", base, with(func(u *UserAuth) { u.Metadata = map[string]any{"a": true} }), true},
+		{"empty metadata default", base, with(func(u *UserAuth) { u.Metadata = map[string]any{} }), false},
 		{"default role drifted to user", base, with(func(u *UserAuth) { u.DefaultRole = "user" }), true},
 		{"email_verified lost", base, with(func(u *UserAuth) { u.EmailVerified = false }), true},
 		{"disabled lost", func() User { b := base; b.Disabled = true; return b }(), ok, true},

internal/importer/verify/compare.go

@@ -39,6 +39,7 @@ const DefaultRoleFallback = "user"
 
 // UserAuth is the auth state the verifier reads back from a target user.
 type UserAuth struct {
+	Metadata      map[string]any
 	DefaultRole   string
 	EmailVerified bool
 	Disabled      bool
@@ -111,7 +112,7 @@ func CompareState(state State, m portable.Manifest, users []User, metadata Metad
 		} else if got, have := state.UserIdent[u.ID]; state.UserIdent != nil && (!have || got != Identity(u.Email, u.Phone, u.DisplayName)) {
 			add("user", u.ID, "user identity fields differ", "user")
 		} else if got, have := state.UserAuth[u.ID]; state.UserAuth != nil && (!have || !authMatches(u, got)) {
-			add("user", u.ID, "user role or status fields differ", "user")
+			add("user", u.ID, "user role, status or metadata fields differ", "user")
 		}
 	}
 	for _, o := range m.Storage.Objects {

internal/importer/verify/identity.go

@@ -1,5 +1,7 @@
 package verify
 
+import "reflect"
+
 // Purpose: one canonical form of a user's nullable identity fields.
 // Inputs: email, phone, display name (nil means SQL NULL). Outputs: a comparable string.
 // Constraints: NULL and empty text stay distinct for email and phone, because hasura-auth
@@ -23,6 +25,9 @@ func Identity(email, phone, display *string) string {
 // email_verified, disabled and role set. A target role beyond the bundle's is a
 // mismatch, except the default role itself, which a target may add on its own.
 func authMatches(want User, got UserAuth) bool {
+	if (len(want.Metadata) != 0 || len(got.Metadata) != 0) && !reflect.DeepEqual(want.Metadata, got.Metadata) {
+		return false
+	}
 	wantDefault := want.DefaultRole
 	if wantDefault == "" {
 		wantDefault = DefaultRoleFallback

internal/importer/verify/live.go

@@ -57,15 +57,15 @@ func (l *Live) Snapshot(ctx context.Context, m portable.Manifest) (State, error)
 	if l.HasAuth || len(l.Users) > 0 {
 		s.UserIdent = map[string]string{}
 		s.UserAuth = map[string]UserAuth{}
-		rows, e := tx.Query(ctx, `SELECT id::text, password_hash, email::text, phone_number, display_name, default_role, email_verified, disabled FROM auth.users`)
+		rows, e := tx.Query(ctx, `SELECT id::text, password_hash, email::text, phone_number, display_name, default_role, email_verified, disabled, metadata FROM auth.users`)
 		if e != nil {
 			return s, errors.New("target users unavailable")
 		}
 		for rows.Next() {
 			var id string
 			var hash, email, phone, display *string
 			var ua UserAuth
-			if rows.Scan(&id, &hash, &email, &phone, &display, &ua.DefaultRole, &ua.EmailVerified, &ua.Disabled) != nil {
+			if rows.Scan(&id, &hash, &email, &phone, &display, &ua.DefaultRole, &ua.EmailVerified, &ua.Disabled, &ua.Metadata) != nil {
 				rows.Close()
 				return s, errors.New("target user read failed")
 			}

scripts/e2e/import-supabase.sh

@@ -137,6 +137,16 @@ jq -e --arg u "$UID1" '(.data.notes|length)==1 and .data.notes[0].user_id==$u' "
 # The auth.uid() default fills the owner column with the caller's id.
 gql '{"query":"mutation { insert_notes_one(object: {title: \"from hasura\"}) { user_id } }"}' >"$W/insert.json" || fail 'insert request'
 jq -e --arg u "$UID1" '.data.insert_notes_one.user_id==$u' "$W/insert.json" >/dev/null || fail 'insert did not fill user_id from auth.uid()'
+# Cross-user mutations must be rejected or affect zero rows.
+for operation in update delete; do
+  if [ "$operation" = update ]; then
+    mutation="mutation { update_notes(where: {user_id: {_neq: \"$UID1\"}}, _set: {title: \"forbidden\"}) { affected_rows } }"
+  else
+    mutation="mutation { delete_notes(where: {user_id: {_neq: \"$UID1\"}}) { affected_rows } }"
+  fi
+  gql "$(jq -nc --arg q "$mutation" '{query:$q}')" >"$W/$operation-denied.json" || fail "$operation request"
+  jq -e --arg field "${operation}_notes" '.errors!=null or .data[$field].affected_rows==0' "$W/$operation-denied.json" >/dev/null || fail "cross-user $operation allowed"
+done
 # The untranslated policy leaves the table closed to the user role.
 gql '{"query":"{ announcements { id } }"}' >"$W/announcements.json" || true
 jq -e '.errors!=null' "$W/announcements.json" >/dev/null || fail 'untranslated policy table is open'