23253461919

Committed 18 Mar 2026 03:45PM UTC coverage: 63.53% (+1.6%) from 61.9%

Build # 23253461919

Build Type

Pull #4966

github

Committed by

web-flow

Commit Message

Merge 7e5d417b9 into bce705147

Pull Request Pull Request #4966: feat: add pg delta declarative sync command

Coverage Stats

517 of 1437 new or added lines in 17 files covered. (35.98%)

8 existing lines in 4 files now uncovered.

9156 of 14412 relevant lines covered (63.53%)

6.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

54.78

/internal/db/pull/pull.go

package pull

import (
        "bytes"
        "context"
        _ "embed"
        "fmt"
        "math"
        "os"
        "path/filepath"
        "strconv"
        "strings"

        "github.com/go-errors/errors"
        "github.com/jackc/pgconn"
        "github.com/jackc/pgx/v4"
        "github.com/spf13/afero"
        "github.com/spf13/viper"
        "github.com/supabase/cli/internal/db/declarative"
        "github.com/supabase/cli/internal/db/diff"
        "github.com/supabase/cli/internal/db/dump"
        "github.com/supabase/cli/internal/db/start"
        "github.com/supabase/cli/internal/migration/format"
        "github.com/supabase/cli/internal/migration/list"
        "github.com/supabase/cli/internal/migration/new"
        "github.com/supabase/cli/internal/migration/repair"
        "github.com/supabase/cli/internal/utils"
        "github.com/supabase/cli/pkg/migration"
)

var (
        errMissing  = errors.New("No migrations found")
        errInSync   = errors.New("No schema changes found")
        errConflict = errors.Errorf("The remote database's migration history does not match local files in %s directory.", utils.MigrationsDir)
)

func Run(ctx context.Context, schema []string, config pgconn.Config, name string, usePgDelta bool, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
        // 1. Check postgres connection
        conn, err := utils.ConnectByConfig(ctx, config, options...)
        if err != nil {
                return err
        }
        defer conn.Close(context.Background())
        // In experimental mode, allow db pull to switch from migration-file output to
        // declarative-file output through pg-delta when explicitly requested.
        if usePgDelta {
                return pullDeclarativePgDelta(ctx, schema, config, fsys, options...)
        }
        if viper.GetBool("EXPERIMENTAL") {
                var buf bytes.Buffer
                if err := migration.DumpRole(ctx, config, &buf, dump.DockerExec); err != nil {
                        return err
                }
                if err := migration.DumpSchema(ctx, config, &buf, dump.DockerExec); err != nil {
                        return err
                }
                // TODO: handle managed schemas
                return format.WriteStructuredSchemas(ctx, &buf, fsys)
        }
        // 2. Pull schema
        timestamp := utils.GetCurrentTimestamp()
        path := new.GetMigrationPath(timestamp, name)
        if err := run(ctx, schema, path, conn, fsys); err != nil {
                return err
        }
        // 3. Insert a row to `schema_migrations`
        fmt.Fprintln(os.Stderr, "Schema written to "+utils.Bold(path))
        if shouldUpdate, err := utils.NewConsole().PromptYesNo(ctx, "Update remote migration history table?", true); err != nil {
                return err
        } else if shouldUpdate {
                return repair.UpdateMigrationTable(ctx, conn, []string{timestamp}, repair.Applied, false, fsys)
        }
        return nil
}

