knowledge-governance/internal/platform/postgres/migrations_test.go

package postgres

import (
	"os"
	"strings"
	"testing"
)

func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
	migrations, err := loadMigrations()
	if err != nil {
		t.Fatalf("loadMigrations() error = %v", err)
	}
	if len(migrations) != 39 {
		t.Fatalf("migration count = %d, want 39", len(migrations))
	}
	if migrations[0].Version != 1 || !strings.Contains(migrations[0].UpSQL, "CREATE TABLE IF NOT EXISTS jobs") {
		t.Fatalf("migration 1 = %#v", migrations[0])
	}
	if migrations[1].Version != 2 || !strings.Contains(migrations[1].UpSQL, "CREATE TABLE IF NOT EXISTS source_items") {
		t.Fatalf("migration 2 = %#v", migrations[1])
	}
	if migrations[2].Version != 3 || !strings.Contains(migrations[2].UpSQL, "CREATE TABLE IF NOT EXISTS acceptance_results") {
		t.Fatalf("migration 3 = %#v", migrations[2])
	}
	if migrations[3].Version != 4 || !strings.Contains(migrations[3].UpSQL, "source_connections_source_type_check") {
		t.Fatalf("migration 4 = %#v", migrations[3])
	}
	if migrations[4].Version != 5 || !strings.Contains(migrations[4].UpSQL, "CREATE TABLE IF NOT EXISTS source_scans") {
		t.Fatalf("migration 5 = %#v", migrations[4])
	}
	if migrations[5].Version != 6 || !strings.Contains(migrations[5].UpSQL, "CREATE TABLE IF NOT EXISTS publication_reviews") {
		t.Fatalf("migration 6 = %#v", migrations[5])
	}
	if migrations[7].Version != 8 || !strings.Contains(migrations[7].UpSQL, "CREATE TABLE IF NOT EXISTS directory_scope_rules") {
		t.Fatalf("migration 8 = %#v", migrations[7])
	}
	if migrations[8].Version != 9 || !strings.Contains(migrations[8].UpSQL, "CREATE TABLE IF NOT EXISTS source_catalog_snapshots") {
		t.Fatalf("migration 9 = %#v", migrations[8])
	}
	if migrations[9].Version != 10 || !strings.Contains(migrations[9].UpSQL, "Visibility policies used by the source catalog") {
		t.Fatalf("migration 10 = %#v", migrations[9])
	}
	if migrations[10].Version != 11 || !strings.Contains(migrations[10].UpSQL, "CREATE TABLE IF NOT EXISTS processing_confirmations") {
		t.Fatalf("migration 11 = %#v", migrations[10])
	}
	if migrations[13].Version != 14 || !strings.Contains(migrations[13].UpSQL, "ADD COLUMN IF NOT EXISTS started_at") {
		t.Fatalf("migration 14 = %#v", migrations[13])
	}
	if migrations[14].Version != 15 || !strings.Contains(migrations[14].UpSQL, "CREATE TABLE IF NOT EXISTS source_catalog_change_sets") || !strings.Contains(migrations[14].UpSQL, "CREATE TABLE IF NOT EXISTS source_catalog_change_items") {
		t.Fatalf("migration 15 = %#v", migrations[14])
	}
	if migrations[15].Version != 16 || !strings.Contains(migrations[15].UpSQL, "origin") || !strings.Contains(migrations[15].UpSQL, "source_catalog_change_items") {
		t.Fatalf("migration 16 = %#v", migrations[15])
	}
	if migrations[16].Version != 17 || !strings.Contains(migrations[16].UpSQL, "current_batch") || !strings.Contains(migrations[16].UpSQL, "total_batches") {
		t.Fatalf("migration 17 = %#v", migrations[16])
	}
	if migrations[17].Version != 18 || !strings.Contains(migrations[17].UpSQL, "jobs_state_check") || !strings.Contains(migrations[17].UpSQL, "cancelled") {
		t.Fatalf("migration 18 = %#v", migrations[17])
	}
	if migrations[18].Version != 19 || !strings.Contains(migrations[18].UpSQL, "CREATE TABLE IF NOT EXISTS operation_requests") || !strings.Contains(migrations[18].UpSQL, "UNIQUE (operation_type, operation_id)") {
		t.Fatalf("migration 19 = %#v", migrations[18])
	}
	if migrations[19].Version != 20 || !strings.Contains(migrations[19].UpSQL, "content_sha256") || !strings.Contains(migrations[19].UpSQL, "IF NOT EXISTS") {
		t.Fatalf("migration 20 = %#v", migrations[19])
	}
	if migrations[20].Version != 21 || !strings.Contains(migrations[20].UpSQL, "identity_confidence") || !strings.Contains(migrations[20].UpSQL, "reason_code") {
		t.Fatalf("migration 21 = %#v", migrations[20])
	}
	if migrations[21].Version != 22 || !strings.Contains(migrations[21].UpSQL, "compare_catalog_change_set") {
		t.Fatalf("migration 22 = %#v", migrations[21])
	}
	if migrations[22].Version != 23 || !strings.Contains(migrations[22].UpSQL, "move_count") || !strings.Contains(migrations[22].UpSQL, "'applying'") || !strings.Contains(migrations[22].UpSQL, "old_scope") {
		t.Fatalf("migration 23 = %#v", migrations[22])
	}
	if migrations[23].Version != 24 || !strings.Contains(migrations[23].UpSQL, "item_seq") || !strings.Contains(migrations[23].UpSQL, "change_set_id") {
		t.Fatalf("migration 24 = %#v", migrations[23])
	}
	if migrations[24].Version != 25 || !strings.Contains(migrations[24].UpSQL, "business_change_set_id") || !strings.Contains(migrations[24].UpSQL, "jobs_processing_identity") {
		t.Fatalf("migration 25 = %#v", migrations[24])
	}
	if migrations[25].Version != 26 || !strings.Contains(migrations[25].UpSQL, "job_failures") || !strings.Contains(migrations[25].UpSQL, "last_error_code") {
		t.Fatalf("migration 26 = %#v", migrations[25])
	}
	if migrations[26].Version != 27 || !strings.Contains(migrations[26].UpSQL, "worker_heartbeats") {
		t.Fatalf("migration 27 = %#v", migrations[26])
	}
	if migrations[27].Version != 28 || !strings.Contains(migrations[27].UpSQL, "release_change_sets") || !strings.Contains(migrations[27].UpSQL, "manifest_sha256") {
		t.Fatalf("migration 28 = %#v", migrations[27])
	}
	if migrations[28].Version != 29 || !strings.Contains(migrations[28].UpSQL, "rolling_back") || !strings.Contains(migrations[28].UpSQL, "failure_level") {
		t.Fatalf("migration 29 = %#v", migrations[28])
	}
	if migrations[29].Version != 30 || !strings.Contains(migrations[29].UpSQL, "release_change_set_snapshot_links") || !strings.Contains(migrations[29].UpSQL, "release_change_set_catalog_links") {
		t.Fatalf("migration 30 = %#v", migrations[29])
	}
	if migrations[30].Version != 31 || !strings.Contains(migrations[30].UpSQL, "source_connection_revisions") || !strings.Contains(migrations[30].UpSQL, "uq_source_connection_active_revision") || !strings.Contains(migrations[30].UpSQL, "next_refresh_at") {
		t.Fatalf("migration 31 = %#v", migrations[30])
	}
	if migrations[31].Version != 32 || !strings.Contains(migrations[31].UpSQL, "CREATE TABLE IF NOT EXISTS users") || !strings.Contains(migrations[31].UpSQL, "user_sessions") {
		t.Fatalf("migration 32 = %#v", migrations[31])
	}
	if migrations[34].Version != 35 || !strings.Contains(migrations[34].UpSQL, "CREATE TABLE IF NOT EXISTS filename_filter_rules") || !strings.Contains(migrations[34].UpSQL, "'filtered'") {
		t.Fatalf("migration 35 = %#v", migrations[34])
	}
}

