From 019b45fca78e5a388bba085389719e15d9354f06 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Mon, 24 Aug 2026 08:57:05 -0700 Subject: [PATCH] feat: unblock workspace import for any archive carrying a webhook endpoint, a notification, or a campaign-scoped suppression: ResetOnImport columns are now left out of the insert so the destination's own DEFAULT applies instead of writing NULL, which aborted the whole transaction on the four NOT NULL ones (webhook_endpoints.consecutive_failures, notifications.email_state and email_attempts, ai_mcp_servers.last_error) and made a webhook endpoint arrive carrying the source's failure streak and auto-disable state, and three tables that sat above something they reference are moved below it (webhook_endpoints under oauth_applications, suppressed_recipients under campaigns, reply_intents under tasks) because referencePlan only clears a reference the run does not write at all, so a forward reference to a table written later landed as a foreign-key violation; both classes are now guarded against the live schema in TestLiveResetColumnsHaveSomethingToFallBackOn and TestLiveTablesAreInDependencyOrder, and a cross-instance export/import round trip carrying all three previously fatal tables completes with failure counters back at their defaults and the suppression rows still attributed to their campaign --- .../docs/guides/workspace-export-import.mdx | 1 + internal/app/orgtransfer/import.go | 28 ++-- internal/app/orgtransfer/spec.go | 44 ++++--- internal/app/orgtransfer/spec_live_test.go | 124 ++++++++++++++++++ 4 files changed, 172 insertions(+), 25 deletions(-) create mode 100644 internal/app/orgtransfer/spec_live_test.go 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) + } + } + } +}