// pullDeclarativePgDelta exports remote schema into declarative SQL files by
// diffing against an empty shadow baseline with pg-delta declarative export.
//
// This path is separate from run() because it does not produce or update
// timestamped migration files.
func pullDeclarativePgDelta(ctx context.Context, schema []string, config pgconn.Config, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
        fmt.Fprintln(os.Stderr, "Preparing declarative schema export using pg-delta...")
        shadow, err := diff.CreateShadowDatabase(ctx, utils.Config.Db.ShadowPort)
        if err != nil {
                return err
        }
        defer utils.DockerRemove(shadow)
        if err := start.WaitForHealthyService(ctx, utils.Config.Db.HealthTimeout, shadow); err != nil {
                return err
        }
        shadowConfig := pgconn.Config{
                Host:     utils.Config.Hostname,
                Port:     utils.Config.Db.ShadowPort,
                User:     "postgres",
                Password: utils.Config.Db.Password,
                Database: "postgres",
        }
        formatOptions := ""
        if utils.Config.Experimental.PgDelta != nil {
                formatOptions = strings.TrimSpace(utils.Config.Experimental.PgDelta.FormatOptions)
        }
        exported, err := diff.DeclarativeExportPgDelta(ctx, shadowConfig, config, schema, formatOptions, options...)
        if err != nil {
                return err
        }
        if err := declarative.WriteDeclarativeSchemas(exported, fsys); err != nil {
                return err
        }
        fmt.Fprintln(os.Stderr, "Declarative schema written to "+utils.Bold(utils.GetDeclarativeDir()))
        return nil
}

func run(ctx context.Context, schema []string, path string, conn *pgx.Conn, fsys afero.Fs) error {
        config := conn.Config().Config
        // 1. Assert `supabase/migrations` and `schema_migrations` are in sync.
        if err := assertRemoteInSync(ctx, conn, fsys); errors.Is(err, errMissing) {
                // Ignore schemas flag when working on the initial pull
                if err = dumpRemoteSchema(ctx, path, config, fsys); err != nil {
                        return err
                }
                // Run a second pass to pull in changes from default privileges and managed schemas
                if err = diffRemoteSchema(ctx, nil, path, config, fsys); errors.Is(err, errInSync) {
                        err = nil
                }
                return err
        } else if err != nil {
                return err
        }
        // 2. Fetch remote schema changes
        return diffRemoteSchema(ctx, schema, path, config, fsys)
}

func dumpRemoteSchema(ctx context.Context, path string, config pgconn.Config, fsys afero.Fs) error {
        // Special case if this is the first migration
        fmt.Fprintln(os.Stderr, "Dumping schema from remote database...")
        if err := utils.MkdirIfNotExistFS(fsys, filepath.Dir(path)); err != nil {
                return err
        }
        f, err := fsys.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
        if err != nil {
                return errors.Errorf("failed to open dump file: %w", err)
        }
        defer f.Close()
        return migration.DumpSchema(ctx, config, f, dump.DockerExec)
}

func diffRemoteSchema(ctx context.Context, schema []string, path string, config pgconn.Config, fsys afero.Fs) error {
        // Diff remote db (source) & shadow db (target) and write it as a new migration.
        output, err := diff.DiffDatabase(ctx, schema, config, os.Stderr, fsys, diff.DiffSchemaMigra, false)
        if err != nil {
                return err
        }
        if trimmed := strings.TrimSpace(output); len(trimmed) == 0 {
                return errors.New(errInSync)
        }
        // Append to existing migration file since we run this after dump
        f, err := fsys.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
        if err != nil {
                return errors.Errorf("failed to open migration file: %w", err)
        }
        defer f.Close()
        if _, err := f.WriteString(output); err != nil {
                return errors.Errorf("failed to write migration file: %w", err)
        }
        return nil
}

func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {
        remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)
        if err != nil {
                return err
        }
        localMigrations, err := list.LoadLocalVersions(fsys)
        if err != nil {
                return err
        }
        // Find any mismatch between local and remote migrations
        var extraRemote, extraLocal []string
        for i, j := 0, 0; i < len(remoteMigrations) || j < len(localMigrations); {
                remoteTimestamp := math.MaxInt
                if i < len(remoteMigrations) {
                        if remoteTimestamp, err = strconv.Atoi(remoteMigrations[i]); err != nil {
                                i++
                                continue
                        }
                }
                localTimestamp := math.MaxInt
                if j < len(localMigrations) {
                        if localTimestamp, err = strconv.Atoi(localMigrations[j]); err != nil {
                                j++
                                continue
                        }
                }
                // Top to bottom chronological order
                if localTimestamp < remoteTimestamp {
                        extraLocal = append(extraLocal, localMigrations[j])
                        j++
                } else if remoteTimestamp < localTimestamp {
                        extraRemote = append(extraRemote, remoteMigrations[i])
                        i++
                } else {
                        i++
                        j++
                }
        }
        // Suggest delete local migrations / reset migration history
        if len(extraRemote)+len(extraLocal) > 0 {
                utils.CmdSuggestion = suggestMigrationRepair(extraRemote, extraLocal)
                return errors.New(errConflict)
        }
        if len(localMigrations) == 0 {
                return errors.New(errMissing)
        }
        return nil
}