func TestProcessingConfirmationMigrationSeparatesPlanningFromUpload(t *testing.T) {
	up, err := os.ReadFile("migrations/000011_processing_confirmations.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000011_processing_confirmations.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"add column if not exists target_scope",
		"update test_publications as publication",
		"review.decided_visibility",
		"source.visibility",
		"check (target_scope in ('internal', 'external'))",
		"create table if not exists processing_confirmations",
		"references test_publications(id) on delete restrict",
		"operator_ref",
		"reason",
		"unique (publication_id)",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("processing confirmation migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	if !strings.Contains(downSQL, "drop table if exists processing_confirmations") || !strings.Contains(downSQL, "drop column if exists target_scope") {
		t.Fatal("processing confirmation down migration must remove only migration 11 objects")
	}
}

func TestVersionedUISystemSettingsMigration(t *testing.T) {
	up, err := os.ReadFile("migrations/000013_ui_system_settings.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000013_ui_system_settings.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{"create table if not exists ui_setting_revisions", "'validated', 'active', 'retired', 'failed'", "create unique index", "where state = 'active'", "insert into ui_setting_revisions"} {
		if !strings.Contains(upSQL, required) {
			t.Fatalf("UI settings migration missing %q", required)
		}
	}
	if !strings.Contains(strings.ToLower(string(down)), "drop table if exists ui_setting_revisions") {
		t.Fatal("UI settings rollback must remove only its revision table")
	}
}

func TestReleaseActionMigrationSeparatesUpsertAndDelete(t *testing.T) {
	up, err := os.ReadFile("migrations/000012_release_actions.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000012_release_actions.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"alter table release_batch_items", "alter table release_item_executions",
		"add column if not exists action", "check (action in ('upsert', 'delete'))",
		"drop constraint if exists release_batch_items_publication_id_key",
		"drop constraint if exists release_batch_items_review_id_key",
	} {
		if !strings.Contains(upSQL, required) {
			t.Fatalf("release action migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	if !strings.Contains(downSQL, "drop column if exists action") {
		t.Fatal("release action rollback must remove action columns")
	}
}

func TestCatalogPolicyMigrationAllowsBothAndBlockedWithoutChangingPublicationReviewScope(t *testing.T) {
	up, err := os.ReadFile("migrations/000010_catalog_policies.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000010_catalog_policies.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"visibility policies used by the source catalog",
		"directory_scope_rules_visibility_check",
		"source_item_scope_overrides_visibility_check",
		"scope_change_events_previous_visibility_check",
		"scope_change_events_decided_visibility_check",
		"'internal', 'external', 'both', 'blocked'",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("catalog policy migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	if !strings.Contains(downSQL, "update source_item_scope_overrides") || !strings.Contains(downSQL, "update directory_scope_rules") {
		t.Fatalf("catalog policy down migration must normalize extended policy values before restoring constraints")
	}
}

func TestSourceCatalogMigrationDefinesImmutableHierarchicalSnapshots(t *testing.T) {
	up, err := os.ReadFile("migrations/000009_source_catalog.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000009_source_catalog.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"create table if not exists source_catalog_snapshots",
		"create table if not exists source_catalog_nodes",
		"create table if not exists source_catalog_heads",
		"check (node_type in ('folder', 'file'))",
		"primary key (snapshot_id, node_id)",
		"foreign key (snapshot_id, parent_node_id)",
		"references source_items(id) on delete restrict",
		"create index if not exists idx_source_catalog_nodes_parent",
		"unique (source_connection_id, snapshot_sha256)",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("source catalog migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	ordered := []string{"drop table if exists source_catalog_heads", "drop table if exists source_catalog_nodes", "drop table if exists source_catalog_snapshots"}
	previous := -1
	for _, statement := range ordered {
		position := strings.Index(downSQL, statement)
		if position < 0 || position <= previous {
			t.Fatalf("source catalog down order invalid for %q", statement)
		}
		previous = position
	}
}

func TestScopeRulesMigrationDefinesAuditedDirectoryAndFileOverrides(t *testing.T) {
	up, err := os.ReadFile("migrations/000008_scope_rules.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000008_scope_rules.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"create table if not exists directory_scope_rules",
		"create table if not exists source_item_scope_overrides",
		"create table if not exists scope_change_events",
		"primary key (source_connection_id, directory_path)",
		"primary key (source_item_id)",
		"check (visibility in ('internal', 'external'))",
		"check (target_type in ('directory', 'item'))",
		"references source_connections(id) on delete restrict",
		"references source_items(id) on delete restrict",
		"create index if not exists idx_scope_change_events_target",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("scope migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	for _, required := range []string{
		"drop table if exists scope_change_events",
		"drop table if exists source_item_scope_overrides",
		"drop table if exists directory_scope_rules",
	} {
		if !strings.Contains(downSQL, required) {
			t.Errorf("scope down migration missing %q", required)
		}
	}
}

func TestJobsMigrationDefinesDurableLeaseQueue(t *testing.T) {
	body, err := os.ReadFile("migrations/000001_jobs.up.sql")
	if err != nil {
		t.Fatalf("read jobs migration: %v", err)
	}
	sql := strings.ToLower(string(body))
	for _, required := range []string{
		"create table if not exists schema_migrations",
		"create table if not exists jobs",
		"idempotency_key",
		"lease_owner",
		"lease_until",
		"available_at",
		"max_attempts",
		"check (state in",
		"unique (idempotency_key)",
	} {
		if !strings.Contains(sql, required) {
			t.Errorf("migration missing %q", required)
		}
	}
}

func TestSourceGovernanceMigrationDefinesIdentityAndStateContracts(t *testing.T) {
	body, err := os.ReadFile("migrations/000002_source_governance.up.sql")
	if err != nil {
		t.Fatalf("read source governance migration: %v", err)
	}
	sql := strings.ToLower(string(body))
	for _, required := range []string{
		"create table if not exists source_connections",
		"create table if not exists source_items",
		"create table if not exists content_objects",
		"create table if not exists source_item_versions",
		"create table if not exists test_publications",
		"unique (source_connection_id, stable_external_id)",
		"unique (sha256)",
		"unique (source_item_id, revision)",
		"unique (source_item_version_id, test_kb_id)",
		"check (source_type in ('pp', 'alist', 'dingtalk'))",
		"check (sha256 ~ '^[0-9a-f]{64}$')",
		"check (source_change in ('unchanged', 'added', 'updated', 'deleted', 'unavailable'))",
		"check (content_state in ('unknown', 'downloading', 'verified', 'invalid', 'over_limit', 'failed'))",
		"check (ingest_state in ('pending', 'uploading', 'uploaded', 'failed'))",
		"check (processing_state in ('pending', 'processing', 'finalizing', 'completed', 'failed', 'stalled'))",
		"check (retrieval_state in ('unpublished', 'enabled', 'disabling', 'disabled', 'enabling', 'failed'))",
		"references source_connections(id)",
		"references source_items(id)",
		"references content_objects(id)",
		"references source_item_versions(id)",
		"raw_record_summary jsonb",
	} {
		if !strings.Contains(sql, required) {
			t.Errorf("source governance migration missing %q", required)
		}
	}
}

func TestSourceGovernanceDownMigrationDropsTablesInDependencyOrder(t *testing.T) {
	body, err := os.ReadFile("migrations/000002_source_governance.down.sql")
	if err != nil {
		t.Fatalf("read source governance down migration: %v", err)
	}
	sql := strings.ToLower(string(body))
	ordered := []string{
		"drop table if exists test_publications",
		"drop table if exists source_item_versions",
		"drop table if exists content_objects",
		"drop table if exists source_items",
		"drop table if exists source_connections",
	}
	previous := -1
	for _, statement := range ordered {
		position := strings.Index(sql, statement)
		if position < 0 {
			t.Fatalf("down migration missing %q", statement)
		}
		if position <= previous {
			t.Fatalf("%q is not in reverse dependency order", statement)
		}
		previous = position
	}
}

func TestTestOrchestrationMigrationDefinesPollingAndAcceptanceContracts(t *testing.T) {
	body, err := os.ReadFile("migrations/000003_test_orchestration.up.sql")
	if err != nil {
		t.Fatalf("read test orchestration migration: %v", err)
	}
	sql := strings.ToLower(string(body))
	for _, required := range []string{
		"alter table test_publications add column if not exists poll_sequence",
		"alter table test_publications add column if not exists last_remote_status",
		"alter table test_publications add column if not exists last_observed_at",
		"alter table test_publications add column if not exists supersedes_publication_id",
		"references test_publications(id) on delete restrict",
		"create table if not exists acceptance_results",
		"references test_publications(id) on delete restrict",
		"check (check_type in ('knowledge_exists', 'parse_completed', 'chunks_present', 'retrieval_hit', 'citation_matches'))",
		"check (state in ('pending', 'passed', 'failed', 'manual_required'))",
		"check (jsonb_typeof(evidence_summary) = 'object')",
		"unique (publication_id, run_id, check_type)",
		"create index if not exists idx_test_publications_poll",
		"create index if not exists idx_acceptance_results_publication",
	} {
		if !strings.Contains(sql, required) {
			t.Errorf("test orchestration migration missing %q", required)
		}
	}
}

func TestTestOrchestrationDownMigrationRemovesDependentObjectsFirst(t *testing.T) {
	body, err := os.ReadFile("migrations/000003_test_orchestration.down.sql")
	if err != nil {
		t.Fatalf("read test orchestration down migration: %v", err)
	}
	sql := strings.ToLower(string(body))
	acceptance := strings.Index(sql, "drop table if exists acceptance_results")
	publicationColumns := strings.Index(sql, "alter table test_publications")
	if acceptance < 0 || publicationColumns < 0 {
		t.Fatalf("down migration missing dependent object removal: %s", sql)
	}
	if acceptance > publicationColumns {
		t.Fatal("acceptance_results must be dropped before test_publications columns")
	}
}

func TestManualSourceMigrationExpandsWithoutRewritingAndDowngradesFailClosed(t *testing.T) {
	up, err := os.ReadFile("migrations/000004_manual_source.up.sql")
	if err != nil {
		t.Fatalf("read manual source up migration: %v", err)
	}
	down, err := os.ReadFile("migrations/000004_manual_source.down.sql")
	if err != nil {
		t.Fatalf("read manual source down migration: %v", err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"drop constraint if exists source_connections_source_type_check",
		"check (source_type in ('pp', 'alist', 'dingtalk', 'manual'))",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("manual source up migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	for _, required := range []string{
		"if exists",
		"source_type = 'manual'",
		"raise exception",
		"check (source_type in ('pp', 'alist', 'dingtalk'))",
	} {
		if !strings.Contains(downSQL, required) {
			t.Errorf("manual source down migration missing %q", required)
		}
	}
}

func TestSourceScanMigrationDefinesAuditableSingleActiveScan(t *testing.T) {
	up, err := os.ReadFile("migrations/000005_source_scans.up.sql")
	if err != nil {
		t.Fatalf("read source scans up migration: %v", err)
	}
	down, err := os.ReadFile("migrations/000005_source_scans.down.sql")
	if err != nil {
		t.Fatalf("read source scans down migration: %v", err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"create table if not exists source_scans",
		"create table if not exists source_scan_changes",
		"references source_connections(id) on delete restrict",
		"references source_scans(id) on delete restrict",
		"check (state in ('pending', 'running', 'completed', 'failed'))",
		"check (change_type in ('added', 'updated', 'unavailable'))",
		"check (snapshot_sha256 = '' or snapshot_sha256 ~ '^[0-9a-f]{64}$')",
		"check (observed_items >= 0)",
		"check (unchanged_items >= 0)",
		"check (added_items >= 0)",
		"check (updated_items >= 0)",
		"check (over_limit_items >= 0)",
		"check (missing_items >= 0)",
		"check (planned_uploads >= 0)",
		"create unique index if not exists idx_source_scans_one_active",
		"where state in ('pending', 'running')",
		"unique (source_scan_id, stable_external_id)",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("source scans up migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	changes := strings.Index(downSQL, "drop table if exists source_scan_changes")
	scans := strings.Index(downSQL, "drop table if exists source_scans")
	if changes < 0 || scans < 0 || changes > scans {
		t.Fatal("source scans down migration must drop changes before scans")
	}
}

func TestReviewAndSealMigrationDefinesImmutableHumanGate(t *testing.T) {
	up, err := os.ReadFile("migrations/000006_review_and_seal.up.sql")
	if err != nil {
		t.Fatalf("read review and seal up migration: %v", err)
	}
	down, err := os.ReadFile("migrations/000006_review_and_seal.down.sql")
	if err != nil {
		t.Fatalf("read review and seal down migration: %v", err)
	}
	upSQL := strings.ToLower(string(up))
	for _, required := range []string{
		"create table if not exists publication_reviews",
		"references test_publications(id) on delete restrict",
		"check (decision in ('approved', 'rejected'))",
		"check (decided_visibility in ('internal', 'external'))",
		"acceptance_run_id",
		"publication_updated_at",
		"create table if not exists release_batches",
		"check (state in ('draft', 'sealed', 'cancelled'))",
		"policy_version",
		"create table if not exists release_batch_items",
		"references publication_reviews(id) on delete restrict",
		"unique (publication_id)",
		"decided_visibility",
	} {
		if !strings.Contains(upSQL, required) {
			t.Errorf("review and seal migration missing %q", required)
		}
	}
	downSQL := strings.ToLower(string(down))
	ordered := []string{
		"drop table if exists release_batch_items",
		"drop table if exists release_batches",
		"drop table if exists publication_reviews",
	}
	previous := -1
	for _, statement := range ordered {
		position := strings.Index(downSQL, statement)
		if position < 0 || position <= previous {
			t.Fatalf("review and seal down migration has invalid dependency order for %q", statement)
		}
		previous = position
	}
}

func TestReleaseExecutionMigrationDefinesAuthorizedAuditableJobs(t *testing.T) {
	up, err := os.ReadFile("migrations/000007_release_executions.up.sql")
	if err != nil {
		t.Fatal(err)
	}
	down, err := os.ReadFile("migrations/000007_release_executions.down.sql")
	if err != nil {
		t.Fatal(err)
	}
	text := strings.ToLower(string(up))
	for _, required := range []string{
		"create table if not exists release_executions",
		"operation in ('publish', 'rollback')",
		"state in ('pending', 'running', 'completed', 'failed')",
		"target_environment = 'production'",
		"references release_batches(id) on delete restrict",
		"references jobs(id) on delete restrict",
		"create unique index if not exists uq_release_publish_batch",
		"create table if not exists release_item_executions",
		"bundle_sha256", "previous_production_revision", "production_revision", "error_code",
	} {
		if !strings.Contains(text, required) {
			t.Fatalf("release execution migration missing %q", required)
		}
	}
	if strings.Contains(text, "api_key") || strings.Contains(text, "password") || strings.Contains(text, "secret") {
		t.Fatal("release execution migration must not persist production credentials")
	}
	downText := strings.ToLower(string(down))
	items := strings.Index(downText, "drop table if exists release_item_executions")
	executions := strings.Index(downText, "drop table if exists release_executions")
	if items < 0 || executions < 0 || items > executions {
		t.Fatal("release execution down migration must drop item executions before executions")
	}
}

func TestClaimSQLUsesSkipLockedAndReclaimsExpiredRunningJobs(t *testing.T) {
	sql := strings.ToLower(claimJobSQL)
	for _, required := range []string{
		"for update of jobs skip locked",
		"ingestion_concurrency",
		"state in ('pending', 'retry')",
		"state = 'running'",
		"lease_until <",
		"attempt = j.attempt + 1",
	} {
		if !strings.Contains(sql, required) {
			t.Errorf("claim SQL missing %q", required)
		}
	}
}

func TestMutationSQLPreservesIdempotencyAndLeaseOwnership(t *testing.T) {
	tests := []struct {
		name     string
		sql      string
		required []string
	}{
		{
			name: "enqueue",
			sql:  enqueueJobSQL,
			required: []string{
				"on conflict (idempotency_key)",
				"do update set idempotency_key = excluded.idempotency_key",
				"returning id",
			},
		},
		{
			name: "complete",
			sql:  completeJobSQL,
			required: []string{
				"state = 'running'",
				"lease_owner = $2",
				"state = 'completed'",
			},
		},
		{
			name: "heartbeat",
			sql:  heartbeatJobSQL,
			required: []string{
				"state = 'running'",
				"lease_owner = $2",
				"lease_until = $4",
			},
		},
		{
			name: "fail",
			sql:  failJobSQL,
			required: []string{
				"case when attempt >= max_attempts then 'failed'",
				"else 'retry'",
				"lease_owner = null",
				"select id from updated",
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sql := strings.ToLower(tt.sql)
			for _, required := range tt.required {
				if !strings.Contains(sql, required) {
					t.Errorf("SQL missing %q", required)
				}
			}
		})
	}
}