diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index f502fbf05..4b362cf3a 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -87,6 +87,7 @@ Some things belong to an instance rather than to a workspace, so they are not ap | Warmup pool membership | Pools are shared across every workspace on an instance, so membership is re-earned rather than asserted by a file | | Domain authentication timings | The verdict travels (public DNS reads the same anywhere), but the destination re-checks before it can stop any sending, so a mailbox is never blocked on an observation the new instance never made | | Scheduled deletions | A pending deletion from the source must never follow the workspace to its new home | +| Failure and delivery counters | A webhook endpoint's failure streak and auto-disable state, and whether a notification's email already went out, describe what happened on the source. They start fresh, so an endpoint is not pre-disabled on the new instance and a notification is not re-sent | An import runs as one transaction. If anything fails, nothing lands and the workspace is untouched. diff --git a/internal/app/orgtransfer/import.go b/internal/app/orgtransfer/import.go index 15403d7bf..f46ffe7ac 100644 --- a/internal/app/orgtransfer/import.go +++ b/internal/app/orgtransfer/import.go @@ -220,10 +220,26 @@ func (s *service) importTable( writable[c.Name] = true } } + + // A reset column is left out of the statement entirely rather than written + // as NULL, so the destination's own DEFAULT fills it. Writing NULL only + // worked for the nullable ones and aborted the whole import on a NOT NULL + // counter such as webhook_endpoints.consecutive_failures. + reset := make(map[string]bool, len(t.ResetOnImport)) + for _, c := range t.ResetOnImport { + reset[c] = true + } + insertCols := make([]string, 0, len(mt.Columns)) + unknown := 0 for _, c := range mt.Columns { - if writable[c] { + switch { + case reset[c]: + // deliberately omitted + case writable[c]: insertCols = append(insertCols, c) + default: + unknown++ } } if len(insertCols) == 0 { @@ -231,9 +247,9 @@ func (s *service) importTable( } var warnings []string - if dropped := len(mt.Columns) - len(insertCols); dropped > 0 { + if unknown > 0 { warnings = append(warnings, fmt.Sprintf( - "%s: %d column(s) in the archive do not exist here and were ignored.", t.Name, dropped)) + "%s: %d column(s) in the archive do not exist here and were ignored.", t.Name, unknown)) } var pk []string @@ -382,12 +398,6 @@ func (s *service) importRow( } } - for _, c := range t.ResetOnImport { - if _, ok := obj[c]; ok { - obj[c] = json.RawMessage(`null`) - } - } - for _, sc := range t.Secrets { raw, ok := obj[sc.Column] if !ok { diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index 885da7aa5..94085e4f2 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -69,8 +69,11 @@ type Table struct { // the rows. Blobs []BlobColumn - // ResetOnImport are columns set to NULL on the way in because they name - // something that exists only on the source instance. + // ResetOnImport are columns left out of the insert because they name + // something that exists only on the source instance: a queue handle, a + // health counter, a last-seen timestamp. Omitting them is what makes the + // destination's own column default apply, which is the only reset that + // works for a NOT NULL column (writing NULL into one aborts the import). ResetOnImport []string // ImportSkip exports the table for the record but never writes it back. @@ -195,11 +198,6 @@ var Tables = []Table{ ResetOnImport: []string{"last_used_at", "last_request_ip"}, Note: "Only the key hash travels, so existing keys keep working after the move without the secret ever leaving the source.", }, - { - Name: "webhook_endpoints", Group: models.OrgDataGroupCore, - Scope: scopeOrg, - ResetOnImport: []string{"last_success_at", "last_failure_at", "last_failure_reason", "consecutive_failures", "first_failure_at", "auto_disabled_at", "disabled_reason"}, - }, { Name: "oauth_applications", Group: models.OrgDataGroupCore, Scope: scopeOrg, @@ -209,6 +207,13 @@ var Tables = []Table{ Scope: scopeOrg, ResetOnImport: []string{"last_used_at"}, }, + { + // Below oauth_applications: an endpoint owned by an OAuth app carries + // oauth_application_id, so the app has to exist first. + Name: "webhook_endpoints", Group: models.OrgDataGroupCore, + Scope: scopeOrg, + ResetOnImport: []string{"last_success_at", "last_failure_at", "last_failure_reason", "consecutive_failures", "first_failure_at", "auto_disabled_at", "disabled_reason"}, + }, { Name: "outreach_settings", Group: models.OrgDataGroupCore, Scope: scopeOrg, @@ -245,11 +250,6 @@ var Tables = []Table{ Name: "contact_activities", Group: models.OrgDataGroupContacts, Scope: scopeOrg, }, - { - Name: "suppressed_recipients", Group: models.OrgDataGroupContacts, - Scope: scopeOrg, - Note: "Suppression must travel, or the destination re-mails people who already opted out.", - }, { Name: "contact_research_runs", Group: models.OrgDataGroupContacts, Scope: scopeOrgAlt, @@ -264,6 +264,15 @@ var Tables = []Table{ Name: "campaigns", Group: models.OrgDataGroupCampaigns, Scope: scopeOrg, }, + { + // A contacts-group table, but it sits here because campaign_id points at + // campaigns and the import applies this list top to bottom. Selecting + // contacts without campaigns is still fine: the reference is nullable, + // so it is cleared rather than dangling. + Name: "suppressed_recipients", Group: models.OrgDataGroupContacts, + Scope: scopeOrg, + Note: "Suppression must travel, or the destination re-mails people who already opted out.", + }, { Name: "campaign_folders", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, @@ -415,10 +424,6 @@ var Tables = []Table{ Name: "reply_templates", Group: models.OrgDataGroupAI, Scope: scopeOrg, }, - { - Name: "reply_intents", Group: models.OrgDataGroupAI, - Scope: scopeOrg, - }, // ---------- warmup ---------- { @@ -497,6 +502,13 @@ var Tables = []Table{ // The handle belongs to the source instance's queue. ResetOnImport: []string{"cloud_task_name"}, }, + { + // An AI-group table, but it sits here because task_id points at tasks. + // Selecting AI without the send pipeline is still fine: the reference is + // nullable, so it is cleared rather than dangling. + Name: "reply_intents", Group: models.OrgDataGroupAI, + Scope: scopeOrg, + }, { Name: "email_tasks", Group: models.OrgDataGroupSending, Scope: `task_id IN ` + orgTasks, diff --git a/internal/app/orgtransfer/spec_live_test.go b/internal/app/orgtransfer/spec_live_test.go new file mode 100644 index 000000000..ccd058caa --- /dev/null +++ b/internal/app/orgtransfer/spec_live_test.go @@ -0,0 +1,124 @@ +package orgtransfer + +import ( + "context" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Live checks of the transfer spec against the real schema. Skipped unless +// WARMBLY_TEST_DB is set (same convention as internal/scheduler): +// +// WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \ +// go test ./internal/app/orgtransfer/ -run Live -v +// +// Both rules below are properties of spec.go paired with the live schema, so +// neither can be checked without a database. Both were broken, and each break +// was a whole archive that refuses to import: a NOT NULL reset column aborted +// on the first row, and a table ordered above something it references aborted +// on a foreign key. A unit test over spec.go alone cannot see either. + +func specLivePool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + pool, err := pgxpool.New(context.Background(), dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(pool.Close) + if err := pool.Ping(context.Background()); err != nil { + t.Fatalf("ping: %v", err) + } + return pool +} + +// A reset column is omitted from the insert so the destination's DEFAULT +// applies. A NOT NULL column with no default has nothing to fall back on, so +// omitting it would abort the import just as writing NULL used to. +func TestLiveResetColumnsHaveSomethingToFallBackOn(t *testing.T) { + pool := specLivePool(t) + ctx := context.Background() + + for _, tbl := range Tables { + if tbl.ImportSkip { + continue + } + for _, col := range tbl.ResetOnImport { + var nullable bool + var hasDefault bool + err := pool.QueryRow(ctx, ` + SELECT is_nullable = 'YES', column_default IS NOT NULL + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2 + `, tbl.Name, col).Scan(&nullable, &hasDefault) + if err != nil { + t.Errorf("%s.%s is in ResetOnImport but not in the schema: %v", tbl.Name, col, err) + continue + } + if !nullable && !hasDefault { + t.Errorf("%s.%s is NOT NULL with no default, so resetting it aborts every import that carries this table", + tbl.Name, col) + } + } + } +} + +// Tables is applied top to bottom, so a table must sit below everything it +// references. A reference to a table that this run does not write is cleared +// by referencePlan; a reference to one it writes LATER is not, and lands as a +// foreign-key violation that fails the whole import. +func TestLiveTablesAreInDependencyOrder(t *testing.T) { + pool := specLivePool(t) + ctx := context.Background() + + pos := make(map[string]int, len(Tables)) + for i, tbl := range Tables { + pos[tbl.Name] = i + } + + for i, tbl := range Tables { + rows, err := pool.Query(ctx, ` + SELECT kcu.column_name, ccu.table_name + FROM information_schema.table_constraints tc + JOIN information_schema.key_column_usage kcu + ON kcu.constraint_name = tc.constraint_name + JOIN information_schema.constraint_column_usage ccu + ON ccu.constraint_name = tc.constraint_name + WHERE tc.constraint_type = 'FOREIGN KEY' + AND tc.table_schema = 'public' AND tc.table_name = $1 + `, tbl.Name) + if err != nil { + t.Fatalf("read foreign keys for %s: %v", tbl.Name, err) + } + type ref struct{ col, target string } + var refs []ref + for rows.Next() { + var r ref + if err := rows.Scan(&r.col, &r.target); err != nil { + rows.Close() + t.Fatalf("scan foreign key: %v", err) + } + refs = append(refs, r) + } + rows.Close() + + for _, r := range refs { + if r.target == tbl.Name { + continue // self-reference, ordered within the table's own rows + } + target, carried := pos[r.target] + if !carried { + continue // not in the archive; referencePlan clears or keeps it + } + if target > i { + t.Errorf("%s is at %d but its %s points at %s at %d: move it below", + tbl.Name, i, r.col, r.target, target) + } + } + } +}