func suggestMigrationRepair(extraRemote, extraLocal []string) string {
        result := fmt.Sprintln("\nMake sure your local git repo is up-to-date. If the error persists, try repairing the migration history table:")
        for _, version := range extraRemote {
                result += fmt.Sprintln(utils.Bold("supabase migration repair --status reverted " + version))
        }
        for _, version := range extraLocal {
                result += fmt.Sprintln(utils.Bold("supabase migration repair --status applied " + version))
        }
        return result
}

1	package pull
2
3	import (
4	"bytes"
5	"context"
6	_ "embed"
7	"fmt"
8	"math"
9	"os"
10	"path/filepath"
11	"strconv"
12	"strings"
13
14	"github.com/go-errors/errors"
15	"github.com/jackc/pgconn"
16	"github.com/jackc/pgx/v4"
17	"github.com/spf13/afero"
18	"github.com/spf13/viper"
19	"github.com/supabase/cli/internal/db/declarative"
20	"github.com/supabase/cli/internal/db/diff"
21	"github.com/supabase/cli/internal/db/dump"
22	"github.com/supabase/cli/internal/db/start"
23	"github.com/supabase/cli/internal/migration/format"
24	"github.com/supabase/cli/internal/migration/list"
25	"github.com/supabase/cli/internal/migration/new"
26	"github.com/supabase/cli/internal/migration/repair"
27	"github.com/supabase/cli/internal/utils"
28	"github.com/supabase/cli/pkg/migration"
29	)
30
31	var (
32	errMissing = errors.New("No migrations found")
33	errInSync = errors.New("No schema changes found")
34	errConflict = errors.Errorf("The remote database's migration history does not match local files in %s directory.", utils.MigrationsDir)
35	)
36
37	func Run(ctx context.Context, schema []string, config pgconn.Config, name string, usePgDelta bool, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {	2✔
38	// 1. Check postgres connection	2✔
39	conn, err := utils.ConnectByConfig(ctx, config, options...)	2✔
40	if err != nil {	3✔
41	return err	1✔
42	}	1✔
43	defer conn.Close(context.Background())	1✔
44	// In experimental mode, allow db pull to switch from migration-file output to	1✔
45	// declarative-file output through pg-delta when explicitly requested.	1✔
46	if usePgDelta {	1✔
NEW 47	return pullDeclarativePgDelta(ctx, schema, config, fsys, options...)	×
NEW 48	}	×
49	if viper.GetBool("EXPERIMENTAL") {	1✔
50	var buf bytes.Buffer	×
51	if err := migration.DumpRole(ctx, config, &buf, dump.DockerExec); err != nil {	×
52	return err	×
53	}	×
54	if err := migration.DumpSchema(ctx, config, &buf, dump.DockerExec); err != nil {	×
55	return err	×
56	}	×
57	// TODO: handle managed schemas
58	return format.WriteStructuredSchemas(ctx, &buf, fsys)	×
59	}
60	// 2. Pull schema
61	timestamp := utils.GetCurrentTimestamp()	1✔
62	path := new.GetMigrationPath(timestamp, name)	1✔
63	if err := run(ctx, schema, path, conn, fsys); err != nil {	2✔
64	return err	1✔
65	}	1✔
66	// 3. Insert a row to `schema_migrations`
67	fmt.Fprintln(os.Stderr, "Schema written to "+utils.Bold(path))	×
68	if shouldUpdate, err := utils.NewConsole().PromptYesNo(ctx, "Update remote migration history table?", true); err != nil {	×
69	return err	×
70	} else if shouldUpdate {	×
71	return repair.UpdateMigrationTable(ctx, conn, []string{timestamp}, repair.Applied, false, fsys)	×
72	}	×
73	return nil	×
74	}
75
76	// pullDeclarativePgDelta exports remote schema into declarative SQL files by
77	// diffing against an empty shadow baseline with pg-delta declarative export.
78	//
79	// This path is separate from run() because it does not produce or update
80	// timestamped migration files.
NEW 81	func pullDeclarativePgDelta(ctx context.Context, schema []string, config pgconn.Config, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {	×
NEW 82	fmt.Fprintln(os.Stderr, "Preparing declarative schema export using pg-delta...")	×
NEW 83	shadow, err := diff.CreateShadowDatabase(ctx, utils.Config.Db.ShadowPort)	×
NEW 84	if err != nil {	×
NEW 85	return err	×
NEW 86	}	×
NEW 87	defer utils.DockerRemove(shadow)	×
NEW 88	if err := start.WaitForHealthyService(ctx, utils.Config.Db.HealthTimeout, shadow); err != nil {	×
NEW 89	return err	×
NEW 90	}	×
NEW 91	shadowConfig := pgconn.Config{	×
NEW 92	Host: utils.Config.Hostname,	×
NEW 93	Port: utils.Config.Db.ShadowPort,	×
NEW 94	User: "postgres",	×
NEW 95	Password: utils.Config.Db.Password,	×
NEW 96	Database: "postgres",	×
NEW 97	}	×
NEW 98	formatOptions := ""	×
NEW 99	if utils.Config.Experimental.PgDelta != nil {	×
NEW 100	formatOptions = strings.TrimSpace(utils.Config.Experimental.PgDelta.FormatOptions)	×
NEW 101	}	×
NEW 102	exported, err := diff.DeclarativeExportPgDelta(ctx, shadowConfig, config, schema, formatOptions, options...)	×
NEW 103	if err != nil {	×
NEW 104	return err	×
NEW 105	}	×
NEW 106	if err := declarative.WriteDeclarativeSchemas(exported, fsys); err != nil {	×
NEW 107	return err	×
NEW 108	}	×
NEW 109	fmt.Fprintln(os.Stderr, "Declarative schema written to "+utils.Bold(utils.GetDeclarativeDir()))	×
NEW 110	return nil	×
111	}
112
113	func run(ctx context.Context, schema []string, path string, conn *pgx.Conn, fsys afero.Fs) error {	3✔
114	config := conn.Config().Config	3✔
115	// 1. Assert `supabase/migrations` and `schema_migrations` are in sync.	3✔
116	if err := assertRemoteInSync(ctx, conn, fsys); errors.Is(err, errMissing) {	4✔
117	// Ignore schemas flag when working on the initial pull	1✔
118	if err = dumpRemoteSchema(ctx, path, config, fsys); err != nil {	1✔
119	return err	×
120	}	×
121	// Run a second pass to pull in changes from default privileges and managed schemas
122	if err = diffRemoteSchema(ctx, nil, path, config, fsys); errors.Is(err, errInSync) {	1✔
123	err = nil	×
124	}	×
125	return err	1✔
126	} else if err != nil {	3✔
127	return err	1✔
128	}	1✔
129	// 2. Fetch remote schema changes
130	return diffRemoteSchema(ctx, schema, path, config, fsys)	1✔
131	}
132
133	func dumpRemoteSchema(ctx context.Context, path string, config pgconn.Config, fsys afero.Fs) error {	1✔
134	// Special case if this is the first migration	1✔
135	fmt.Fprintln(os.Stderr, "Dumping schema from remote database...")	1✔
136	if err := utils.MkdirIfNotExistFS(fsys, filepath.Dir(path)); err != nil {	1✔
137	return err	×
138	}	×
139	f, err := fsys.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_TRUNC, 0644)	1✔
140	if err != nil {	1✔
141	return errors.Errorf("failed to open dump file: %w", err)	×
142	}	×
143	defer f.Close()	1✔
144	return migration.DumpSchema(ctx, config, f, dump.DockerExec)	1✔
145	}
146
147	func diffRemoteSchema(ctx context.Context, schema []string, path string, config pgconn.Config, fsys afero.Fs) error {	2✔
148	// Diff remote db (source) & shadow db (target) and write it as a new migration.	2✔
149	output, err := diff.DiffDatabase(ctx, schema, config, os.Stderr, fsys, diff.DiffSchemaMigra, false)	2✔
150	if err != nil {	4✔
151	return err	2✔
152	}	2✔
153	if trimmed := strings.TrimSpace(output); len(trimmed) == 0 {	×
154	return errors.New(errInSync)	×
155	}	×
156	// Append to existing migration file since we run this after dump
157	f, err := fsys.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_APPEND, 0644)	×
158	if err != nil {	×
159	return errors.Errorf("failed to open migration file: %w", err)	×
160	}	×
161	defer f.Close()	×
162	if _, err := f.WriteString(output); err != nil {	×
163	return errors.Errorf("failed to write migration file: %w", err)	×
164	}	×
165	return nil	×
166	}
167
168	func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {	7✔
169	remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)	7✔
170	if err != nil {	8✔
171	return err	1✔
172	}	1✔
173	localMigrations, err := list.LoadLocalVersions(fsys)	6✔
174	if err != nil {	7✔
175	return err	1✔
176	}	1✔
177	// Find any mismatch between local and remote migrations
178	var extraRemote, extraLocal []string	5✔
179	for i, j := 0, 0; i < len(remoteMigrations) \|\| j < len(localMigrations); {	9✔
180	remoteTimestamp := math.MaxInt	4✔
181	if i < len(remoteMigrations) {	7✔
182	if remoteTimestamp, err = strconv.Atoi(remoteMigrations[i]); err != nil {	3✔
183	i++	×
184	continue	×
185	}
186	}
187	localTimestamp := math.MaxInt	4✔
188	if j < len(localMigrations) {	7✔
189	if localTimestamp, err = strconv.Atoi(localMigrations[j]); err != nil {	3✔
190	j++	×
191	continue	×
192	}
193	}
194	// Top to bottom chronological order
195	if localTimestamp < remoteTimestamp {	6✔
196	extraLocal = append(extraLocal, localMigrations[j])	2✔
197	j++	2✔
198	} else if remoteTimestamp < localTimestamp {	5✔
199	extraRemote = append(extraRemote, remoteMigrations[i])	1✔
200	i++	1✔
201	} else {	2✔
202	i++	1✔
203	j++	1✔
204	}	1✔
205	}
206	// Suggest delete local migrations / reset migration history
207	if len(extraRemote)+len(extraLocal) > 0 {	7✔
208	utils.CmdSuggestion = suggestMigrationRepair(extraRemote, extraLocal)	2✔
209	return errors.New(errConflict)	2✔
210	}	2✔
211	if len(localMigrations) == 0 {	5✔
212	return errors.New(errMissing)	2✔
213	}	2✔
214	return nil	1✔
215	}
216
217	func suggestMigrationRepair(extraRemote, extraLocal []string) string {	2✔
218	result := fmt.Sprintln("\nMake sure your local git repo is up-to-date. If the error persists, try repairing the migration history table:")	2✔
219	for _, version := range extraRemote {	3✔
220	result += fmt.Sprintln(utils.Bold("supabase migration repair --status reverted " + version))	1✔
221	}	1✔
222	for _, version := range extraLocal {	4✔
223	result += fmt.Sprintln(utils.Bold("supabase migration repair --status applied " + version))	2✔
224	}	2✔
225	return result	2✔
226	}

supabase / cli / 23253461919

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous