Merge pull request #176 from warmbly/fix/orgtransfer-import-blockers

Workspace import fails on any archive with a webhook endpoint, a notification, or a campaign-scoped suppression

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.
 

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 {

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,

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)
+			}
+		}
+	}
+}