fix(cli): repair remote db pull with pg-delta and per-connection role step-down

apps/cli-go/docs/supabase/db/diff.md

@@ -10,6 +10,8 @@ By default, all schemas in the target database are diffed. Use the `--schema pub
 
 Projects created by a recent `supabase init` default to the pg-delta diff engine (`[experimental.pgdelta] enabled = true` in `config.toml`). Existing projects are unaffected and keep using migra unless they opt in. To fall back to the legacy migra engine, set `enabled = false` under `[experimental.pgdelta]`, or pass `--use-migra` for a single run.
 
+With the pg-delta engine the diff SQL is formatted by default with the same settings the declarative export uses (uppercase keywords, wrapped at a max width of 180, indented and column-aligned); execution-aware transaction boundaries are preserved as per-unit header comments in the output. Configure overrides with `[experimental.pgdelta] format_options`, or set `format_options = "null"` to emit raw, unformatted statements.
+
 While the diff command is able to capture most schema changes, there are cases where it is known to fail. Currently, this could happen if you schema contains:
 
 - Changes to publication

apps/cli-go/docs/supabase/db/pull.md

@@ -12,6 +12,10 @@ If no entries exist in the migration history table, the default diff engine uses
 
 Pass `--diff-engine pg-delta` to keep the migration-file `db pull` workflow while using pg-delta for the shadow diff step. On initial pull, pg-delta replaces `pg_dump` and produces the full migration from the shadow diff alone. Pass `--declarative` to switch to the declarative pg-delta export workflow instead.
 
+pg-delta plans are execution-aware: when a plan crosses a transaction boundary — for example `ALTER TYPE ... ADD VALUE` followed by a statement that uses the new enum value, which cannot run in the same transaction — `db pull` writes one ordered migration file per plan unit instead of a single file (for example `<ts>_remote_schema_schema_changes.sql` and `<ts+1s>_remote_schema_after_enum_values.sql`), each recorded in the migration history. The common case (a single unit) still produces exactly one `<ts>_remote_schema.sql` file.
+
+By default the emitted SQL is formatted with the same settings the declarative export uses (uppercase keywords, wrapped at a max width of 180, indented and column-aligned). Configure overrides with `[experimental.pgdelta] format_options` in `config.toml`, or set `format_options = "null"` to opt out and emit raw, unformatted statements.
+
 When `[experimental.pgdelta] enabled = true` (the default for projects created by a recent `supabase init`), the migration-file `db pull` workflow uses pg-delta for the shadow diff step by default; it does not switch to declarative output. Existing projects without the section are unaffected and keep using migra. To fall back to the legacy migra engine, set `enabled = false` under `[experimental.pgdelta]`, or pass `--diff-engine migra` for a single run.
 
 When pulling from a remote database with `--db-url`, prefer a direct connection (`db.<project-ref>.supabase.co:5432`) over the connection pooler so pg-delta can introspect the full catalog reliably.

apps/cli-go/internal/db/declarative/declarative.go

@@ -204,7 +204,7 @@ func SyncToMigrations(ctx context.Context, schema []string, file string, noCache
 	if len(strings.TrimSpace(file)) == 0 {
 		file = "declarative_sync"
 	}
-	if err := diff.SaveDiff(result.DiffSQL, file, fsys); err != nil {
+	if err := diff.SaveDiff(diff.DatabaseDiff{SQL: result.DiffSQL}, file, fsys); err != nil {
 		return err
 	}
 	if len(result.DropWarnings) > 0 {

apps/cli-go/internal/db/diff/diff.go

@@ -38,7 +38,7 @@ func Run(ctx context.Context, schema []string, file string, config pgconn.Config
 	out := result.SQL
 	branch := utils.GetGitBranch(fsys)
 	fmt.Fprintln(os.Stderr, "Finished "+utils.Aqua("supabase db diff")+" on branch "+utils.Aqua(branch)+".\n")
-	if err := SaveDiff(out, file, fsys); err != nil {
+	if err := SaveDiff(result, file, fsys); err != nil {
 		return err
 	}
 	drops := findDropStatements(out)
@@ -225,22 +225,31 @@ func DiffDatabase(ctx context.Context, schema []string, config pgconn.Config, w
 	} else {
 		fmt.Fprintln(w, "Diffing schemas...")
 	}
-	if IsPgDeltaDebugEnabled() && usePgDelta {
-		// Capture the shadow baseline catalog and edge-runtime stderr so an
-		// empty diff can be inspected later. DiffPgDeltaRefDetailed mirrors the
-		// pg-delta differ but additionally surfaces stderr, which differ() drops.
-		debugCapture := &PgDeltaDebugCapture{}
-		if snapshot, exportErr := exportCatalogPgDelta(ctx, utils.ToPostgresURL(shadowConfig), "postgres", options...); exportErr == nil {
-			debugCapture.SourceCatalog = snapshot
-		} else {
-			fmt.Fprintf(w, "Warning: failed to export shadow pg-delta catalog: %v\n", exportErr)
+	if usePgDelta {
+		// pg-delta always goes through the diffPgDeltaRefDetailed seam so callers get
+		// the execution-aware per-unit files (db pull writes one migration file each);
+		// db diff/declarative flatten them back via SQL. This mirrors the config-based
+		// differ (DiffPgDelta) exactly, so it is safe to bypass the injected differ()
+		// here — differ() remains the migra engine path below.
+		var debugCapture *PgDeltaDebugCapture
+		if IsPgDeltaDebugEnabled() {
+			// Capture the shadow baseline catalog and edge-runtime stderr so an
+			// empty diff can be inspected later.
+			debugCapture = &PgDeltaDebugCapture{}
+			if snapshot, exportErr := exportCatalogPgDelta(ctx, utils.ToPostgresURL(shadowConfig), "postgres", options...); exportErr == nil {
+				debugCapture.SourceCatalog = snapshot
+			} else {
+				fmt.Fprintf(w, "Warning: failed to export shadow pg-delta catalog: %v\n", exportErr)
+			}
 		}
-		result, err := DiffPgDeltaRefDetailed(ctx, utils.ToPostgresURL(shadowConfig), utils.ToPostgresURL(config), schema, pgDeltaFormatOptions(), options...)
+		result, err := diffPgDeltaRefDetailed(ctx, utils.ToPostgresURL(shadowConfig), utils.ToPostgresURL(config), schema, pgDeltaFormatOptions(), options...)
 		if err != nil {
 			return DatabaseDiff{}, err
 		}
-		debugCapture.Stderr = result.Stderr
-		return DatabaseDiff{SQL: result.SQL, Debug: debugCapture}, nil
+		if debugCapture != nil {
+			debugCapture.Stderr = result.Stderr
+		}
+		return DatabaseDiff{SQL: joinPgDeltaFiles(result.Files), Files: result.Files, Debug: debugCapture}, nil
 	}
 	output, err := differ(ctx, shadowConfig, config, schema, options...)
 	if err != nil {

apps/cli-go/internal/db/diff/diff_test.go

@@ -224,12 +224,23 @@ func TestRun(t *testing.T) {
 			Reply("CREATE FUNCTION").
 			Query(tableSQL).
 			Reply("CREATE TABLE")
+		// pg-delta bypasses the injected DiffFunc and runs the real edge-runtime
+		// pipeline, so stub the seam DiffDatabase uses (mirrors exportCatalogPgDelta).
+		// The migra differ must never be reached on this path.
+		originalDiffPgDelta := diffPgDeltaRefDetailed
+		t.Cleanup(func() { diffPgDeltaRefDetailed = originalDiffPgDelta })
 		diffCalled := false
-		differ := func(_ context.Context, _, target pgconn.Config, schema []string, _ ...func(*pgx.ConnConfig)) (string, error) {
+		diffPgDeltaRefDetailed = func(_ context.Context, _, targetRef string, schema []string, _ string, _ ...func(*pgx.ConnConfig)) (PgDeltaDiffResult, error) {
 			diffCalled = true
-			assert.Equal(t, "contrib_regression", target.Database)
+			assert.Contains(t, targetRef, "contrib_regression")
 			assert.Equal(t, []string{"public"}, schema)
-			return generated, nil
+			return PgDeltaDiffResult{
+				Files: []PgDeltaPlanFile{{Order: 1, Name: "schema_changes", TransactionMode: "transactional", SQL: generated}},
+			}, nil
+		}
+		differ := func(context.Context, pgconn.Config, pgconn.Config, []string, ...func(*pgx.ConnConfig)) (string, error) {
+			t.Fatal("migra differ must not be called on the pg-delta path")
+			return "", nil
 		}
 		localConfig := pgconn.Config{
 			Host:     utils.Config.Hostname,

apps/cli-go/internal/db/diff/pgadmin.go

@@ -5,6 +5,7 @@ import (
 	_ "embed"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
@@ -17,13 +18,26 @@ import (
 var warnDiff = `WARNING: The diff tool is not foolproof, so you may need to manually rearrange and modify the generated migration.
 Run ` + utils.Aqua("supabase db reset") + ` to verify that the new migration does not generate errors.`
 
-func SaveDiff(out, file string, fsys afero.Fs) error {
+func SaveDiff(result DatabaseDiff, file string, fsys afero.Fs) error {
+	out := result.SQL
 	if len(out) < 2 {
 		fmt.Fprintln(os.Stderr, "No schema changes found")
 	} else if len(file) > 0 {
-		path := new.GetMigrationPath(utils.GetCurrentTimestamp(), file)
-		if err := utils.WriteFile(path, []byte(out), fsys); err != nil {
-			return err
+		// A pg-delta plan that crosses a transaction boundary yields more than one
+		// ordered unit; writing them into a single migration file would later fail
+		// when `db push`/`reset` applies it as one transaction. Write one migration
+		// file per unit in that case (Go's `WritePgDeltaMigrations`). The migra /
+		// pgadmin engines and single-unit pg-delta plans keep the exact single-file
+		// path, byte-identical to before.
+		if len(result.Files) > 1 {
+			if _, err := WritePgDeltaMigrations(result.Files, time.Now(), file, fsys); err != nil {
+				return err
+			}
+		} else {
+			path := new.GetMigrationPath(utils.GetCurrentTimestamp(), file)
+			if err := utils.WriteFile(path, []byte(out), fsys); err != nil {
+				return err
+			}
 		}
 		fmt.Fprintln(os.Stderr, warnDiff)
 	} else {
@@ -44,7 +58,7 @@ func RunPgAdmin(ctx context.Context, schema []string, file string, config pgconn
 		return err
 	}
 
-	return SaveDiff(output, file, fsys)
+	return SaveDiff(DatabaseDiff{SQL: output}, file, fsys)
 }
 
 var output string

apps/cli-go/internal/db/diff/pgadmin_test.go

@@ -0,0 +1,88 @@
+package diff
+
+import (
+	"testing"
+
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/supabase/cli/internal/utils"
+)
+
+func TestSaveDiff(t *testing.T) {
+	t.Run("reports no changes on empty diff", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		require.NoError(t, SaveDiff(DatabaseDiff{SQL: ""}, "my_diff", fsys))
+		// Nothing written when there are no schema changes.
+		entries, err := afero.ReadDir(fsys, utils.MigrationsDir)
+		assert.Error(t, err)
+		assert.Empty(t, entries)
+	})
+
+	t.Run("writes a single migration file for a single-unit plan", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		files := []PgDeltaPlanFile{
+			{Order: 1, Name: "schema_changes", TransactionMode: "transactional", SQL: "create table a ();"},
+		}
+		result := DatabaseDiff{SQL: joinPgDeltaFiles(files), Files: files}
+		require.NoError(t, SaveDiff(result, "my_diff", fsys))
+		entries, err := afero.ReadDir(fsys, utils.MigrationsDir)
+		require.NoError(t, err)
+		require.Len(t, entries, 1)
+		// A single-unit plan keeps the plain `<ts>_<name>.sql` name and the exact
+		// diff SQL, byte-identical to the pre-multi-file behavior (no trailing newline
+		// added, no unit-name suffix).
+		assert.Regexp(t, `^\d{14}_my_diff\.sql$`, entries[0].Name())
+		contents, err := afero.ReadFile(fsys, utils.MigrationsDir+"/"+entries[0].Name())
+		require.NoError(t, err)
+		assert.Equal(t, "create table a ();", string(contents))
+	})
+
+	t.Run("writes one migration file per unit for a multi-unit plan", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		files := []PgDeltaPlanFile{
+			{Order: 1, Name: "schema_changes", TransactionMode: "transactional", SQL: "alter type mood add value 'ok';"},
+			{Order: 2, Name: "after_enum_values", TransactionMode: "transactional", SQL: "insert into t values ('ok');"},
+		}
+		result := DatabaseDiff{SQL: joinPgDeltaFiles(files), Files: files}
+		require.NoError(t, SaveDiff(result, "my_diff", fsys))
+		entries, err := afero.ReadDir(fsys, utils.MigrationsDir)
+		require.NoError(t, err)
+		require.Len(t, entries, 2)
+		// Multi-unit plans split into one ordered file per unit, each suffixed with the
+		// unit name, so `db push`/`reset` applies each unit as its own transaction.
+		assert.Regexp(t, `^\d{14}_my_diff_schema_changes\.sql$`, entries[0].Name())
+		assert.Regexp(t, `^\d{14}_my_diff_after_enum_values\.sql$`, entries[1].Name())
+	})
+
+	t.Run("prints diff to stdout when no file is given", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		require.NoError(t, SaveDiff(DatabaseDiff{SQL: "create table a ();"}, "", fsys))
+		entries, _ := afero.ReadDir(fsys, utils.MigrationsDir)
+		assert.Empty(t, entries)
+	})
+
+	t.Run("creates nested parent directories for a nested single-unit name", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		require.NoError(t, SaveDiff(DatabaseDiff{SQL: "create table a ();"}, "snapshots/remote", fsys))
+		matches, err := afero.Glob(fsys, utils.MigrationsDir+"/*_snapshots/remote.sql")
+		require.NoError(t, err)
+		require.Len(t, matches, 1)
+		contents, err := afero.ReadFile(fsys, matches[0])
+		require.NoError(t, err)
+		assert.Equal(t, "create table a ();", string(contents))
+	})
+
+	t.Run("creates nested parent directories for a nested multi-unit name", func(t *testing.T) {
+		fsys := afero.NewMemMapFs()
+		files := []PgDeltaPlanFile{
+			{Order: 1, Name: "schema_changes", TransactionMode: "transactional", SQL: "alter type mood add value 'ok';"},
+			{Order: 2, Name: "after_enum_values", TransactionMode: "transactional", SQL: "insert into t values ('ok');"},
+		}
+		result := DatabaseDiff{SQL: joinPgDeltaFiles(files), Files: files}
+		require.NoError(t, SaveDiff(result, "snapshots/remote", fsys))
+		matches, err := afero.Glob(fsys, utils.MigrationsDir+"/*_snapshots/remote_*.sql")
+		require.NoError(t, err)
+		require.Len(t, matches, 2)
+	})
+}

apps/cli-go/internal/db/diff/pgdelta.go

@@ -43,6 +43,34 @@ type DeclarativeOutput struct {
 	Files   []DeclarativeFile `json:"files"`
 }
 
+// PgDeltaPlanFile is one execution-aware migration unit rendered by pg-delta's
+// renderPlanFiles: a numbered SQL file whose header comments record the unit
+// number, transaction mode and boundary reason.
+type PgDeltaPlanFile struct {
+	Order           int    `json:"order"`
+	Name            string `json:"name"`
+	TransactionMode string `json:"transactionMode"`
+	SQL             string `json:"sql"`
+}
+
+// PgDeltaDiffOutput is the top-level diff envelope emitted by templates/pgdelta.ts.
+type PgDeltaDiffOutput struct {
+	Version int               `json:"version"`
+	Files   []PgDeltaPlanFile `json:"files"`
+}
+
+// joinPgDeltaFiles flattens the per-unit files back into a single SQL string for
+// callers (db diff, declarative sync) that consume one blob. The per-unit header
+// comments keep the transaction boundaries visible in the reviewed output; empty
+// files produce an empty string, preserving "no changes" detection.
+func joinPgDeltaFiles(files []PgDeltaPlanFile) string {
+	blocks := make([]string, len(files))
+	for i, file := range files {
+		blocks[i] = file.SQL
+	}
+	return strings.Join(blocks, "\n\n")
+}
+
 func isPostgresURL(ref string) bool {
 	return strings.HasPrefix(ref, "postgres://") || strings.HasPrefix(ref, "postgresql://")
 }
@@ -101,7 +129,7 @@ func DiffPgDeltaRef(ctx context.Context, sourceRef, targetRef string, schema []s
 	if err != nil {
 		return "", err
 	}
-	return result.SQL, nil
+	return joinPgDeltaFiles(result.Files), nil
 }
 
 // DiffPgDeltaRefDetailed is like DiffPgDeltaRef but also returns edge-runtime stderr.
@@ -136,15 +164,37 @@ func DiffPgDeltaRefDetailed(ctx context.Context, sourceRef, targetRef string, sc
 	if err := utils.RunEdgeRuntimeScript(ctx, env, script, binds, "error diffing schema", &stdout, &stderr, utils.PgDeltaNpmRegistryOption()); err != nil {
 		return PgDeltaDiffResult{}, err
 	}
-	return PgDeltaDiffResult{
-		SQL:    stdout.String(),
-		Stderr: stderr.String(),
-	}, nil
+	return parsePgDeltaDiffOutput(stdout.String(), stderr.String())
+}
+
+// parsePgDeltaDiffOutput turns the pg-delta diff script's stdout envelope into a
+// result. The template always prints the envelope on the success path, even for
+// an empty plan (`{"version":1,"files":[]}`); a truly empty stdout means no
+// envelope was produced, which we surface as "no changes" (empty Files) rather
+// than an error. Non-empty stdout that is not valid envelope JSON is a parse
+// error carrying the edge-runtime stderr for diagnosis.
+func parsePgDeltaDiffOutput(stdout, stderr string) (PgDeltaDiffResult, error) {
+	result := PgDeltaDiffResult{Stderr: stderr}
+	if len(strings.TrimSpace(stdout)) == 0 {
+		return result, nil
+	}
+	var envelope PgDeltaDiffOutput
+	if err := json.Unmarshal([]byte(stdout), &envelope); err != nil {
+		return PgDeltaDiffResult{}, errors.Errorf("failed to parse pg-delta diff output: %w:\n%s", err, stderr)
+	}
+	result.Files = envelope.Files
+	return result, nil
 }
 
 // exportCatalogPgDelta is overridden in tests to mock catalog export.
 var exportCatalogPgDelta = ExportCatalogPgDelta
 
+// diffPgDeltaRefDetailed is the seam DiffDatabase uses for the pg-delta engine.
+// Tests override it to stub the real edge-runtime pipeline (which the injected
+// DiffFunc differ cannot, since pg-delta bypasses differ), the same pattern as
+// exportCatalogPgDelta above.
+var diffPgDeltaRefDetailed = DiffPgDeltaRefDetailed
+
 // DeclarativeExportPgDelta exports target schema as declarative file payloads
 // while keeping a config-based API for existing call sites.
 func DeclarativeExportPgDelta(ctx context.Context, source, target pgconn.Config, schema []string, formatOptions string, options ...func(*pgx.ConnConfig)) (DeclarativeOutput, error) {

apps/cli-go/internal/db/diff/pgdelta_debug.go

@@ -17,9 +17,10 @@ func IsPgDeltaDebugEnabled() bool {
 	}
 }
 
-// PgDeltaDiffResult holds pg-delta diff output and edge-runtime stderr.
+// PgDeltaDiffResult holds the parsed pg-delta diff envelope (one file per
+// execution-aware plan unit) and the edge-runtime stderr.
 type PgDeltaDiffResult struct {
-	SQL    string
+	Files  []PgDeltaPlanFile
 	Stderr string
 }
 
@@ -31,7 +32,10 @@ type PgDeltaDebugCapture struct {
 
 // DatabaseDiff is the result of diffing a target database against a shadow baseline.
 type DatabaseDiff struct {
-	SQL   string
+	SQL string
+	// Files carries the per-unit pg-delta plan files (empty for the migra engine).
+	// SQL is the flattened join of these, kept for callers that consume one blob.
+	Files []PgDeltaPlanFile
 	Debug *PgDeltaDebugCapture
 }