21473922934

Committed 29 Jan 2026 10:05AM UTC coverage: 55.543% (-0.6%) from 56.148%

Build # 21473922934

Build Type

Pull #4722

github

Committed by

web-flow

Commit Message

Merge 0c72ab1a1 into 230667a9f

Pull Request Pull Request #4722: feat: experimental db pull as declarative schemas

Run Details

259 of 599 new or added lines in 2 files covered. (43.24%)

38 existing lines in 3 files now uncovered.

7109 of 12799 relevant lines covered (55.54%)

6.33 hits per line

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

68.03

/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/diff"
        "github.com/supabase/cli/internal/db/dump"
        "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, 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())
        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
}

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)
        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/diff"
20	"github.com/supabase/cli/internal/db/dump"
21	"github.com/supabase/cli/internal/migration/format"
22	"github.com/supabase/cli/internal/migration/list"
23	"github.com/supabase/cli/internal/migration/new"
24	"github.com/supabase/cli/internal/migration/repair"
25	"github.com/supabase/cli/internal/utils"
26	"github.com/supabase/cli/pkg/migration"
27	)
28
29	var (
30	errMissing = errors.New("No migrations found")
31	errInSync = errors.New("No schema changes found")
32	errConflict = errors.Errorf("The remote database's migration history does not match local files in %s directory.", utils.MigrationsDir)
33	)
34
35	func Run(ctx context.Context, schema []string, config pgconn.Config, name string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {	2✔
36	// 1. Check postgres connection	2✔
37	conn, err := utils.ConnectByConfig(ctx, config, options...)	2✔
38	if err != nil {	3✔
39	return err	1✔
40	}	1✔
41	defer conn.Close(context.Background())	1✔
42	if viper.GetBool("EXPERIMENTAL") {	1✔
NEW 43	var buf bytes.Buffer	×
NEW 44	if err := migration.DumpRole(ctx, config, &buf, dump.DockerExec); err != nil {	×
NEW 45	return err	×
NEW 46	}	×
NEW 47	if err := migration.DumpSchema(ctx, config, &buf, dump.DockerExec); err != nil {	×
NEW 48	return err	×
NEW 49	}	×
50	// TODO: handle managed schemas
NEW 51	return format.WriteStructuredSchemas(ctx, &buf, fsys)	×
52	}
53	// 2. Pull schema
54	timestamp := utils.GetCurrentTimestamp()	1✔
55	path := new.GetMigrationPath(timestamp, name)	1✔
56	if err := run(ctx, schema, path, conn, fsys); err != nil {	2✔
57	return err	1✔
58	}	1✔
59	// 3. Insert a row to `schema_migrations`
60	fmt.Fprintln(os.Stderr, "Schema written to "+utils.Bold(path))	×
61	if shouldUpdate, err := utils.NewConsole().PromptYesNo(ctx, "Update remote migration history table?", true); err != nil {	×
62	return err	×
63	} else if shouldUpdate {	×
64	return repair.UpdateMigrationTable(ctx, conn, []string{timestamp}, repair.Applied, false, fsys)	×
65	}	×
66	return nil	×
67	}
68
69	func run(ctx context.Context, schema []string, path string, conn *pgx.Conn, fsys afero.Fs) error {	3✔
70	config := conn.Config().Config	3✔
71	// 1. Assert `supabase/migrations` and `schema_migrations` are in sync.	3✔
72	if err := assertRemoteInSync(ctx, conn, fsys); errors.Is(err, errMissing) {	4✔
73	// Ignore schemas flag when working on the initial pull	1✔
74	if err = dumpRemoteSchema(ctx, path, config, fsys); err != nil {	1✔
75	return err	×
76	}	×
77	// Run a second pass to pull in changes from default privileges and managed schemas
78	if err = diffRemoteSchema(ctx, nil, path, config, fsys); errors.Is(err, errInSync) {	1✔
79	err = nil	×
80	}	×
81	return err	1✔
82	} else if err != nil {	3✔
83	return err	1✔
84	}	1✔
85	// 2. Fetch remote schema changes
86	return diffRemoteSchema(ctx, schema, path, config, fsys)	1✔
87	}
88
89	func dumpRemoteSchema(ctx context.Context, path string, config pgconn.Config, fsys afero.Fs) error {	1✔
90	// Special case if this is the first migration	1✔
91	fmt.Fprintln(os.Stderr, "Dumping schema from remote database...")	1✔
92	if err := utils.MkdirIfNotExistFS(fsys, filepath.Dir(path)); err != nil {	1✔
93	return err	×
94	}	×
95	f, err := fsys.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_TRUNC, 0644)	1✔
96	if err != nil {	1✔
97	return errors.Errorf("failed to open dump file: %w", err)	×
98	}	×
99	defer f.Close()	1✔
100	return migration.DumpSchema(ctx, config, f, dump.DockerExec)	1✔
101	}
102
103	func diffRemoteSchema(ctx context.Context, schema []string, path string, config pgconn.Config, fsys afero.Fs) error {	2✔
104	// Diff remote db (source) & shadow db (target) and write it as a new migration.	2✔
105	output, err := diff.DiffDatabase(ctx, schema, config, os.Stderr, fsys, diff.DiffSchemaMigra)	2✔
106	if err != nil {	4✔
107	return err	2✔
108	}	2✔
109	if trimmed := strings.TrimSpace(output); len(trimmed) == 0 {	×
110	return errors.New(errInSync)	×
111	}	×
112	// Append to existing migration file since we run this after dump
113	f, err := fsys.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_APPEND, 0644)	×
114	if err != nil {	×
115	return errors.Errorf("failed to open migration file: %w", err)	×
116	}	×
117	defer f.Close()	×
118	if _, err := f.WriteString(output); err != nil {	×
119	return errors.Errorf("failed to write migration file: %w", err)	×
120	}	×
121	return nil	×
122	}
123
124	func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {	7✔
125	remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)	7✔
126	if err != nil {	8✔
127	return err	1✔
128	}	1✔
129	localMigrations, err := list.LoadLocalVersions(fsys)	6✔
130	if err != nil {	7✔
131	return err	1✔
132	}	1✔
133	// Find any mismatch between local and remote migrations
134	var extraRemote, extraLocal []string	5✔
135	for i, j := 0, 0; i < len(remoteMigrations) \|\| j < len(localMigrations); {	9✔
136	remoteTimestamp := math.MaxInt	4✔
137	if i < len(remoteMigrations) {	7✔
138	if remoteTimestamp, err = strconv.Atoi(remoteMigrations[i]); err != nil {	3✔
139	i++	×
140	continue	×
141	}
142	}
143	localTimestamp := math.MaxInt	4✔
144	if j < len(localMigrations) {	7✔
145	if localTimestamp, err = strconv.Atoi(localMigrations[j]); err != nil {	3✔
146	j++	×
147	continue	×
148	}
149	}
150	// Top to bottom chronological order
151	if localTimestamp < remoteTimestamp {	6✔
152	extraLocal = append(extraLocal, localMigrations[j])	2✔
153	j++	2✔
154	} else if remoteTimestamp < localTimestamp {	5✔
155	extraRemote = append(extraRemote, remoteMigrations[i])	1✔
156	i++	1✔
157	} else {	2✔
158	i++	1✔
159	j++	1✔
160	}	1✔
161	}
162	// Suggest delete local migrations / reset migration history
163	if len(extraRemote)+len(extraLocal) > 0 {	7✔
164	utils.CmdSuggestion = suggestMigrationRepair(extraRemote, extraLocal)	2✔
165	return errors.New(errConflict)	2✔
166	}	2✔
167	if len(localMigrations) == 0 {	5✔
168	return errors.New(errMissing)	2✔
169	}	2✔
170	return nil	1✔
171	}
172
173	func suggestMigrationRepair(extraRemote, extraLocal []string) string {	2✔
174	result := fmt.Sprintln("\nMake sure your local git repo is up-to-date. If the error persists, try repairing the migration history table:")	2✔
175	for _, version := range extraRemote {	3✔
176	result += fmt.Sprintln(utils.Bold("supabase migration repair --status reverted " + version))	1✔
177	}	1✔
178	for _, version := range extraLocal {	4✔
179	result += fmt.Sprintln(utils.Bold("supabase migration repair --status applied " + version))	2✔
180	}	2✔
181	return result	2✔
182	}

supabase / cli / 21473922934

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