query-orchestration/internal/database/repository/milestone2_migrations_test.go

//go:build milestone2

// Package repository_test — Milestone 2 §2.1 and §2.2 failing migration
// shape tests.
//
// Build tag: this file is gated behind `milestone2` so the default
// `task fullsuite:ci` run (which does not pass the tag) skips every
// test in here. That keeps Milestone 1's green bar intact while the
// RED tests sit on disk as the TDD target for the golang engineer.
// Once migrations 129 and 130 land, the engineer runs
//
//	go test -tags=milestone2 ./internal/database/repository/...
//
// to flip them to green before running fullsuite:ci without the tag
// and confirming the whole suite still passes.
//
// These tests fail by design until the golang engineer lands migrations
// 129 (document_custom_metadata table) and 130 (schema invariant triggers).
// Every test in this file asserts DB-level structure or behaviour defined
// in plan Section 4.3 and Section 4.7 of plans/mutable.metadata.plan.combo.v4.md
// against a real Postgres 17 testcontainer stood up by internal/test.CreateDB.
//
// No mocks. Every assertion is an introspection or behavioural write
// against the real DB. See client_metadata_schemas_migration_test.go for
// the same pattern applied to migration 127.
package repository_test

import (
	"context"
	"errors"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/serviceconfig"
	"queryorchestration/internal/test"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/stretchr/testify/require"
)

// ---------- §2.1 migration 129 — document_custom_metadata table ----------

