23352230634

Committed 20 Mar 2026 04:25PM UTC coverage: 63.16% (+1.5%) from 61.699%

Build # 23352230634

Build Type

Pull #4966

github

Committed by

web-flow

Commit Message

Merge 5ff0bc8de into 17e02c85f

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

Coverage Stats

520 of 1449 new or added lines in 18 files covered. (35.89%)

39 existing lines in 4 files now uncovered.

9191 of 14552 relevant lines covered (63.16%)

6.85 hits per line

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

48.48

/internal/db/diff/diff.go

package diff

import (
        "bytes"
        "context"
        "fmt"
        "io"
        "io/fs"
        "os"
        "path/filepath"
        "regexp"
        "sort"
        "strconv"
        "strings"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/docker/docker/api/types/container"
        "github.com/docker/docker/api/types/network"
        "github.com/docker/go-connections/nat"
        "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/start"
        "github.com/supabase/cli/internal/pgdelta"
        "github.com/supabase/cli/internal/utils"
        "github.com/supabase/cli/pkg/migration"
        "github.com/supabase/cli/pkg/parser"
)

type DiffFunc func(context.Context, pgconn.Config, pgconn.Config, []string, ...func(*pgx.ConnConfig)) (string, error)

func Run(ctx context.Context, schema []string, file string, config pgconn.Config, differ DiffFunc, usePgDelta bool, fsys afero.Fs, options ...func(*pgx.ConnConfig)) (err error) {
        out, err := DiffDatabase(ctx, schema, config, os.Stderr, fsys, differ, usePgDelta, options...)
        if err != nil {
                return err
        }
        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 {
                return err
        }
        drops := findDropStatements(out)
        if len(drops) > 0 {
                fmt.Fprintln(os.Stderr, "Found drop statements in schema diff. Please double check if these are expected:")
                fmt.Fprintln(os.Stderr, utils.Yellow(strings.Join(drops, "\n")))
        }
        return nil
}

func loadDeclaredSchemas(fsys afero.Fs) ([]string, error) {
        // When pg-delta is enabled, declarative path is the source of truth (config or default).
        if utils.IsPgDeltaEnabled() {
                declDir := utils.GetDeclarativeDir()
                if exists, err := afero.DirExists(fsys, declDir); err == nil && exists {
                        var declared []string
                        if err := afero.Walk(fsys, declDir, func(path string, info fs.FileInfo, err error) error {
                                if err != nil {
                                        return err
                                }
                                if info.Mode().IsRegular() && filepath.Ext(info.Name()) == ".sql" {
                                        declared = append(declared, path)
                                }
                                return nil
                        }); err != nil {
                                return nil, errors.Errorf("failed to walk declarative dir: %w", err)
                        }
                        sort.Strings(declared)
                        return declared, nil
                }
        }
        if schemas := utils.Config.Db.Migrations.SchemaPaths; len(schemas) > 0 {
                return schemas.Files(afero.NewIOFS(fsys))
        }
        if exists, err := afero.DirExists(fsys, utils.SchemasDir); err != nil {
                return nil, errors.Errorf("failed to check schemas: %w", err)
        } else if !exists {
                return nil, nil
        }
        var declared []string
        if err := afero.Walk(fsys, utils.SchemasDir, func(path string, info fs.FileInfo, err error) error {
                if err != nil {
                        return err
                }
                if info.Mode().IsRegular() && filepath.Ext(info.Name()) == ".sql" {
                        declared = append(declared, path)
                }
                return nil
        }); err != nil {
                return nil, errors.Errorf("failed to walk dir: %w", err)
        }
        // Keep file application order deterministic so diff output stays stable across
        // filesystems and operating systems. This is only if no schema paths in config are set.
        sort.Strings(declared)
        return declared, nil
}

// https://github.com/djrobstep/migra/blob/master/migra/statements.py#L6
var dropStatementPattern = regexp.MustCompile(`(?i)drop\s+`)

