From f3238bed15e51c22dd0c0b16cf5ca0c7ed916f6a Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Sun, 28 Jun 2026 12:52:17 +0700 Subject: [PATCH 1/7] Add source sequence SELECT check to access preflight --- pkg/stream/preflight/access.go | 85 ++++++++++ pkg/stream/preflight/access_test.go | 232 +++++++++++++++++++++++++++- pkg/stream/preflight/builder.go | 7 +- 3 files changed, 316 insertions(+), 8 deletions(-) diff --git a/pkg/stream/preflight/access.go b/pkg/stream/preflight/access.go index a53f8232..f4572201 100644 --- a/pkg/stream/preflight/access.go +++ b/pkg/stream/preflight/access.go @@ -21,6 +21,17 @@ func (c *SourceTableSelectPrivilegesCheck) Name() string { return "source_table_select_privileges" } +// SourceSequenceSelectPrivilegesCheck verifies that the source Postgres role +// can read every in-scope sequence pgstream may need to snapshot. +type SourceSequenceSelectPrivilegesCheck struct { + Source postgres.AcquireFunc + Selection stream.TableSelection +} + +func (c *SourceSequenceSelectPrivilegesCheck) Name() string { + return "source_sequence_select_privileges" +} + const sourceTableSelectPrivilegesQuery = ` SELECT current_user, @@ -35,6 +46,31 @@ WHERE c.relkind IN ('r', 'p') ORDER BY n.nspname, c.relname ` +const sourceSequenceSelectPrivilegesQuery = ` +SELECT + current_user, + tn.nspname AS table_schema, + t.relname AS table_name, + sn.nspname AS sequence_schema, + s.relname AS sequence_name, + has_sequence_privilege(s.oid, 'SELECT') AS has_select +FROM pg_class t +JOIN pg_namespace tn ON tn.oid = t.relnamespace +JOIN pg_attribute a ON a.attrelid = t.oid +JOIN pg_attrdef ad ON ad.adrelid = t.oid AND ad.adnum = a.attnum +JOIN pg_depend d ON d.refobjid = t.oid AND d.refobjsubid = a.attnum +JOIN pg_class s ON s.oid = d.objid +JOIN pg_namespace sn ON sn.oid = s.relnamespace +WHERE t.relkind IN ('r', 'p') + AND s.relkind = 'S' + AND d.deptype = 'a' + AND tn.nspname NOT IN ('pg_catalog', 'information_schema', 'pgstream') + AND tn.nspname NOT LIKE 'pg_toast%' + AND a.attnum > 0 + AND NOT a.attisdropped +ORDER BY tn.nspname, t.relname, sn.nspname, s.relname +` + func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error) { conn, err := c.Source(ctx) if err != nil { @@ -66,6 +102,37 @@ func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, return findings, nil } +func (c *SourceSequenceSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error) { + conn, err := c.Source(ctx) + if err != nil { + return nil, fmt.Errorf("connecting to source: %w", err) + } + + rows, err := conn.Query(ctx, sourceSequenceSelectPrivilegesQuery) + if err != nil { + return nil, fmt.Errorf("querying source sequence privileges: %w", err) + } + defer rows.Close() + + var findings []Finding + for rows.Next() { + var row sourceSequenceSelectPrivilegeRow + if err := rows.Scan(&row.Role, &row.TableSchema, &row.Table, &row.SequenceSchema, &row.Sequence, &row.HasSelect); err != nil { + return nil, fmt.Errorf("scanning row: %w", err) + } + if !c.Selection.IsTableInScope(row.TableSchema, row.Table) { + continue + } + if !row.HasSelect { + findings = append(findings, Finding{Message: sourceSequenceSelectPrivilegeMessage(row)}) + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating rows: %w", err) + } + return findings, nil +} + type sourceTableSelectPrivilegeRow struct { Role string Schema string @@ -73,6 +140,15 @@ type sourceTableSelectPrivilegeRow struct { HasSelect bool } +type sourceSequenceSelectPrivilegeRow struct { + Role string + TableSchema string + Table string + SequenceSchema string + Sequence string + HasSelect bool +} + func sourceTableSelectPrivilegeMessage(row sourceTableSelectPrivilegeRow) string { quotedTable := postgres.QuoteIdentifier(row.Schema) + "." + postgres.QuoteIdentifier(row.Table) quotedRole := postgres.QuoteIdentifier(row.Role) @@ -81,3 +157,12 @@ func sourceTableSelectPrivilegeMessage(row sourceTableSelectPrivilegeRow) string row.Role, row.Schema, row.Table, quotedTable, quotedRole, ) } + +func sourceSequenceSelectPrivilegeMessage(row sourceSequenceSelectPrivilegeRow) string { + quotedSequence := postgres.QuoteIdentifier(row.SequenceSchema) + "." + postgres.QuoteIdentifier(row.Sequence) + quotedRole := postgres.QuoteIdentifier(row.Role) + return fmt.Sprintf( + "source role %q lacks SELECT on sequence %s.%s; run GRANT SELECT ON SEQUENCE %s TO %s", + row.Role, row.SequenceSchema, row.Sequence, quotedSequence, quotedRole, + ) +} diff --git a/pkg/stream/preflight/access_test.go b/pkg/stream/preflight/access_test.go index 423fae4e..25cf3e49 100644 --- a/pkg/stream/preflight/access_test.go +++ b/pkg/stream/preflight/access_test.go @@ -179,6 +179,167 @@ func TestSourceTableSelectPrivilegesCheck_Name(t *testing.T) { require.Equal(t, "source_table_select_privileges", (&SourceTableSelectPrivilegesCheck{}).Name()) } +func TestSourceSequenceSelectPrivilegesCheck_Run_AllSequencesHaveSelect(t *testing.T) { + t.Parallel() + + check := &SourceSequenceSelectPrivilegesCheck{ + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "public", Table: "orders", SequenceSchema: "public", Sequence: "orders_id_seq", HasSelect: true}, + {Role: "pgstream_user", TableSchema: "public", Table: "users", SequenceSchema: "public", Sequence: "users_id_seq", HasSelect: true}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Empty(t, findings) +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_MissingSelectReturnsFinding(t *testing.T) { + t.Parallel() + + check := &SourceSequenceSelectPrivilegesCheck{ + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "public", Table: "orders", SequenceSchema: "public", Sequence: "orders_id_seq", HasSelect: false}, + {Role: "pgstream_user", TableSchema: "public", Table: "users", SequenceSchema: "public", Sequence: "users_id_seq", HasSelect: true}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, `source role "pgstream_user"`) + require.Contains(t, findings[0].Message, "public.orders_id_seq") + require.Contains(t, findings[0].Message, "GRANT SELECT ON SEQUENCE") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_MultipleMissingSelect(t *testing.T) { + t.Parallel() + + check := &SourceSequenceSelectPrivilegesCheck{ + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "billing", Table: "invoices", SequenceSchema: "billing", Sequence: "invoices_id_seq", HasSelect: false}, + {Role: "pgstream_user", TableSchema: "public", Table: "orders", SequenceSchema: "public", Sequence: "orders_id_seq", HasSelect: false}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 2) + require.Contains(t, findings[0].Message, "billing.invoices_id_seq") + require.Contains(t, findings[1].Message, "public.orders_id_seq") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_SourceAcquireFails(t *testing.T) { + t.Parallel() + + checkErr := errors.New("boom") + check := &SourceSequenceSelectPrivilegesCheck{ + Source: func(context.Context) (postgres.Querier, error) { + return nil, checkErr + }, + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, checkErr) + require.ErrorContains(t, err, "connecting to source") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_QueryFails(t *testing.T) { + t.Parallel() + + queryErr := errors.New("query failed") + check := &SourceSequenceSelectPrivilegesCheck{ + Source: func(context.Context) (postgres.Querier, error) { + return &mocks.Querier{ + QueryFn: func(context.Context, uint, string, ...any) (postgres.Rows, error) { + return nil, queryErr + }, + }, nil + }, + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, queryErr) + require.ErrorContains(t, err, "querying source sequence privileges") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_ScanFails(t *testing.T) { + t.Parallel() + + scanErr := errors.New("scan failed") + check := &SourceSequenceSelectPrivilegesCheck{ + Source: sourceWithMockRows(&mocks.Rows{ + NextFn: func(i uint) bool { return i == 1 }, + ScanFn: func(uint, ...any) error { + return scanErr + }, + ErrFn: func() error { return nil }, + }), + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, scanErr) + require.ErrorContains(t, err, "scanning row") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_RowsErr(t *testing.T) { + t.Parallel() + + rowsErr := errors.New("rows failed") + check := &SourceSequenceSelectPrivilegesCheck{ + Source: sourceWithMockRows(&mocks.Rows{ + NextFn: func(uint) bool { return false }, + ScanFn: func(uint, ...any) error { + t.Fatal("Scan should not be called") + return nil + }, + ErrFn: func() error { return rowsErr }, + }), + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, rowsErr) + require.ErrorContains(t, err, "iterating rows") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_FiltersTablesByScope(t *testing.T) { + t.Parallel() + + sel, err := stream.NewTableSelection([]string{"public.*"}, []string{"public.audit_log"}) + require.NoError(t, err) + + check := &SourceSequenceSelectPrivilegesCheck{ + Selection: sel, + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "billing", Table: "invoices", SequenceSchema: "billing", Sequence: "invoices_id_seq", HasSelect: false}, + {Role: "pgstream_user", TableSchema: "public", Table: "audit_log", SequenceSchema: "public", Sequence: "audit_log_id_seq", HasSelect: false}, + {Role: "pgstream_user", TableSchema: "public", Table: "orders", SequenceSchema: "public", Sequence: "orders_id_seq", HasSelect: false}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, "public.orders_id_seq") +} + +func TestSourceSequenceSelectPrivilegesCheck_Name(t *testing.T) { + t.Parallel() + + require.Equal(t, "source_sequence_select_privileges", (&SourceSequenceSelectPrivilegesCheck{}).Name()) +} func TestSourceTableSelectPrivilegeMessage(t *testing.T) { t.Parallel() @@ -209,6 +370,20 @@ func TestSourceTableSelectPrivilegeMessage_QuotesOnlyRemediation(t *testing.T) { require.Contains(t, msg, `GRANT SELECT ON TABLE "Reporting"."DailyRollup" TO "Replicator"`) } +func TestSourceSequenceSelectPrivilegeMessage(t *testing.T) { + t.Parallel() + + msg := sourceSequenceSelectPrivilegeMessage(sourceSequenceSelectPrivilegeRow{ + Role: "pgstream_user", + SequenceSchema: "public", + Sequence: "orders_id_seq", + }) + + require.Contains(t, msg, `source role "pgstream_user"`) + require.Contains(t, msg, "lacks SELECT on sequence public.orders_id_seq;") + require.Contains(t, msg, `GRANT SELECT ON SEQUENCE "public"."orders_id_seq" TO "pgstream_user"`) +} + func TestBuildAccessChecks(t *testing.T) { t.Parallel() @@ -224,13 +399,13 @@ func TestBuildAccessChecks(t *testing.T) { cfg: &stream.Config{}, }, { - name: "source postgres url returns select privilege check with unfiltered selection", + name: "source postgres url returns access checks", cfg: &stream.Config{ Listener: stream.ListenerConfig{ Postgres: &stream.PostgresListenerConfig{URL: "postgres://source"}, }, }, - wantChecks: 1, + wantChecks: 2, }, { name: "snapshot+filter Include unions through AccessTableSelection", @@ -251,7 +426,7 @@ func TestBuildAccessChecks(t *testing.T) { }, }, }, - wantChecks: 1, + wantChecks: 2, wantInclude: []string{"public.orders", "public.users"}, }, } @@ -268,10 +443,14 @@ func TestBuildAccessChecks(t *testing.T) { return } require.NotNil(t, cleanup) - check, ok := checks[0].(*SourceTableSelectPrivilegesCheck) + tableCheck, ok := checks[0].(*SourceTableSelectPrivilegesCheck) + require.True(t, ok) + require.ElementsMatch(t, tc.wantInclude, tableCheck.Selection.Include(), "Include lists differ") + require.ElementsMatch(t, tc.wantExclude, tableCheck.Selection.Exclude(), "Exclude lists differ") + sequenceCheck, ok := checks[1].(*SourceSequenceSelectPrivilegesCheck) require.True(t, ok) - require.ElementsMatch(t, tc.wantInclude, check.Selection.Include(), "Include lists differ") - require.ElementsMatch(t, tc.wantExclude, check.Selection.Exclude(), "Exclude lists differ") + require.ElementsMatch(t, tc.wantInclude, sequenceCheck.Selection.Include(), "Include lists differ") + require.ElementsMatch(t, tc.wantExclude, sequenceCheck.Selection.Exclude(), "Exclude lists differ") }) } } @@ -290,8 +469,9 @@ func TestBuildChecks_SelectedAccessOnly(t *testing.T) { checks, cleanup := BuildChecks(cfg, []Category{CategoryAccess}) require.NotNil(t, cleanup) - require.Len(t, checks, 1) + require.Len(t, checks, 2) require.Equal(t, "source_table_select_privileges", checks[0].Name()) + require.Equal(t, "source_sequence_select_privileges", checks[1].Name()) } func sourceWithRows(t *testing.T, rows []sourceTableSelectPrivilegeRow) postgres.AcquireFunc { @@ -335,3 +515,41 @@ func privilegeRows(t *testing.T, rows []sourceTableSelectPrivilegeRow) postgres. ErrFn: func() error { return nil }, } } + +func sourceWithSequenceRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRow) postgres.AcquireFunc { + t.Helper() + return sourceWithMockRows(sequencePrivilegeRows(t, rows)) +} + +func sequencePrivilegeRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRow) postgres.Rows { + t.Helper() + return &mocks.Rows{ + NextFn: func(i uint) bool { + return int(i) <= len(rows) + }, + ScanFn: func(i uint, dest ...any) error { + require.Len(t, dest, 6) + row := rows[i-1] + role, ok := dest[0].(*string) + require.True(t, ok) + tableSchema, ok := dest[1].(*string) + require.True(t, ok) + table, ok := dest[2].(*string) + require.True(t, ok) + sequenceSchema, ok := dest[3].(*string) + require.True(t, ok) + sequence, ok := dest[4].(*string) + require.True(t, ok) + hasSelect, ok := dest[5].(*bool) + require.True(t, ok) + *role = row.Role + *tableSchema = row.TableSchema + *table = row.Table + *sequenceSchema = row.SequenceSchema + *sequence = row.Sequence + *hasSelect = row.HasSelect + return nil + }, + ErrFn: func() error { return nil }, + } +} diff --git a/pkg/stream/preflight/builder.go b/pkg/stream/preflight/builder.go index b5d22d27..f643f2b1 100644 --- a/pkg/stream/preflight/builder.go +++ b/pkg/stream/preflight/builder.go @@ -78,10 +78,15 @@ func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc) { return nil, nil } src := postgres.NewLazyConn(url) + selection := cfg.AccessTableSelection() return []Check{ &SourceTableSelectPrivilegesCheck{ Source: src.Acquire, - Selection: cfg.AccessTableSelection(), + Selection: selection, + }, + &SourceSequenceSelectPrivilegesCheck{ + Source: src.Acquire, + Selection: selection, }, }, src.Close } From 4c528117821eee7d0a603af35dc6cb7ccd3048d7 Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Tue, 30 Jun 2026 00:20:28 +0700 Subject: [PATCH 2/7] format access preflight tests --- pkg/stream/preflight/access_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/stream/preflight/access_test.go b/pkg/stream/preflight/access_test.go index 25cf3e49..fba05be6 100644 --- a/pkg/stream/preflight/access_test.go +++ b/pkg/stream/preflight/access_test.go @@ -340,6 +340,7 @@ func TestSourceSequenceSelectPrivilegesCheck_Name(t *testing.T) { require.Equal(t, "source_sequence_select_privileges", (&SourceSequenceSelectPrivilegesCheck{}).Name()) } + func TestSourceTableSelectPrivilegeMessage(t *testing.T) { t.Parallel() From f87dd2f981751f97015bbe941337dc7382dba13b Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Tue, 30 Jun 2026 23:27:40 +0700 Subject: [PATCH 3/7] Check identity-backed sequences in access preflight --- pkg/stream/preflight/access.go | 3 +- .../preflight/access_integration_test.go | 76 +++++++++++++++++++ 2 files changed, 77 insertions(+), 2 deletions(-) create mode 100644 pkg/stream/preflight/access_integration_test.go diff --git a/pkg/stream/preflight/access.go b/pkg/stream/preflight/access.go index f4572201..14066744 100644 --- a/pkg/stream/preflight/access.go +++ b/pkg/stream/preflight/access.go @@ -57,13 +57,12 @@ SELECT FROM pg_class t JOIN pg_namespace tn ON tn.oid = t.relnamespace JOIN pg_attribute a ON a.attrelid = t.oid -JOIN pg_attrdef ad ON ad.adrelid = t.oid AND ad.adnum = a.attnum JOIN pg_depend d ON d.refobjid = t.oid AND d.refobjsubid = a.attnum JOIN pg_class s ON s.oid = d.objid JOIN pg_namespace sn ON sn.oid = s.relnamespace WHERE t.relkind IN ('r', 'p') AND s.relkind = 'S' - AND d.deptype = 'a' + AND d.deptype IN ('a', 'i') AND tn.nspname NOT IN ('pg_catalog', 'information_schema', 'pgstream') AND tn.nspname NOT LIKE 'pg_toast%' AND a.attnum > 0 diff --git a/pkg/stream/preflight/access_integration_test.go b/pkg/stream/preflight/access_integration_test.go new file mode 100644 index 00000000..4014865f --- /dev/null +++ b/pkg/stream/preflight/access_integration_test.go @@ -0,0 +1,76 @@ +// SPDX-License-Identifier: Apache-2.0 + +package preflight + +import ( + "context" + "net/url" + "os" + "testing" + + "github.com/stretchr/testify/require" + + pglib "github.com/xataio/pgstream/internal/postgres" + "github.com/xataio/pgstream/internal/testcontainers" +) + +func TestSourceSequenceSelectPrivilegesCheck_Run_IdentityColumnsAreChecked(t *testing.T) { + if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" { + t.Skip("skipping integration test...") + } + + ctx := context.Background() + + var pgURL string + cleanup, err := testcontainers.SetupPostgresContainer(ctx, &pgURL, testcontainers.Postgres14) + require.NoError(t, err) + defer cleanup() + + adminConn, err := pglib.NewConn(ctx, pgURL) + require.NoError(t, err) + defer adminConn.Close(ctx) + + _, err = adminConn.Exec(ctx, ` + CREATE TABLE public.identity_orders ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + payload text + ) + `) + require.NoError(t, err) + + _, err = adminConn.Exec(ctx, `CREATE ROLE identity_reader LOGIN PASSWORD 'secret'`) + require.NoError(t, err) + _, err = adminConn.Exec(ctx, `GRANT USAGE ON SCHEMA public TO identity_reader`) + require.NoError(t, err) + _, err = adminConn.Exec(ctx, `GRANT SELECT ON TABLE public.identity_orders TO identity_reader`) + require.NoError(t, err) + + var sequenceName string + err = adminConn.QueryRow(ctx, []any{&sequenceName}, `SELECT pg_get_serial_sequence('public.identity_orders', 'id')`) + require.NoError(t, err) + require.Equal(t, "public.identity_orders_id_seq", sequenceName) + + readerURL := mustRewritePGUser(t, pgURL, "identity_reader", "secret") + check := &SourceSequenceSelectPrivilegesCheck{ + Source: func(ctx context.Context) (pglib.Querier, error) { + return pglib.NewConn(ctx, readerURL) + }, + } + + findings, err := check.Run(ctx) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, "public.identity_orders_id_seq") + require.Contains(t, findings[0].Message, `source role "identity_reader"`) + require.Contains(t, findings[0].Message, `GRANT SELECT ON SEQUENCE "public"."identity_orders_id_seq" TO "identity_reader"`) +} + +func mustRewritePGUser(t *testing.T, rawURL, user, password string) string { + t.Helper() + + parsed, err := url.Parse(rawURL) + require.NoError(t, err) + parsed.User = url.UserPassword(user, password) + return parsed.String() +} From aec7621ad7ea0a6270bdd6207c4cd8c6211094c1 Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Wed, 1 Jul 2026 10:15:49 +0700 Subject: [PATCH 4/7] Check standalone sequences in access preflight --- pkg/stream/preflight/access.go | 83 ++++++++++++++++++- .../preflight/access_integration_test.go | 13 ++- pkg/stream/preflight/access_test.go | 66 ++++++++++++++- 3 files changed, 153 insertions(+), 9 deletions(-) diff --git a/pkg/stream/preflight/access.go b/pkg/stream/preflight/access.go index 14066744..ec685d50 100644 --- a/pkg/stream/preflight/access.go +++ b/pkg/stream/preflight/access.go @@ -53,7 +53,8 @@ SELECT t.relname AS table_name, sn.nspname AS sequence_schema, s.relname AS sequence_name, - has_sequence_privilege(s.oid, 'SELECT') AS has_select + has_sequence_privilege(s.oid, 'SELECT') AS has_select, + false AS is_standalone FROM pg_class t JOIN pg_namespace tn ON tn.oid = t.relnamespace JOIN pg_attribute a ON a.attrelid = t.oid @@ -67,7 +68,30 @@ WHERE t.relkind IN ('r', 'p') AND tn.nspname NOT LIKE 'pg_toast%' AND a.attnum > 0 AND NOT a.attisdropped -ORDER BY tn.nspname, t.relname, sn.nspname, s.relname +UNION ALL +SELECT + current_user, + n.nspname AS table_schema, + '' AS table_name, + n.nspname AS sequence_schema, + c.relname AS sequence_name, + has_sequence_privilege(c.oid, 'SELECT') AS has_select, + true AS is_standalone +FROM pg_class c +JOIN pg_namespace n ON n.oid = c.relnamespace +WHERE c.relkind = 'S' + AND n.nspname NOT IN ('pg_catalog', 'information_schema', 'pgstream') + AND n.nspname NOT LIKE 'pg_toast%' + AND NOT EXISTS ( + SELECT 1 + FROM pg_depend d + JOIN pg_class t ON t.oid = d.refobjid + WHERE d.objid = c.oid + AND d.refobjsubid > 0 + AND d.deptype IN ('a', 'i') + AND t.relkind IN ('r', 'p') + ) +ORDER BY table_schema, table_name, sequence_schema, sequence_name ` func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error) { @@ -102,6 +126,11 @@ func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, } func (c *SourceSequenceSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error) { + include, exclude, err := selectionMaps(c.Selection) + if err != nil { + return nil, fmt.Errorf("parsing table selection: %w", err) + } + conn, err := c.Source(ctx) if err != nil { return nil, fmt.Errorf("connecting to source: %w", err) @@ -116,10 +145,10 @@ func (c *SourceSequenceSelectPrivilegesCheck) Run(ctx context.Context) ([]Findin var findings []Finding for rows.Next() { var row sourceSequenceSelectPrivilegeRow - if err := rows.Scan(&row.Role, &row.TableSchema, &row.Table, &row.SequenceSchema, &row.Sequence, &row.HasSelect); err != nil { + if err := rows.Scan(&row.Role, &row.TableSchema, &row.Table, &row.SequenceSchema, &row.Sequence, &row.HasSelect, &row.IsStandalone); err != nil { return nil, fmt.Errorf("scanning row: %w", err) } - if !c.Selection.IsTableInScope(row.TableSchema, row.Table) { + if !sequenceInScope(row, c.Selection, include, exclude) { continue } if !row.HasSelect { @@ -146,6 +175,52 @@ type sourceSequenceSelectPrivilegeRow struct { SequenceSchema string Sequence string HasSelect bool + IsStandalone bool +} + +func selectionMaps(selection stream.TableSelection) (include, exclude postgres.SchemaTableMap, err error) { + if len(selection.Include()) > 0 { + include, err = postgres.NewSchemaTableMap(selection.Include()) + if err != nil { + return nil, nil, fmt.Errorf("include: %w", err) + } + } + if len(selection.Exclude()) > 0 { + exclude, err = postgres.NewSchemaTableMap(selection.Exclude()) + if err != nil { + return nil, nil, fmt.Errorf("exclude: %w", err) + } + } + return include, exclude, nil +} + +func sequenceInScope(row sourceSequenceSelectPrivilegeRow, selection stream.TableSelection, include, exclude postgres.SchemaTableMap) bool { + if !row.IsStandalone { + return selection.IsTableInScope(row.TableSchema, row.Table) + } + return standaloneSequenceSchemaInScope(row.SequenceSchema, selection, include, exclude) +} + +func standaloneSequenceSchemaInScope(schema string, selection stream.TableSelection, include, exclude postgres.SchemaTableMap) bool { + if selection.IsUnfiltered() { + return true + } + if include != nil { + return schemaHasWildcard(include, schema) + } + if exclude != nil { + return !schemaHasWildcard(exclude, schema) + } + return true +} + +func schemaHasWildcard(tables postgres.SchemaTableMap, schema string) bool { + schemaTables := tables.GetSchemaTables(schema) + if len(schemaTables) == 0 { + return false + } + _, found := schemaTables["*"] + return found } func sourceTableSelectPrivilegeMessage(row sourceTableSelectPrivilegeRow) string { diff --git a/pkg/stream/preflight/access_integration_test.go b/pkg/stream/preflight/access_integration_test.go index 4014865f..bbf40b66 100644 --- a/pkg/stream/preflight/access_integration_test.go +++ b/pkg/stream/preflight/access_integration_test.go @@ -44,6 +44,8 @@ func TestSourceSequenceSelectPrivilegesCheck_Run_IdentityColumnsAreChecked(t *te require.NoError(t, err) _, err = adminConn.Exec(ctx, `GRANT SELECT ON TABLE public.identity_orders TO identity_reader`) require.NoError(t, err) + _, err = adminConn.Exec(ctx, `CREATE SEQUENCE public.free_standing_seq`) + require.NoError(t, err) var sequenceName string err = adminConn.QueryRow(ctx, []any{&sequenceName}, `SELECT pg_get_serial_sequence('public.identity_orders', 'id')`) @@ -60,10 +62,13 @@ func TestSourceSequenceSelectPrivilegesCheck_Run_IdentityColumnsAreChecked(t *te findings, err := check.Run(ctx) require.NoError(t, err) - require.Len(t, findings, 1) - require.Contains(t, findings[0].Message, "public.identity_orders_id_seq") - require.Contains(t, findings[0].Message, `source role "identity_reader"`) - require.Contains(t, findings[0].Message, `GRANT SELECT ON SEQUENCE "public"."identity_orders_id_seq" TO "identity_reader"`) + require.Len(t, findings, 2) + allMessages := findings[0].Message + findings[1].Message + require.Contains(t, allMessages, "public.identity_orders_id_seq") + require.Contains(t, allMessages, "public.free_standing_seq") + require.Contains(t, allMessages, `source role "identity_reader"`) + require.Contains(t, allMessages, `GRANT SELECT ON SEQUENCE "public"."identity_orders_id_seq" TO "identity_reader"`) + require.Contains(t, allMessages, `GRANT SELECT ON SEQUENCE "public"."free_standing_seq" TO "identity_reader"`) } func mustRewritePGUser(t *testing.T, rawURL, user, password string) string { diff --git a/pkg/stream/preflight/access_test.go b/pkg/stream/preflight/access_test.go index fba05be6..78b4b75f 100644 --- a/pkg/stream/preflight/access_test.go +++ b/pkg/stream/preflight/access_test.go @@ -335,6 +335,67 @@ func TestSourceSequenceSelectPrivilegesCheck_Run_FiltersTablesByScope(t *testing require.Contains(t, findings[0].Message, "public.orders_id_seq") } +func TestSourceSequenceSelectPrivilegesCheck_Run_IncludesStandaloneSequencesForWildcardSchemas(t *testing.T) { + t.Parallel() + + sel, err := stream.NewTableSelection([]string{"public.*"}, nil) + require.NoError(t, err) + + check := &SourceSequenceSelectPrivilegesCheck{ + Selection: sel, + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "public", Table: "", SequenceSchema: "public", Sequence: "free_standing_seq", HasSelect: false, IsStandalone: true}, + {Role: "pgstream_user", TableSchema: "billing", Table: "", SequenceSchema: "billing", Sequence: "billing_seq", HasSelect: false, IsStandalone: true}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, "public.free_standing_seq") +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_ExcludesStandaloneSequencesForTableOnlySelection(t *testing.T) { + t.Parallel() + + sel, err := stream.NewTableSelection([]string{"public.orders"}, nil) + require.NoError(t, err) + + check := &SourceSequenceSelectPrivilegesCheck{ + Selection: sel, + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "public", Table: "", SequenceSchema: "public", Sequence: "free_standing_seq", HasSelect: false, IsStandalone: true}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Empty(t, findings) +} + +func TestSourceSequenceSelectPrivilegesCheck_Run_ExcludesStandaloneSequencesForExcludedSchema(t *testing.T) { + t.Parallel() + + sel, err := stream.NewTableSelection(nil, []string{"public.*"}) + require.NoError(t, err) + + check := &SourceSequenceSelectPrivilegesCheck{ + Selection: sel, + Source: sourceWithSequenceRows(t, []sourceSequenceSelectPrivilegeRow{ + {Role: "pgstream_user", TableSchema: "public", Table: "", SequenceSchema: "public", Sequence: "free_standing_seq", HasSelect: false, IsStandalone: true}, + {Role: "pgstream_user", TableSchema: "billing", Table: "", SequenceSchema: "billing", Sequence: "billing_seq", HasSelect: false, IsStandalone: true}, + }), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, "billing.billing_seq") +} + func TestSourceSequenceSelectPrivilegesCheck_Name(t *testing.T) { t.Parallel() @@ -529,7 +590,7 @@ func sequencePrivilegeRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRow return int(i) <= len(rows) }, ScanFn: func(i uint, dest ...any) error { - require.Len(t, dest, 6) + require.Len(t, dest, 7) row := rows[i-1] role, ok := dest[0].(*string) require.True(t, ok) @@ -543,12 +604,15 @@ func sequencePrivilegeRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRow require.True(t, ok) hasSelect, ok := dest[5].(*bool) require.True(t, ok) + isStandalone, ok := dest[6].(*bool) + require.True(t, ok) *role = row.Role *tableSchema = row.TableSchema *table = row.Table *sequenceSchema = row.SequenceSchema *sequence = row.Sequence *hasSelect = row.HasSelect + *isStandalone = row.IsStandalone return nil }, ErrFn: func() error { return nil }, From 86eee09e83b1bcef13092fa052316b7fc5fce8ef Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Sat, 11 Jul 2026 14:29:04 +0700 Subject: [PATCH 5/7] Add target CREATEDB preflight check --- pkg/stream/config.go | 20 +++++ pkg/stream/preflight/access.go | 46 ++++++++++ pkg/stream/preflight/access_test.go | 125 ++++++++++++++++++++++++++++ pkg/stream/preflight/builder.go | 30 +++++-- 4 files changed, 216 insertions(+), 5 deletions(-) diff --git a/pkg/stream/config.go b/pkg/stream/config.go index 2f4d3bae..f320ee6c 100644 --- a/pkg/stream/config.go +++ b/pkg/stream/config.go @@ -186,6 +186,26 @@ func (c *Config) isInjectorEnabled() bool { return c.Processor.Injector != nil && c.Processor.Injector.URL != "" } +func (c *Config) SnapshotTargetPostgresURL() string { + if c.Listener.Postgres == nil || + c.Listener.Postgres.Snapshot == nil || + c.Listener.Postgres.Snapshot.Schema == nil || + c.Listener.Postgres.Snapshot.Schema.DumpRestore == nil { + return "" + } + return c.Listener.Postgres.Snapshot.Schema.DumpRestore.TargetPGURL +} + +func (c *Config) SnapshotCreateTargetDB() bool { + if c.Listener.Postgres == nil || + c.Listener.Postgres.Snapshot == nil || + c.Listener.Postgres.Snapshot.Schema == nil || + c.Listener.Postgres.Snapshot.Schema.DumpRestore == nil { + return false + } + return c.Listener.Postgres.Snapshot.Schema.DumpRestore.CreateTargetDB +} + // restoreConflictTargetsBeforeData reports whether the schema snapshot must // restore primary keys, unique constraints and unique indexes before the data // snapshot runs. This is required when the postgres batch writer emits diff --git a/pkg/stream/preflight/access.go b/pkg/stream/preflight/access.go index ec685d50..6af73229 100644 --- a/pkg/stream/preflight/access.go +++ b/pkg/stream/preflight/access.go @@ -32,6 +32,14 @@ func (c *SourceSequenceSelectPrivilegesCheck) Name() string { return "source_sequence_select_privileges" } +type TargetCreateDBPrivilegeCheck struct { + Target postgres.AcquireFunc +} + +func (c *TargetCreateDBPrivilegeCheck) Name() string { + return "target_createdb_privilege" +} + const sourceTableSelectPrivilegesQuery = ` SELECT current_user, @@ -94,6 +102,14 @@ WHERE c.relkind = 'S' ORDER BY table_schema, table_name, sequence_schema, sequence_name ` +const targetCreateDBPrivilegeQuery = ` +SELECT + current_user, + rolcreatedb +FROM pg_roles +WHERE rolname = current_user +` + func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error) { conn, err := c.Source(ctx) if err != nil { @@ -161,6 +177,27 @@ func (c *SourceSequenceSelectPrivilegesCheck) Run(ctx context.Context) ([]Findin return findings, nil } +func (c *TargetCreateDBPrivilegeCheck) Run(ctx context.Context) ([]Finding, error) { + conn, err := c.Target(ctx) + if err != nil { + return nil, fmt.Errorf("connecting to target: %w", err) + } + var role string + var hasCreateDB bool + + if err := conn.QueryRow(ctx, []any{&role, &hasCreateDB}, targetCreateDBPrivilegeQuery); err != nil { + return nil, fmt.Errorf("querying target CREATEDB privilege: %w", err) + } + + if hasCreateDB { + return nil, nil + } + return []Finding{ + {Message: targetCreateDBPrivilegeMessage(role)}, + }, nil + +} + type sourceTableSelectPrivilegeRow struct { Role string Schema string @@ -240,3 +277,12 @@ func sourceSequenceSelectPrivilegeMessage(row sourceSequenceSelectPrivilegeRow) row.Role, row.SequenceSchema, row.Sequence, quotedSequence, quotedRole, ) } + +func targetCreateDBPrivilegeMessage(role string) string { + quotedRole := postgres.QuoteIdentifier(role) + return fmt.Sprintf( + "target role %q lacks CREATEDB; run ALTER ROLE %s CREATEDB", + role, quotedRole, + ) + +} diff --git a/pkg/stream/preflight/access_test.go b/pkg/stream/preflight/access_test.go index 78b4b75f..4c4394bf 100644 --- a/pkg/stream/preflight/access_test.go +++ b/pkg/stream/preflight/access_test.go @@ -11,6 +11,7 @@ import ( "github.com/xataio/pgstream/internal/postgres" "github.com/xataio/pgstream/internal/postgres/mocks" + pgdumprestore "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/schema/pgdumprestore" "github.com/xataio/pgstream/pkg/stream" "github.com/xataio/pgstream/pkg/wal/listener/snapshot/adapter" snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder" @@ -402,6 +403,79 @@ func TestSourceSequenceSelectPrivilegesCheck_Name(t *testing.T) { require.Equal(t, "source_sequence_select_privileges", (&SourceSequenceSelectPrivilegesCheck{}).Name()) } +func TestTargetCreateDBPrivilegeCheck_Run_HasCreateDB(t *testing.T) { + t.Parallel() + + check := &TargetCreateDBPrivilegeCheck{ + Target: targetWithCreateDBPrivilege(t, "pgstreamtarget", true), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Empty(t, findings) +} + +func TestTargetCreateDBPrivilegeCheck_Run_MissingCreateDBReturnsFinding(t *testing.T) { + t.Parallel() + + check := &TargetCreateDBPrivilegeCheck{ + Target: targetWithCreateDBPrivilege(t, "pgstreamtarget", false), + } + + findings, err := check.Run(context.Background()) + + require.NoError(t, err) + require.Len(t, findings, 1) + require.Contains(t, findings[0].Message, `target role "pgstreamtarget"`) + require.Contains(t, findings[0].Message, "lacks CREATEDB") + require.Contains(t, findings[0].Message, `ALTER ROLE "pgstreamtarget" CREATEDB`) +} + +func TestTargetCreateDBPrivilegeCheck_Run_TargetAcquireFails(t *testing.T) { + t.Parallel() + + checkErr := errors.New("boom") + check := &TargetCreateDBPrivilegeCheck{ + Target: func(context.Context) (postgres.Querier, error) { + return nil, checkErr + }, + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, checkErr) + require.ErrorContains(t, err, "connecting to target") +} + +func TestTargetCreateDBPrivilegeCheck_Run_QueryFails(t *testing.T) { + t.Parallel() + + queryErr := errors.New("query failed") + check := &TargetCreateDBPrivilegeCheck{ + Target: func(context.Context) (postgres.Querier, error) { + return &mocks.Querier{ + QueryRowFn: func(context.Context, []any, string, ...any) error { + return queryErr + }, + }, nil + }, + } + + findings, err := check.Run(context.Background()) + + require.Nil(t, findings) + require.ErrorIs(t, err, queryErr) + require.ErrorContains(t, err, "querying target CREATEDB privilege") +} + +func TestTargetCreateDBPrivilegeCheck_Name(t *testing.T) { + t.Parallel() + + require.Equal(t, "target_createdb_privilege", (&TargetCreateDBPrivilegeCheck{}).Name()) +} + func TestSourceTableSelectPrivilegeMessage(t *testing.T) { t.Parallel() @@ -446,6 +520,16 @@ func TestSourceSequenceSelectPrivilegeMessage(t *testing.T) { require.Contains(t, msg, `GRANT SELECT ON SEQUENCE "public"."orders_id_seq" TO "pgstream_user"`) } +func TestTargetCreateDBPrivilegeMessage(t *testing.T) { + t.Parallel() + + msg := targetCreateDBPrivilegeMessage("pgstreamtarget") + + require.Contains(t, msg, `target role "pgstreamtarget"`) + require.Contains(t, msg, "lacks CREATEDB") + require.Contains(t, msg, `ALTER ROLE "pgstreamtarget" CREATEDB`) +} + func TestBuildAccessChecks(t *testing.T) { t.Parallel() @@ -469,6 +553,25 @@ func TestBuildAccessChecks(t *testing.T) { }, wantChecks: 2, }, + { + name: "create_target_db adds target createdb check", + cfg: &stream.Config{ + Listener: stream.ListenerConfig{ + Postgres: &stream.PostgresListenerConfig{ + URL: "postgres://source", + Snapshot: &snapshotbuilder.SnapshotListenerConfig{ + Schema: &snapshotbuilder.SchemaSnapshotConfig{ + DumpRestore: &pgdumprestore.Config{ + TargetPGURL: "postgres://target", + CreateTargetDB: true, + }, + }, + }, + }, + }, + }, + wantChecks: 3, + }, { name: "snapshot+filter Include unions through AccessTableSelection", cfg: &stream.Config{ @@ -513,6 +616,10 @@ func TestBuildAccessChecks(t *testing.T) { require.True(t, ok) require.ElementsMatch(t, tc.wantInclude, sequenceCheck.Selection.Include(), "Include lists differ") require.ElementsMatch(t, tc.wantExclude, sequenceCheck.Selection.Exclude(), "Exclude lists differ") + if tc.wantChecks > 2 { + _, ok = checks[2].(*TargetCreateDBPrivilegeCheck) + require.True(t, ok) + } }) } } @@ -583,6 +690,24 @@ func sourceWithSequenceRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRo return sourceWithMockRows(sequencePrivilegeRows(t, rows)) } +func targetWithCreateDBPrivilege(t *testing.T, role string, hasCreateDB bool) postgres.AcquireFunc { + t.Helper() + return func(context.Context) (postgres.Querier, error) { + return &mocks.Querier{ + QueryRowFn: func(_ context.Context, dest []any, _ string, _ ...any) error { + require.Len(t, dest, 2) + roleDest, ok := dest[0].(*string) + require.True(t, ok) + createDBDest, ok := dest[1].(*bool) + require.True(t, ok) + *roleDest = role + *createDBDest = hasCreateDB + return nil + }, + }, nil + } +} + func sequencePrivilegeRows(t *testing.T, rows []sourceSequenceSelectPrivilegeRow) postgres.Rows { t.Helper() return &mocks.Rows{ diff --git a/pkg/stream/preflight/builder.go b/pkg/stream/preflight/builder.go index 4a14f0c2..8fa3a9c5 100644 --- a/pkg/stream/preflight/builder.go +++ b/pkg/stream/preflight/builder.go @@ -74,13 +74,13 @@ func BuildReplicationChecks(cfg *stream.Config) ([]Check, CleanupFunc) { // BuildAccessChecks returns the access-preflight checks applicable to cfg, // plus a cleanup function that closes the shared source connection. func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc) { - url := cfg.SourcePostgresURL() - if url == "" { + sourceURL := cfg.SourcePostgresURL() + if sourceURL == "" { return nil, nil } - src := postgres.NewLazyConn(url) + src := postgres.NewLazyConn(sourceURL) selection := cfg.AccessTableSelection() - return []Check{ + checks := []Check{ &SourceTableSelectPrivilegesCheck{ Source: src.Acquire, Selection: selection, @@ -89,7 +89,27 @@ func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc) { Source: src.Acquire, Selection: selection, }, - }, src.Close + } + + cleanups := []CleanupFunc{src.Close} + + if cfg.SnapshotCreateTargetDB() { + if targetURL := cfg.SnapshotTargetPostgresURL(); targetURL != "" { + target := postgres.NewLazyConn(targetURL) + checks = append(checks, &TargetCreateDBPrivilegeCheck{Target: target.Acquire}) + cleanups = append(cleanups, target.Close) + } + } + + return checks, func(ctx context.Context) error { + var firstErr error + for _, cleanup := range cleanups { + if err := cleanup(ctx); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr + } } // BuildSchemaChecks returns the schema-preflight checks applicable to cfg, From 2eb9dea58f857d85e9bed2c8c5cf89643f0c21fa Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Thu, 16 Jul 2026 22:52:27 +0700 Subject: [PATCH 6/7] connect without target database for CREATEDB check --- pkg/stream/preflight/access_test.go | 32 +++++++++++++++++++++++++++++ pkg/stream/preflight/builder.go | 25 ++++++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/pkg/stream/preflight/access_test.go b/pkg/stream/preflight/access_test.go index 4c4394bf..d1d635ef 100644 --- a/pkg/stream/preflight/access_test.go +++ b/pkg/stream/preflight/access_test.go @@ -643,6 +643,38 @@ func TestBuildChecks_SelectedAccessOnly(t *testing.T) { require.Equal(t, "source_sequence_select_privileges", checks[1].Name()) } +func TestRemoveDatabaseFromConnectionString(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + connection string + want string + }{ + { + name: "database is removed", + connection: "postgres://pgstream:secret@localhost:5432/target_db?sslmode=disable", + want: "postgres://pgstream:secret@localhost:5432/?sslmode=disable", + }, + { + name: "postgres database is preserved", + connection: "postgres://pgstream:secret@localhost:5432/postgres?sslmode=disable", + want: "postgres://pgstream:secret@localhost:5432/postgres?sslmode=disable", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := removeDatabaseFromConnectionString(tc.connection) + + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} + func sourceWithRows(t *testing.T, rows []sourceTableSelectPrivilegeRow) postgres.AcquireFunc { t.Helper() return sourceWithMockRows(privilegeRows(t, rows)) diff --git a/pkg/stream/preflight/builder.go b/pkg/stream/preflight/builder.go index 91154555..4818e7dc 100644 --- a/pkg/stream/preflight/builder.go +++ b/pkg/stream/preflight/builder.go @@ -4,6 +4,7 @@ package preflight import ( "context" + "strings" "github.com/xataio/pgstream/internal/postgres" "github.com/xataio/pgstream/pkg/stream" @@ -116,6 +117,16 @@ func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc) { if cfg.SnapshotCreateTargetDB() { if targetURL := cfg.SnapshotTargetPostgresURL(); targetURL != "" { + var err error + targetURL, err = removeDatabaseFromConnectionString(targetURL) + if err != nil { + checks = append(checks, &TargetCreateDBPrivilegeCheck{ + Target: func(context.Context) (postgres.Querier, error) { + return nil, err + }, + }) + return checks, joinCleanups(cleanups) + } target := postgres.NewLazyConn(targetURL) checks = append(checks, &TargetCreateDBPrivilegeCheck{Target: target.Acquire}) cleanups = append(cleanups, target.Close) @@ -133,6 +144,20 @@ func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc) { } } +// removeDatabaseFromConnectionString lets the CREATEDB preflight connect to +// the target server before the configured target database exists. +func removeDatabaseFromConnectionString(url string) (string, error) { + pgCfg, err := postgres.ParseConfig(url) + if err != nil { + return "", err + } + if pgCfg.Database == "" || pgCfg.Database == "postgres" { + return url, nil + } + + return strings.ReplaceAll(url, "/"+pgCfg.Database, "/"), nil +} + // BuildSchemaChecks returns the schema-preflight checks applicable to cfg, // plus a cleanup function that closes the shared source (and, when the target // is Postgres, target) connection. Schema checks cover every table pgstream From 5745369c94d0af5cc3de596f858cf2c5bccb40aa Mon Sep 17 00:00:00 2001 From: lghuy05 Date: Thu, 16 Jul 2026 22:59:24 +0700 Subject: [PATCH 7/7] resolve gofumpt --- pkg/stream/preflight/access.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/pkg/stream/preflight/access.go b/pkg/stream/preflight/access.go index 6af73229..63e660e9 100644 --- a/pkg/stream/preflight/access.go +++ b/pkg/stream/preflight/access.go @@ -195,7 +195,6 @@ func (c *TargetCreateDBPrivilegeCheck) Run(ctx context.Context) ([]Finding, erro return []Finding{ {Message: targetCreateDBPrivilegeMessage(role)}, }, nil - } type sourceTableSelectPrivilegeRow struct { @@ -284,5 +283,4 @@ func targetCreateDBPrivilegeMessage(role string) string { "target role %q lacks CREATEDB; run ALTER ROLE %s CREATEDB", role, quotedRole, ) - }