17692433225

Committed 13 Sep 2025 05:37AM UTC coverage: 54.818% (-0.2%) from 55.003%

Build # 17692433225

Build Type

push

github

Committed by

web-flow

Commit Message

fix: pull changes from managed schemas separately (#4155)

* fix: pull managed schemas automatically

* Update diff_test.go

Run Details

18 of 63 new or added lines in 1 file covered. (28.57%)

3 existing lines in 2 files now uncovered.

6263 of 11425 relevant lines covered (54.82%)

6.08 hits per line

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

59.52

/internal/db/pull/pull.go

package pull

import (
        "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/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/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)
        managedSchemas = []string{"auth", "storage", "realtime"}
)

func Run(ctx context.Context, schema []string, config pgconn.Config, name string, 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())
        // 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
}

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

supabase / cli / 17692433225

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