func findDropStatements(out string) []string {
        lines, err := parser.SplitAndTrim(strings.NewReader(out))
        if err != nil {
                return nil
        }
        var drops []string
        for _, line := range lines {
                if dropStatementPattern.MatchString(line) {
                        drops = append(drops, line)
                }
        }
        return drops
}

func CreateShadowDatabase(ctx context.Context, port uint16) (string, error) {
        // Disable background workers in shadow database
        config := start.NewContainerConfig("-c", "max_worker_processes=0")
        hostPort := strconv.FormatUint(uint64(port), 10)
        hostConfig := container.HostConfig{
                PortBindings: nat.PortMap{"5432/tcp": []nat.PortBinding{{HostPort: hostPort}}},
                AutoRemove:   true,
        }
        networkingConfig := network.NetworkingConfig{}
        if utils.Config.Db.MajorVersion <= 14 {
                hostConfig.Tmpfs = map[string]string{"/docker-entrypoint-initdb.d": ""}
        }
        return utils.DockerStart(ctx, config, hostConfig, networkingConfig, "")
}

func ConnectShadowDatabase(ctx context.Context, timeout time.Duration, options ...func(*pgx.ConnConfig)) (conn *pgx.Conn, err error) {
        // Retry until connected, cancelled, or timeout
        policy := start.NewBackoffPolicy(ctx, timeout)
        config := pgconn.Config{Port: utils.Config.Db.ShadowPort}
        connect := func() (*pgx.Conn, error) {
                return utils.ConnectLocalPostgres(ctx, config, options...)
        }
        return backoff.RetryWithData(connect, policy)
}

// Required to bypass pg_cron check: https://github.com/citusdata/pg_cron/blob/main/pg_cron.sql#L3
const CREATE_TEMPLATE = "CREATE DATABASE contrib_regression TEMPLATE postgres"

func MigrateShadowDatabase(ctx context.Context, container string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
        migrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))
        if err != nil {
                return err
        }
        conn, err := ConnectShadowDatabase(ctx, 10*time.Second, options...)
        if err != nil {
                return err
        }
        defer conn.Close(context.Background())
        if err := start.SetupDatabase(ctx, conn, container[:12], os.Stderr, fsys); err != nil {
                return err
        }
        if _, err := conn.Exec(ctx, CREATE_TEMPLATE); err != nil {
                return errors.Errorf("failed to create template database: %w", err)
        }
        return migration.ApplyMigrations(ctx, migrations, conn, afero.NewIOFS(fsys))
}

func DiffDatabase(ctx context.Context, schema []string, config pgconn.Config, w io.Writer, fsys afero.Fs, differ DiffFunc, usePgDelta bool, options ...func(*pgx.ConnConfig)) (string, error) {
        fmt.Fprintln(w, "Creating shadow database...")
        shadow, err := 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
        }
        if err := MigrateShadowDatabase(ctx, shadow, fsys, options...); 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",
        }
        if utils.IsLocalDatabase(config) {
                if declared, err := loadDeclaredSchemas(fsys); len(declared) > 0 {
                        config = shadowConfig
                        config.Database = "contrib_regression"
                        if usePgDelta {
                                declDir := utils.GetDeclarativeDir()
                                if exists, _ := afero.DirExists(fsys, declDir); exists {
                                        if err := pgdelta.ApplyDeclarative(ctx, config, fsys); err != nil {
                                                return "", err
                                        }
                                } else {
                                        if err := migrateBaseDatabase(ctx, config, declared, fsys, options...); err != nil {
                                                return "", err
                                        }
                                }
                        } else {
                                if err := migrateBaseDatabase(ctx, config, declared, fsys, options...); err != nil {
                                        return "", err
                                }
                        }
                } else if err != nil {
                        return "", err
                }
        }
        // Load all user defined schemas
        if len(schema) > 0 {
                fmt.Fprintln(w, "Diffing schemas:", strings.Join(schema, ","))
        } else {
                fmt.Fprintln(w, "Diffing schemas...")
        }
        return differ(ctx, shadowConfig, config, schema, options...)
}