// TestMigration129_DocumentCustomMetadataShape asserts the structural shape
// defined by plan Section 4.3. Fails until migration 129 exists.
//
// v4 explicitly removes v3's `schema_id` column from this table (see plan
// Section 4.3 "No schema_id column (v4 simplification)"), so the test
// asserts the column is NOT present.
func TestMigration129_DocumentCustomMetadataShape(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	t.Run("table_exists", func(t *testing.T) {
		var exists bool
		err := pool.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM information_schema.tables
				WHERE table_schema = 'public'
				  AND table_name = 'document_custom_metadata'
			)
		`).Scan(&exists)
		require.NoError(t, err)
		require.True(t, exists, "document_custom_metadata table must exist")
	})

	t.Run("columns", func(t *testing.T) {
		type colSpec struct {
			dataType   string
			isNullable string // "YES" or "NO"
		}
		expected := map[string]colSpec{
			"id":          {"uuid", "NO"},
			"document_id": {"uuid", "NO"},
			"metadata":    {"jsonb", "NO"},
			"version":     {"integer", "NO"},
			"created_at":  {"timestamp with time zone", "NO"},
			"created_by":  {"character varying", "NO"},
		}

		rows, err := pool.Query(ctx, `
			SELECT column_name, data_type, is_nullable
			FROM information_schema.columns
			WHERE table_schema = 'public'
			  AND table_name = 'document_custom_metadata'
		`)
		require.NoError(t, err)
		defer rows.Close()

		found := map[string]colSpec{}
		for rows.Next() {
			var name, dataType, nullable string
			require.NoError(t, rows.Scan(&name, &dataType, &nullable))
			found[name] = colSpec{dataType: dataType, isNullable: nullable}
		}
		require.NoError(t, rows.Err())

		require.Equal(t, len(expected), len(found),
			"document_custom_metadata column count mismatch; expected %d, got %v",
			len(expected), found)
		for name, want := range expected {
			got, ok := found[name]
			require.Truef(t, ok, "missing column %q", name)
			require.Equalf(t, want.dataType, got.dataType,
				"column %q data_type mismatch", name)
			require.Equalf(t, want.isNullable, got.isNullable,
				"column %q is_nullable mismatch", name)
		}
	})

	t.Run("no_schema_id_column_v4_simplification", func(t *testing.T) {
		// v4 deliberately removes the v3 schema_id column. If a future
		// contributor accidentally reintroduces it via migration 129 or
		// later, this assertion fails loudly with a pointer back to the
		// plan's Section 4.3 design note.
		var exists bool
		err := pool.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM information_schema.columns
				WHERE table_schema = 'public'
				  AND table_name = 'document_custom_metadata'
				  AND column_name = 'schema_id'
			)
		`).Scan(&exists)
		require.NoError(t, err)
		require.False(t, exists,
			"document_custom_metadata must NOT have a schema_id column in v4 — see plan Section 4.3 'No schema_id column (v4 simplification)'")
	})

	t.Run("primary_key_on_id", func(t *testing.T) {
		cols, err := fetchDCMConstraintColumns(ctx, pool, "p", "")
		require.NoError(t, err)
		require.Equal(t, []string{"id"}, cols,
			"document_custom_metadata primary key must be on (id)")
	})

	t.Run("unique_document_id_version", func(t *testing.T) {
		cols, err := fetchDCMConstraintColumns(ctx, pool, "u", "uq_doc_custom_metadata_version")
		require.NoError(t, err)
		require.Equal(t, []string{"document_id", "version"}, cols,
			"uq_doc_custom_metadata_version must be UNIQUE on (document_id, version)")
	})

	t.Run("fk_document_cascade", func(t *testing.T) {
		var (
			confdeltype string
			refTable    string
			localColumn string
			foreignCol  string
		)
		err := pool.QueryRow(ctx, `
			SELECT
				c.confdeltype::text,
				rt.relname,
				la.attname,
				fa.attname
			FROM pg_constraint c
			JOIN pg_class t  ON t.oid  = c.conrelid
			JOIN pg_class rt ON rt.oid = c.confrelid
			JOIN pg_attribute la ON la.attrelid = c.conrelid  AND la.attnum = c.conkey[1]
			JOIN pg_attribute fa ON fa.attrelid = c.confrelid AND fa.attnum = c.confkey[1]
			WHERE c.contype = 'f'
			  AND t.relname = 'document_custom_metadata'
			  AND la.attname = 'document_id'
		`).Scan(&confdeltype, &refTable, &localColumn, &foreignCol)
		require.NoError(t, err,
			"FK on document_custom_metadata.document_id must exist")
		require.Equal(t, "documents", refTable, "FK must reference documents table")
		require.Equal(t, "document_id", localColumn)
		require.Equal(t, "id", foreignCol, "FK must reference documents(id)")
		require.Equal(t, "c", confdeltype,
			"FK must be ON DELETE CASCADE (confdeltype='c')")
	})

	t.Run("index_doc_version_desc", func(t *testing.T) {
		// idx_dcm_doc_version_desc is the only index v4 needs on this table.
		// It backs GetCurrentDocumentCustomMetadata's `ORDER BY version DESC
		// LIMIT 1` lookup (plan Section 4.5).
		var exists bool
		err := pool.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM pg_indexes
				WHERE schemaname = 'public'
				  AND tablename  = 'document_custom_metadata'
				  AND indexname  = 'idx_dcm_doc_version_desc'
			)
		`).Scan(&exists)
		require.NoError(t, err)
		require.True(t, exists,
			"idx_dcm_doc_version_desc must exist on document_custom_metadata")
	})
}

// TestMigration129_FKCascadeDeletesMetadata proves that deleting a document
// row cascades through to the document_custom_metadata table, removing any
// metadata rows for that document. This is the §2.1 "delete of a document
// removes its metadata rows" behavioural assertion.
func TestMigration129_FKCascadeDeletesMetadata(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "DCM_FK_CASCADE"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "dcm-fk-cascade",
		Clientid: clientID,
	})
	require.NoError(t, err)

	// Create a document.
	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "dcm-fk-cascade-doc",
	})
	require.NoError(t, err)

	// Insert two metadata rows directly.
	_, err = pool.Exec(ctx, `
		INSERT INTO document_custom_metadata (document_id, metadata, version, created_by)
		VALUES ($1, $2::jsonb, 1, 'tester'),
		       ($1, $3::jsonb, 2, 'tester')
	`, docID, `{"k":"v1"}`, `{"k":"v2"}`)
	require.NoError(t, err)

	var before int
	require.NoError(t, pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM document_custom_metadata WHERE document_id = $1`,
		docID).Scan(&before))
	require.Equal(t, 2, before)

	// Delete the document — cascade should remove metadata rows.
	_, err = pool.Exec(ctx, `DELETE FROM documents WHERE id = $1`, docID)
	require.NoError(t, err)

	var after int
	require.NoError(t, pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM document_custom_metadata WHERE document_id = $1`,
		docID).Scan(&after))
	require.Equal(t, 0, after,
		"ON DELETE CASCADE must remove every metadata row for the deleted document")
}

// TestMigration129_UniqueDocumentVersionRejected proves the
// uq_doc_custom_metadata_version UNIQUE constraint rejects a second row with
// the same (document_id, version) tuple. Behavioural counterpart to the
// shape-level assertion in TestMigration129_DocumentCustomMetadataShape.
func TestMigration129_UniqueDocumentVersionRejected(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "DCM_UNIQ"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "dcm-unique-test",
		Clientid: clientID,
	})
	require.NoError(t, err)

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "dcm-unique-doc",
	})
	require.NoError(t, err)

	const insertSQL = `
		INSERT INTO document_custom_metadata (document_id, metadata, version, created_by)
		VALUES ($1, $2::jsonb, $3, $4)
	`
	_, err = pool.Exec(ctx, insertSQL, docID, `{"k":"v1"}`, 1, "tester")
	require.NoError(t, err, "first insert must succeed")

	_, err = pool.Exec(ctx, insertSQL, docID, `{"k":"v2"}`, 1, "tester")
	require.Error(t, err, "second insert with same (document_id,version) must fail")

	var pgErr *pgconn.PgError
	require.True(t, errors.As(err, &pgErr),
		"expected *pgconn.PgError, got %T: %v", err, err)
	require.Equal(t, "23505", pgErr.Code,
		"expected unique_violation SQLSTATE 23505")
	require.Equal(t, "uq_doc_custom_metadata_version", pgErr.ConstraintName,
		"expected the duplicate to trip uq_doc_custom_metadata_version")
}

// TestMigration129_UpDownRoundtrip proves the migration 129 down.sql is
// complete: running it tears the table out cleanly, and re-running the
// up.sql re-creates it.
func TestMigration129_UpDownRoundtrip(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const (
		upPath   = "../migrations/00000000000129_create_document_custom_metadata.up.sql"
		downPath = "../migrations/00000000000129_create_document_custom_metadata.down.sql"
	)
	upSQL, err := os.ReadFile(upPath)
	require.NoError(t, err, "must be able to read migration 129 up.sql")
	downSQL, err := os.ReadFile(downPath)
	require.NoError(t, err, "must be able to read migration 129 down.sql")

	// Leave DB in migrated state on test exit.
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), string(upSQL))
	})

	// Migration 130 may create triggers that reference document_custom_metadata
	// (Trigger 1 reads from it via EXISTS). If 130 has already landed, bringing
	// 129 down while 130 is still attached would fail; the test engineer keeps
	// this file ordered so 130 is torn down first. For now, 130 does not exist,
	// so a plain down is fine.
	_, err = pool.Exec(ctx, string(downSQL))
	require.NoError(t, err, "down migration 129 must apply cleanly")

	var tableOID *string
	err = pool.QueryRow(ctx,
		`SELECT to_regclass('document_custom_metadata')::text`).Scan(&tableOID)
	require.NoError(t, err)
	require.Nil(t, tableOID,
		"document_custom_metadata table must be gone after down")

	_, err = pool.Exec(ctx, string(upSQL))
	require.NoError(t, err, "up migration 129 must re-apply cleanly")

	err = pool.QueryRow(ctx,
		`SELECT to_regclass('document_custom_metadata')::text`).Scan(&tableOID)
	require.NoError(t, err)
	require.NotNil(t, tableOID,
		"document_custom_metadata must exist again after re-applying up")
}

// ---------- §2.2 migration 130 — schema invariant triggers ----------

// TestMigration130_TriggersExist asserts the presence of both trigger
// functions and their attachments per plan Section 4.7. Fails until
// migration 130 exists.
func TestMigration130_TriggersExist(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	t.Run("trigger1_function_exists", func(t *testing.T) {
		var exists bool
		err := pool.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM pg_proc p
				JOIN pg_namespace n ON n.oid = p.pronamespace
				WHERE n.nspname = 'public'
				  AND p.proname = 'trg_prevent_schema_reassignment'
			)
		`).Scan(&exists)
		require.NoError(t, err)
		require.True(t, exists,
			"trg_prevent_schema_reassignment() function must exist in pg_proc")
	})

	t.Run("trigger1_attached_to_documents_before_update", func(t *testing.T) {
		// pg_trigger.tgtype is a bitmask. For BEFORE UPDATE the relevant
		// bits are (TRIGGER_TYPE_BEFORE=2) | (TRIGGER_TYPE_UPDATE=16) = 18.
		// We also check the row-level bit (TRIGGER_TYPE_ROW=1) is set.
		var (
			exists  bool
			tgtype  int16
			tgname  string
		)
		err := pool.QueryRow(ctx, `
			SELECT TRUE, t.tgtype, t.tgname
			FROM pg_trigger t
			JOIN pg_class c ON c.oid = t.tgrelid
			JOIN pg_proc  p ON p.oid = t.tgfoid
			WHERE c.relname = 'documents'
			  AND p.proname = 'trg_prevent_schema_reassignment'
			  AND t.tgisinternal = FALSE
		`).Scan(&exists, &tgtype, &tgname)
		require.NoError(t, err,
			"trigger bound to documents and calling trg_prevent_schema_reassignment must exist")
		require.True(t, exists)
		// BEFORE bit = 2, ROW bit = 1, UPDATE bit = 16.
		require.NotZero(t, tgtype&int16(2),
			"trigger must fire BEFORE the row write (tgtype BEFORE bit)")
		require.NotZero(t, tgtype&int16(1),
			"trigger must be ROW-level (tgtype ROW bit)")
		require.NotZero(t, tgtype&int16(16),
			"trigger must fire on UPDATE (tgtype UPDATE bit)")
	})

	t.Run("trigger2_function_exists", func(t *testing.T) {
		var exists bool
		err := pool.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM pg_proc p
				JOIN pg_namespace n ON n.oid = p.pronamespace
				WHERE n.nspname = 'public'
				  AND p.proname = 'trg_prevent_legacy_extraction_on_custom_document'
			)
		`).Scan(&exists)
		require.NoError(t, err)
		require.True(t, exists,
			"trg_prevent_legacy_extraction_on_custom_document() function must exist")
	})

	t.Run("trigger2_attached_to_documentfieldextractions_before_insert", func(t *testing.T) {
		// Postgres folds unquoted identifiers to lowercase, so the real
		// stored relname is 'documentfieldextractions' (see plan migration
		// 112 — the CREATE TABLE is unquoted). Any new DDL that references
		// documents."documentFieldExtractions" would fail the same way the
		// §1.3 composite FK did with the "clientId" case.
		var (
			exists bool
			tgtype int16
		)
		err := pool.QueryRow(ctx, `
			SELECT TRUE, t.tgtype
			FROM pg_trigger t
			JOIN pg_class c ON c.oid = t.tgrelid
			JOIN pg_proc  p ON p.oid = t.tgfoid
			WHERE c.relname = 'documentfieldextractions'
			  AND p.proname = 'trg_prevent_legacy_extraction_on_custom_document'
			  AND t.tgisinternal = FALSE
		`).Scan(&exists, &tgtype)
		require.NoError(t, err,
			"trigger bound to documentfieldextractions must exist")
		require.True(t, exists)
		// BEFORE bit = 2, ROW bit = 1, INSERT bit = 4.
		require.NotZero(t, tgtype&int16(2),
			"trigger 2 must fire BEFORE the insert")
		require.NotZero(t, tgtype&int16(1),
			"trigger 2 must be ROW-level")
		require.NotZero(t, tgtype&int16(4),
			"trigger 2 must fire on INSERT")
	})

	t.Run("v3_removed_triggers_absent", func(t *testing.T) {
		// v4 intentionally does NOT have these two v3 triggers:
		//   - trg_validate_schema_client_match  (replaced by composite FK)
		//   - trg_enforce_consistent_schema_id  (column removed entirely)
		// If a migration accidentally re-introduces either, the test fails
		// with a pointer back to the plan decision.
		var count int
		err := pool.QueryRow(ctx, `
			SELECT COUNT(*) FROM pg_proc p
			WHERE p.proname IN (
				'trg_validate_schema_client_match',
				'trg_enforce_consistent_schema_id'
			)
		`).Scan(&count)
		require.NoError(t, err)
		require.Zero(t, count,
			"v4 intentionally omits trg_validate_schema_client_match and trg_enforce_consistent_schema_id — see plan Section 4.7 'Removed in v4' subsections")
	})
}

// TestMigration130_Trigger1_RejectsReassignWhenMetadataExists covers the
// §2.2 Trigger 1 behavioural case #1: document has custom metadata, attempt
// to change custom_schema_id → RAISE EXCEPTION "custom metadata already
// exists".
func TestMig130_Trg1_RejectsReassignWhenMetadata(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG1_META"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg1-meta",
		Clientid: clientID,
	})
	require.NoError(t, err)

	// Seed two schemas so we can attempt a real reassignment.
	schemaA := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg1-meta-A")
	schemaB := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg1-meta-B")

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg1-meta-doc",
	})
	require.NoError(t, err)

	// Bind document to schemaA.
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaA, docID)
	require.NoError(t, err)

	// Write a metadata row.
	_, err = pool.Exec(ctx, `
		INSERT INTO document_custom_metadata (document_id, metadata, version, created_by)
		VALUES ($1, '{"k":"v"}'::jsonb, 1, 'tester')
	`, docID)
	require.NoError(t, err)

	// Attempt to reassign to schemaB — trigger must block.
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaB, docID)
	require.Error(t, err, "reassignment must be rejected while metadata exists")
	require.Contains(t, err.Error(), "custom metadata already exists",
		"trigger 1 must raise with 'custom metadata already exists' message")
}

// TestMigration130_Trigger1_RejectsReassignWhenLegacyExtractionsExist covers
// §2.2 Trigger 1 behavioural case #2.
func TestMig130_Trg1_RejectsReassignWhenLegacy(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG1_LEGACY"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg1-legacy",
		Clientid: clientID,
	})
	require.NoError(t, err)

	schemaA := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg1-legacy-A")

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg1-legacy-doc",
	})
	require.NoError(t, err)

	// Insert a legacy extraction version row. The table is
	// documentFieldExtractionVersions per plan Section 4.7 Trigger 1.
	// Postgres stored the unquoted identifier as 'documentfieldextractionversions'.
	insertExtractionVersionForMilestone2(t, ctx, pool, docID)

	// Attempt to assign a schema — trigger must block because legacy
	// extractions already exist. Note: this is a NULL -> NOT-NULL change,
	// which Trigger 1 also blocks (OLD.custom_schema_id IS DISTINCT FROM
	// NEW.custom_schema_id evaluates true when OLD is NULL).
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaA, docID)
	require.Error(t, err, "assignment must be rejected while legacy extractions exist")
	require.Contains(t, err.Error(), "legacy field extractions already exist",
		"trigger 1 must raise with 'legacy field extractions already exist' message")
}

// TestMigration130_Trigger1_AllowsPreBindingChange covers §2.2 Trigger 1
// behavioural case #3: document with no metadata and no extractions can
// freely change custom_schema_id from one schema to another.
func TestMigration130_Trigger1_AllowsPreBindingChange(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG1_PRE"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg1-pre",
		Clientid: clientID,
	})
	require.NoError(t, err)

	schemaA := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg1-pre-A")
	schemaB := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg1-pre-B")

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg1-pre-doc",
	})
	require.NoError(t, err)

	// Bind to schemaA.
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaA, docID)
	require.NoError(t, err)

	// No metadata, no extractions. Rebinding to schemaB must succeed.
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaB, docID)
	require.NoError(t, err,
		"rebinding before any metadata or extractions exist must succeed")

	// Verify the DB state is now schemaB.
	var bound uuid.UUID
	err = pool.QueryRow(ctx,
		`SELECT custom_schema_id FROM documents WHERE id = $1`, docID).Scan(&bound)
	require.NoError(t, err)
	require.Equal(t, schemaB, bound)
}

// TestMigration130_Trigger2_RejectsLegacyInsertOnCustomDocument covers §2.2
// Trigger 2 behavioural case #1: document has custom_schema_id set, attempt
// direct INSERT into documentFieldExtractions → RAISE EXCEPTION with
// SQLSTATE 23514 (check_violation).
func TestMig130_Trg2_RejectsLegacyOnCustomDoc(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG2_CUSTOM"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg2-custom",
		Clientid: clientID,
	})
	require.NoError(t, err)

	schemaA := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg2-custom-A")

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg2-custom-doc",
	})
	require.NoError(t, err)

	// Bind document to schema.
	_, err = pool.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaA, docID)
	require.NoError(t, err)

	// Attempt to insert a legacy extraction — trigger 2 must block.
	// createdBy is required by the table (NOT NULL varchar(255)).
	_, err = pool.Exec(ctx,
		`INSERT INTO documentFieldExtractions (documentId, createdBy) VALUES ($1, 'tester')`,
		docID)
	require.Error(t, err, "legacy extraction insert must be rejected for custom-bound document")

	var pgErr *pgconn.PgError
	require.True(t, errors.As(err, &pgErr),
		"expected *pgconn.PgError, got %T: %v", err, err)
	require.Equal(t, "23514", pgErr.Code,
		"trigger 2 must raise with ERRCODE='check_violation' (SQLSTATE 23514)")
	require.Contains(t, err.Error(), "bound to custom schema",
		"trigger 2 error message must include 'bound to custom schema'")
}

// TestMigration130_Trigger2_AllowsLegacyInsertOnLegacyDocument covers §2.2
// Trigger 2 behavioural case #2 (no-op guard path): document has
// custom_schema_id NULL, legacy insert must succeed.
func TestMig130_Trg2_AllowsLegacyOnLegacyDoc(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG2_LEGACY"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg2-legacy",
		Clientid: clientID,
	})
	require.NoError(t, err)

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg2-legacy-doc",
	})
	require.NoError(t, err)

	// custom_schema_id stays NULL. Insert should succeed. createdBy is
	// required by the table (NOT NULL varchar(255)).
	_, err = pool.Exec(ctx,
		`INSERT INTO documentFieldExtractions (documentId, createdBy) VALUES ($1, 'tester')`,
		docID)
	require.NoError(t, err,
		"legacy extraction insert on NULL-custom-schema document must succeed")
}

// TestMigration130_Trigger2_ConcurrencyBlocksInsertDuringAssign covers §2.2
// Trigger 2 concurrency case:
//   - tx A takes SELECT documents FOR UPDATE on the target document, then
//     UPDATE documents SET custom_schema_id = S (the AssignSchema pattern).
//   - tx B attempts INSERT documentFieldExtractions on the same document.
//
// Expected: tx B's trigger function reads documents.custom_schema_id with
// FOR SHARE, which blocks until tx A commits. After tx A commits, tx B's
// trigger observes custom_schema_id IS NOT NULL and raises check_violation.
//
// This is the defense-in-depth proof for the race described in plan
// Section 7.3.
func TestMig130_Trg2_ConcurrencyBlocksInsert(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const clientID = "TRG2_RACE"
	resetClientForMilestone2Test(t, ctx, pool, clientID)
	err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{
		Name:     "trg2-race",
		Clientid: clientID,
	})
	require.NoError(t, err)

	schemaA := seedRawSchemaForMilestone2(t, ctx, pool, clientID, "trg2-race-A")

	docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{
		Clientid: clientID,
		Hash:     "trg2-race-doc",
	})
	require.NoError(t, err)

	// Begin two transactions on separate connections via pool.Begin —
	// pgxpool allocates a fresh connection per Begin, so the two tx's
	// do not share the same underlying session and the FOR SHARE / FOR
	// UPDATE interaction runs against real concurrency semantics.
	txA, err := pool.Begin(ctx)
	require.NoError(t, err)
	defer func() { _ = txA.Rollback(ctx) }()

	var lockedID uuid.UUID
	err = txA.QueryRow(ctx,
		`SELECT id FROM documents WHERE id = $1 FOR UPDATE`, docID).Scan(&lockedID)
	require.NoError(t, err)

	// Do the UPDATE inside tx A but DON'T commit yet.
	_, err = txA.Exec(ctx,
		`UPDATE documents SET custom_schema_id = $1 WHERE id = $2`, schemaA, docID)
	require.NoError(t, err)

	// Fire tx B on a separate goroutine. Its INSERT trigger will take
	// FOR SHARE on the documents row and block until tx A commits.
	var wg sync.WaitGroup
	insertErrCh := make(chan error, 1)
	wg.Add(1)
	go func() {
		defer wg.Done()
		// Use a fresh context with a generous timeout so this goroutine
		// cannot hang the test forever if the trigger doesn't block.
		gctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		txB, err := pool.Begin(gctx)
		if err != nil {
			insertErrCh <- err
			return
		}
		defer func() { _ = txB.Rollback(gctx) }()
		_, err = txB.Exec(gctx,
			`INSERT INTO documentFieldExtractions (documentId, createdBy) VALUES ($1, 'tester')`,
			docID)
		if err != nil {
			insertErrCh <- err
			return
		}
		// If the INSERT itself succeeded, the test will fail on the
		// downstream `require.Error` below. But if it did succeed, we
		// still need to roll back so we don't leak a commit.
		insertErrCh <- nil
	}()

	// Give the goroutine a moment to actually block on the FOR SHARE lock
	// before we commit tx A. 250ms is generous — a normal lock acquisition
	// is microseconds, but we want to see the trigger's FOR SHARE enter
	// the wait state rather than race us to an error reply.
	time.Sleep(250 * time.Millisecond)

	// Before committing, verify the INSERT has NOT yet completed — if it
	// has, the trigger's FOR SHARE lock is missing.
	select {
	case err := <-insertErrCh:
		t.Fatalf("tx B's INSERT returned too early (before tx A committed): %v", err)
	default:
	}

	// Commit tx A — this releases the FOR UPDATE lock and lets tx B's
	// trigger proceed with its FOR SHARE read, see custom_schema_id set,
	// and raise.
	require.NoError(t, txA.Commit(ctx))

	// Wait for tx B to finish and verify it reported the check_violation.
	wg.Wait()
	insertErr := <-insertErrCh
	require.Error(t, insertErr,
		"tx B's INSERT must be rejected after tx A committed the schema assign")

	var pgErr *pgconn.PgError
	require.True(t, errors.As(insertErr, &pgErr),
		"expected *pgconn.PgError, got %T: %v", insertErr, insertErr)
	require.Equal(t, "23514", pgErr.Code,
		"expected check_violation SQLSTATE 23514 from trigger 2")
}

// TestMigration130_UpDownRoundtrip proves the migration 130 down.sql is
// complete: triggers and functions are all removed on down, re-added on up.
func TestMigration130_UpDownRoundtrip(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.SkipNow()
	}
	ctx := t.Context()

	cfg := &serviceconfig.BaseConfig{}
	test.CreateDB(t, cfg)
	pool := cfg.GetDBPool()
	require.NotNil(t, pool)

	const (
		upPath   = "../migrations/00000000000130_add_schema_invariant_triggers.up.sql"
		downPath = "../migrations/00000000000130_add_schema_invariant_triggers.down.sql"
	)
	upSQL, err := os.ReadFile(upPath)
	require.NoError(t, err, "must be able to read migration 130 up.sql")
	downSQL, err := os.ReadFile(downPath)
	require.NoError(t, err, "must be able to read migration 130 down.sql")

	// Leave DB in migrated state on test exit.
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), string(upSQL))
	})

	// 1. Down.
	_, err = pool.Exec(ctx, string(downSQL))
	require.NoError(t, err, "down migration 130 must apply cleanly")

	for _, proname := range []string{
		"trg_prevent_schema_reassignment",
		"trg_prevent_legacy_extraction_on_custom_document",
	} {
		var exists bool
		err = pool.QueryRow(ctx, `
			SELECT EXISTS (SELECT 1 FROM pg_proc WHERE proname = $1)
		`, proname).Scan(&exists)
		require.NoError(t, err)
		require.Falsef(t, exists,
			"%s must be gone after down migration 130", proname)
	}

	// 2. Back up.
	_, err = pool.Exec(ctx, string(upSQL))
	require.NoError(t, err, "up migration 130 must re-apply cleanly")

	for _, proname := range []string{
		"trg_prevent_schema_reassignment",
		"trg_prevent_legacy_extraction_on_custom_document",
	} {
		var exists bool
		err = pool.QueryRow(ctx, `
			SELECT EXISTS (SELECT 1 FROM pg_proc WHERE proname = $1)
		`, proname).Scan(&exists)
		require.NoError(t, err)
		require.Truef(t, exists,
			"%s must exist again after re-running up migration 130", proname)
	}
}

// ---------- helpers ----------

// resetClientForMilestone2Test wipes every row owned by clientID in the
// order required by the FK graph so the test starts from a clean slate
// even when the testcontainer Postgres volume is reused across runs.
func resetClientForMilestone2Test(t testing.TB, ctx context.Context, pool queryable, clientID string) {
	t.Helper()
	// queryable does not expose Exec, so grab it via type assertion. The
	// real *pgxpool.Pool implements Exec; we assert to that shape here.
	type execer interface {
		Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	}
	pe, ok := pool.(execer)
	require.True(t, ok, "pool does not implement Exec")

	stmts := []string{
		`DELETE FROM document_custom_metadata
		  WHERE document_id IN (SELECT id FROM documents WHERE clientId = $1)`,
		`DELETE FROM documentFieldExtractionVersions
		  WHERE documentId IN (SELECT id FROM documents WHERE clientId = $1)`,
		`DELETE FROM documentFieldExtractions
		  WHERE documentId IN (SELECT id FROM documents WHERE clientId = $1)`,
		`UPDATE documents SET custom_schema_id = NULL WHERE clientId = $1`,
		`DELETE FROM client_metadata_schemas WHERE client_id = $1`,
		`DELETE FROM documentEntries
		  WHERE documentId IN (SELECT id FROM documents WHERE clientId = $1)`,
		`DELETE FROM documents WHERE clientId = $1`,
		`DELETE FROM folders WHERE clientId = $1`,
		`DELETE FROM clientCanSync WHERE clientId = $1`,
		`DELETE FROM clients WHERE clientId = $1`,
	}
	for _, stmt := range stmts {
		_, err := pe.Exec(ctx, stmt, clientID)
		// Some DELETEs may fail if the table hasn't been created yet
		// (e.g., document_custom_metadata before migration 129 lands).
		// The failing tests below are the ones that need the table to
		// exist; those tests will fail on a later assertion rather than
		// here. Swallow "relation does not exist" errors so the reset
		// never masks the real reason a test fails.
		if err != nil && !strings.Contains(err.Error(), "does not exist") {
			t.Fatalf("reset stmt failed: %s\nerr: %v", stmt, err)
		}
	}
}

// seedRawSchemaForMilestone2 inserts a client_metadata_schemas row directly
// (skipping the customschema.Service path so this test has no dependency on
// Milestone 1 wiring). Returns the new schema row ID. Each call gets a
// unique `name` so repeated invocations inside one test don't trip
// uq_client_schema_name_version.
func seedRawSchemaForMilestone2(t testing.TB, ctx context.Context, pool queryable, clientID, name string) uuid.UUID {
	t.Helper()
	type execer interface {
		QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	}
	pe, ok := pool.(execer)
	require.True(t, ok, "pool does not implement QueryRow")

	const sql = `
		INSERT INTO client_metadata_schemas
			(client_id, name, schema_def, version, created_by)
		VALUES ($1, $2, $3::jsonb, 1, 'tester')
		RETURNING id
	`
	var id uuid.UUID
	err := pe.QueryRow(ctx, sql, clientID, name, `{"type":"object","additionalProperties":false}`).Scan(&id)
	require.NoError(t, err)
	return id
}

// insertExtractionVersionForMilestone2 inserts a legacy field extraction row
// and its matching version row for the given document. Trigger 1's "legacy
// field extractions already exist" branch checks documentFieldExtractionVersions
// (which is what `HasFieldExtraction` reuses in the service layer), so we must
// land both the parent documentFieldExtractions row (required by FK) and the
// version row the trigger actually looks for.
//
// The minimum column set for each table is exercised here so the helper works
// against the real DB without touching sqlc. documentFieldExtractions has
// `createdBy varchar(255) NOT NULL`, and documentFieldExtractionVersions has
// `fieldExtractionId uuid NOT NULL` plus `version bigint NOT NULL` plus
// `createdBy varchar(255) NOT NULL` (see migrations 112/113/117).
//
// **Important**: this helper is only ever called BEFORE the document has a
// custom_schema_id bound. Trigger 2 (created in migration 130) would block
// the documentFieldExtractions INSERT otherwise.
func insertExtractionVersionForMilestone2(t testing.TB, ctx context.Context, pool queryable, docID uuid.UUID) {
	t.Helper()
	type execer interface {
		Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
		QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	}
	pe, ok := pool.(execer)
	require.True(t, ok)

	var feID uuid.UUID
	err := pe.QueryRow(ctx,
		`INSERT INTO documentFieldExtractions (documentId, createdBy) VALUES ($1, $2) RETURNING id`,
		docID, "tester").Scan(&feID)
	require.NoError(t, err, "seed a legacy field extractions row")

	_, err = pe.Exec(ctx,
		`INSERT INTO documentFieldExtractionVersions (fieldExtractionId, documentId, version, createdBy)
		 VALUES ($1, $2, $3, $4)`,
		feID, docID, 1, "tester")
	require.NoError(t, err, "seed a legacy extraction version row")
}

// fetchDCMConstraintColumns returns the column list for a constraint on
// document_custom_metadata. If conname is empty, it takes the first match
// of the requested contype (used for the PRIMARY KEY lookup where the
// generated pk_ name is not worth hard-coding).
func fetchDCMConstraintColumns(ctx context.Context, q queryable, wantContype, conname string) ([]string, error) {
	var rows pgx.Rows
	var err error
	if conname == "" {
		rows, err = q.Query(ctx, `
			SELECT a.attname
			FROM pg_constraint c
			JOIN pg_class     t ON t.oid = c.conrelid
			JOIN unnest(c.conkey) WITH ORDINALITY AS k(attnum, ord) ON TRUE
			JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum
			WHERE t.relname = 'document_custom_metadata'
			  AND c.contype = $1
			ORDER BY k.ord
		`, wantContype)
	} else {
		rows, err = q.Query(ctx, `
			SELECT a.attname
			FROM pg_constraint c
			JOIN pg_class     t ON t.oid = c.conrelid
			JOIN unnest(c.conkey) WITH ORDINALITY AS k(attnum, ord) ON TRUE
			JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum
			WHERE c.conname = $1
			  AND t.relname = 'document_custom_metadata'
			  AND c.contype = $2
			ORDER BY k.ord
		`, conname, wantContype)
	}
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var cols []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		cols = append(cols, name)
	}
	return cols, rows.Err()
}