func migrateBaseDatabase(ctx context.Context, config pgconn.Config, migrations []string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
        fmt.Fprintln(os.Stderr, "Creating local database from declarative schemas:")
        msg := make([]string, len(migrations))
        for i, m := range migrations {
                msg[i] = fmt.Sprintf(" • %s", utils.Bold(m))
        }
        fmt.Fprintln(os.Stderr, strings.Join(msg, "\n"))
        conn, err := utils.ConnectLocalPostgres(ctx, config, options...)
        if err != nil {
                return err
        }
        defer conn.Close(context.Background())
        return migration.SeedGlobals(ctx, migrations, conn, afero.NewIOFS(fsys))
}

func diffWithStream(ctx context.Context, env []string, script string, stdout io.Writer) error {
        cmd := []string{"edge-runtime", "start", "--main-service=."}
        if viper.GetBool("DEBUG") {
                cmd = append(cmd, "--verbose")
        }
        cmdString := strings.Join(cmd, " ")
        entrypoint := []string{"sh", "-c", `cat <<'EOF' > index.ts && ` + cmdString + `
` + script + `
EOF
`}
        var stderr bytes.Buffer
        if err := utils.DockerRunOnceWithConfig(
                ctx,
                container.Config{
                        Image:      utils.Config.EdgeRuntime.Image,
                        Env:        env,
                        Entrypoint: entrypoint,
                },
                container.HostConfig{
                        Binds:       []string{utils.EdgeRuntimeId + ":/root/.cache/deno:rw"},
                        NetworkMode: network.NetworkHost,
                },
                network.NetworkingConfig{},
                "",
                stdout,
                &stderr,
        // The "main worker has been destroyed" message may not appear at the start of stderr
        // (e.g. preceded by other Deno runtime output), so use Contains instead of HasPrefix.
        ); err != nil && !strings.Contains(stderr.String(), "main worker has been destroyed") {
                return errors.Errorf("error diffing schema: %w:\n%s", err, stderr.String())
        }
        return nil
}

1	package diff
2
3	import (
4	"bytes"
5	"context"
6	"fmt"
7	"io"
8	"io/fs"
9	"os"
10	"path/filepath"
11	"regexp"
12	"sort"
13	"strconv"
14	"strings"
15	"time"
16
17	"github.com/cenkalti/backoff/v4"
18	"github.com/docker/docker/api/types/container"
19	"github.com/docker/docker/api/types/network"
20	"github.com/docker/go-connections/nat"
21	"github.com/go-errors/errors"
22	"github.com/jackc/pgconn"
23	"github.com/jackc/pgx/v4"
24	"github.com/spf13/afero"
25	"github.com/spf13/viper"
26	"github.com/supabase/cli/internal/db/start"
27	"github.com/supabase/cli/internal/pgdelta"
28	"github.com/supabase/cli/internal/utils"
29	"github.com/supabase/cli/pkg/migration"
30	"github.com/supabase/cli/pkg/parser"
31	)
32
33	type DiffFunc func(context.Context, pgconn.Config, pgconn.Config, []string, ...func(*pgx.ConnConfig)) (string, error)
34
35	func Run(ctx context.Context, schema []string, file string, config pgconn.Config, differ DiffFunc, usePgDelta bool, fsys afero.Fs, options ...func(*pgx.ConnConfig)) (err error) {	2✔
36	out, err := DiffDatabase(ctx, schema, config, os.Stderr, fsys, differ, usePgDelta, options...)	2✔
37	if err != nil {	3✔
38	return err	1✔
39	}	1✔
40	branch := utils.GetGitBranch(fsys)	1✔
41	fmt.Fprintln(os.Stderr, "Finished "+utils.Aqua("supabase db diff")+" on branch "+utils.Aqua(branch)+".\n")	1✔
42	if err := SaveDiff(out, file, fsys); err != nil {	1✔
43	return err	×
44	}	×
45	drops := findDropStatements(out)	1✔
46	if len(drops) > 0 {	1✔
47	fmt.Fprintln(os.Stderr, "Found drop statements in schema diff. Please double check if these are expected:")	×
48	fmt.Fprintln(os.Stderr, utils.Yellow(strings.Join(drops, "\n")))	×
49	}	×
50	return nil	1✔
51	}
52
53	func loadDeclaredSchemas(fsys afero.Fs) ([]string, error) {	1✔
54	// When pg-delta is enabled, declarative path is the source of truth (config or default).	1✔
55	if utils.IsPgDeltaEnabled() {	1✔
NEW 56	declDir := utils.GetDeclarativeDir()	×
NEW 57	if exists, err := afero.DirExists(fsys, declDir); err == nil && exists {	×
NEW 58	var declared []string	×
NEW 59	if err := afero.Walk(fsys, declDir, func(path string, info fs.FileInfo, err error) error {	×
NEW 60	if err != nil {	×
NEW 61	return err	×
NEW 62	}	×
NEW 63	if info.Mode().IsRegular() && filepath.Ext(info.Name()) == ".sql" {	×
NEW 64	declared = append(declared, path)	×
NEW 65	}	×
NEW 66	return nil	×
NEW 67	}); err != nil {	×
NEW 68	return nil, errors.Errorf("failed to walk declarative dir: %w", err)	×
NEW 69	}	×
NEW 70	sort.Strings(declared)	×
NEW 71	return declared, nil	×
72	}
73	}
74	if schemas := utils.Config.Db.Migrations.SchemaPaths; len(schemas) > 0 {	1✔
UNCOV 75	return schemas.Files(afero.NewIOFS(fsys))	×
76	}	×
77	if exists, err := afero.DirExists(fsys, utils.SchemasDir); err != nil {	1✔
78	return nil, errors.Errorf("failed to check schemas: %w", err)	×
79	} else if !exists {	1✔
80	return nil, nil	×
81	}	×
82	var declared []string	1✔
83	if err := afero.Walk(fsys, utils.SchemasDir, func(path string, info fs.FileInfo, err error) error {	10✔
84	if err != nil {	9✔
85	return err	×
86	}	×
87	if info.Mode().IsRegular() && filepath.Ext(info.Name()) == ".sql" {	13✔
88	declared = append(declared, path)	4✔
89	}	4✔
90	return nil	9✔
91	}); err != nil {	×
92	return nil, errors.Errorf("failed to walk dir: %w", err)	×
93	}	×
94	// Keep file application order deterministic so diff output stays stable across
95	// filesystems and operating systems. This is only if no schema paths in config are set.
96	sort.Strings(declared)	1✔
97	return declared, nil	1✔
98	}
99
100	// https://github.com/djrobstep/migra/blob/master/migra/statements.py#L6
101	var dropStatementPattern = regexp.MustCompile(`(?i)drop\s+`)
102
103	func findDropStatements(out string) []string {	2✔
104	lines, err := parser.SplitAndTrim(strings.NewReader(out))	2✔
105	if err != nil {	2✔
106	return nil	×
107	}	×
108	var drops []string	2✔
109	for _, line := range lines {	6✔
110	if dropStatementPattern.MatchString(line) {	6✔
111	drops = append(drops, line)	2✔
112	}	2✔
113	}
114	return drops	2✔
115	}
116
117	func CreateShadowDatabase(ctx context.Context, port uint16) (string, error) {	14✔
118	// Disable background workers in shadow database	14✔
119	config := start.NewContainerConfig("-c", "max_worker_processes=0")	14✔
120	hostPort := strconv.FormatUint(uint64(port), 10)	14✔
121	hostConfig := container.HostConfig{	14✔
122	PortBindings: nat.PortMap{"5432/tcp": []nat.PortBinding{{HostPort: hostPort}}},	14✔
123	AutoRemove: true,	14✔
124	}	14✔
125	networkingConfig := network.NetworkingConfig{}	14✔
126	if utils.Config.Db.MajorVersion <= 14 {	20✔
127	hostConfig.Tmpfs = map[string]string{"/docker-entrypoint-initdb.d": ""}	6✔
128	}	6✔
129	return utils.DockerStart(ctx, config, hostConfig, networkingConfig, "")	14✔
130	}
131
132	func ConnectShadowDatabase(ctx context.Context, timeout time.Duration, options ...func(pgx.ConnConfig)) (conn pgx.Conn, err error) {	9✔
133	// Retry until connected, cancelled, or timeout	9✔
134	policy := start.NewBackoffPolicy(ctx, timeout)	9✔
135	config := pgconn.Config{Port: utils.Config.Db.ShadowPort}	9✔
136	connect := func() (*pgx.Conn, error) {	18✔
137	return utils.ConnectLocalPostgres(ctx, config, options...)	9✔
138	}	9✔
139	return backoff.RetryWithData(connect, policy)	9✔
140	}
141
142	// Required to bypass pg_cron check: https://github.com/citusdata/pg_cron/blob/main/pg_cron.sql#L3
143	const CREATE_TEMPLATE = "CREATE DATABASE contrib_regression TEMPLATE postgres"
144
145	func MigrateShadowDatabase(ctx context.Context, container string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {	7✔
146	migrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))	7✔
147	if err != nil {	8✔
148	return err	1✔
149	}	1✔
150	conn, err := ConnectShadowDatabase(ctx, 10*time.Second, options...)	6✔
151	if err != nil {	7✔
152	return err	1✔
153	}	1✔
154	defer conn.Close(context.Background())	5✔
155	if err := start.SetupDatabase(ctx, conn, container[:12], os.Stderr, fsys); err != nil {	7✔
156	return err	2✔
157	}	2✔
158	if _, err := conn.Exec(ctx, CREATE_TEMPLATE); err != nil {	3✔
159	return errors.Errorf("failed to create template database: %w", err)	×
160	}	×
161	return migration.ApplyMigrations(ctx, migrations, conn, afero.NewIOFS(fsys))	3✔
162	}
163
164	func DiffDatabase(ctx context.Context, schema []string, config pgconn.Config, w io.Writer, fsys afero.Fs, differ DiffFunc, usePgDelta bool, options ...func(*pgx.ConnConfig)) (string, error) {	8✔
165	fmt.Fprintln(w, "Creating shadow database...")	8✔
166	shadow, err := CreateShadowDatabase(ctx, utils.Config.Db.ShadowPort)	8✔
167	if err != nil {	12✔
168	return "", err	4✔
169	}	4✔
170	defer utils.DockerRemove(shadow)	4✔
171	if err := start.WaitForHealthyService(ctx, utils.Config.Db.HealthTimeout, shadow); err != nil {	5✔
172	return "", err	1✔
173	}	1✔
174	if err := MigrateShadowDatabase(ctx, shadow, fsys, options...); err != nil {	4✔
175	return "", err	1✔
176	}	1✔
177	shadowConfig := pgconn.Config{	2✔
178	Host: utils.Config.Hostname,	2✔
179	Port: utils.Config.Db.ShadowPort,	2✔
180	User: "postgres",	2✔
181	Password: utils.Config.Db.Password,	2✔
182	Database: "postgres",	2✔
183	}	2✔
184	if utils.IsLocalDatabase(config) {	2✔
185	if declared, err := loadDeclaredSchemas(fsys); len(declared) > 0 {	×
186	config = shadowConfig	×
187	config.Database = "contrib_regression"	×
NEW 188	if usePgDelta {	×
NEW 189	declDir := utils.GetDeclarativeDir()	×
NEW 190	if exists, _ := afero.DirExists(fsys, declDir); exists {	×
NEW 191	if err := pgdelta.ApplyDeclarative(ctx, config, fsys); err != nil {	×
NEW 192	return "", err	×
NEW 193	}	×
NEW 194	} else {	×
NEW 195	if err := migrateBaseDatabase(ctx, config, declared, fsys, options...); err != nil {	×
NEW 196	return "", err	×
NEW 197	}	×
198	}
NEW 199	} else {	×
NEW 200	if err := migrateBaseDatabase(ctx, config, declared, fsys, options...); err != nil {	×
NEW 201	return "", err	×
NEW 202	}	×
203	}
204	} else if err != nil {	×
205	return "", err	×
206	}	×
207	}
208	// Load all user defined schemas
209	if len(schema) > 0 {	4✔
210	fmt.Fprintln(w, "Diffing schemas:", strings.Join(schema, ","))	2✔
211	} else {	2✔
212	fmt.Fprintln(w, "Diffing schemas...")	×
213	}	×
214	return differ(ctx, shadowConfig, config, schema, options...)	2✔
215	}
216
217	func migrateBaseDatabase(ctx context.Context, config pgconn.Config, migrations []string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {	×
218	fmt.Fprintln(os.Stderr, "Creating local database from declarative schemas:")	×
219	msg := make([]string, len(migrations))	×
220	for i, m := range migrations {	×
221	msg[i] = fmt.Sprintf(" • %s", utils.Bold(m))	×
222	}	×
223	fmt.Fprintln(os.Stderr, strings.Join(msg, "\n"))	×
224	conn, err := utils.ConnectLocalPostgres(ctx, config, options...)	×
225	if err != nil {	×
226	return err	×
227	}	×
228	defer conn.Close(context.Background())	×
229	return migration.SeedGlobals(ctx, migrations, conn, afero.NewIOFS(fsys))	×
230	}
231
UNCOV 232	func diffWithStream(ctx context.Context, env []string, script string, stdout io.Writer) error {	×
UNCOV 233	cmd := []string{"edge-runtime", "start", "--main-service=."}	×
UNCOV 234	if viper.GetBool("DEBUG") {	×
235	cmd = append(cmd, "--verbose")	×
236	}	×
UNCOV 237	cmdString := strings.Join(cmd, " ")	×
UNCOV 238	entrypoint := []string{"sh", "-c", `cat <<'EOF' > index.ts && ` + cmdString + `	×
UNCOV 239	` + script + `	×
UNCOV 240	EOF	×
UNCOV 241	`}	×
UNCOV 242	var stderr bytes.Buffer	×
UNCOV 243	if err := utils.DockerRunOnceWithConfig(	×
UNCOV 244	ctx,	×
UNCOV 245	container.Config{	×
UNCOV 246	Image: utils.Config.EdgeRuntime.Image,	×
UNCOV 247	Env: env,	×
UNCOV 248	Entrypoint: entrypoint,	×
UNCOV 249	},	×
UNCOV 250	container.HostConfig{	×
UNCOV 251	Binds: []string{utils.EdgeRuntimeId + ":/root/.cache/deno:rw"},	×
UNCOV 252	NetworkMode: network.NetworkHost,	×
UNCOV 253	},	×
UNCOV 254	network.NetworkingConfig{},	×
UNCOV 255	"",	×
UNCOV 256	stdout,	×
UNCOV 257	&stderr,	×
UNCOV 258	// The "main worker has been destroyed" message may not appear at the start of stderr	×
UNCOV 259	// (e.g. preceded by other Deno runtime output), so use Contains instead of HasPrefix.	×
UNCOV 260	); err != nil && !strings.Contains(stderr.String(), "main worker has been destroyed") {	×
UNCOV 261	return errors.Errorf("error diffing schema: %w:\n%s", err, stderr.String())	×
UNCOV 262	}	×
UNCOV 263	return nil	×
264	}

supabase / cli / 23352230634

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