package integration

// 000107 and 000108 proved against a real PostgreSQL with the data they exist
// to repair.
//
// WHY THIS FILE EXISTS. Both backfills are data repairs, and the only thing
// that can prove a data repair works is running it on data. CI applies
// migrations to a FRESH, EMPTY database, so every INSERT in both files is
// suppressed by its own HAVING count(*) > 0 and not one row is written. Their
// census SELECTs do execute, which is how a missing FROM was caught - but the
// write path, the part that lifts a counter above real rows on a database that
// has been used, had no execution coverage at all until this file.
//
// The ordered reproduction is the whole point and it is faithful to the
// production ordering: build the pre-000107 state, insert fixtures, then apply
// 000107 and 000108 over that data, in that order.
//
// SCOPE, stated honestly. This is not a replay of 000001 through 000108. It is
// a scratch schema holding exactly the three tables the two migrations name -
// warehouses, payrolls, document_number_sequences - which is their complete
// dependency surface, verified by reading the files. Everything the migrations
// read or write is present, so every statement in both files executes for
// real; nothing else is stubbed. Replaying 108 files would add dependencies and
// buy no additional coverage of these two.
//
// The migration SQL is read from the EMBEDDED FS (migrations.FS), not copied
// into this file, so the test exercises the bytes that will ship. A hand-copied
// string would keep passing after someone edited the migration, which is the
// exact failure this file exists to prevent. This follows the convention
// internal/integration/authz_migration_test.go established for the same reason.

import (
	"context"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/TaruLab/keluargamart-backend/migrations"
	"github.com/TaruLab/keluargamart-backend/pkg/seq"
)

// The dependency surface of 000107 + 000108, and nothing else. The DDL is the
// relevant columns of the real tables (000001_init_schema.sql), with the NOT
// NULL and UNIQUE constraints kept where they are load-bearing for the
// fixtures below.
const (
	seqTableDDL = `CREATE TABLE document_number_sequences (
    prefix  TEXT NOT NULL,
    doc_date DATE NOT NULL,
    counter BIGINT NOT NULL DEFAULT 0,
    PRIMARY KEY (prefix, doc_date)
)`

	warehousesDDL = `CREATE TABLE warehouses (
    id         BIGSERIAL PRIMARY KEY,
    code       VARCHAR(20) NOT NULL UNIQUE,
    name       VARCHAR(255) NOT NULL,
    is_active  BOOLEAN NOT NULL DEFAULT true,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`

	payrollsDDL = `CREATE TABLE payrolls (
    id             BIGSERIAL PRIMARY KEY,
    payroll_number VARCHAR(50) NOT NULL UNIQUE,
    employee_id    BIGINT NOT NULL,
    period_month   SMALLINT NOT NULL,
    period_year    SMALLINT NOT NULL,
    net_salary     NUMERIC(15,2) NOT NULL DEFAULT 0,
    status         TEXT NOT NULL DEFAULT 'draft',
    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT check_payrolls_employee_period UNIQUE (employee_id, period_month, period_year)
)`
)

const (
	warehouseMigration = "000107_warehouse_code_sequence_anchor.sql"
	payrollMigration   = "000108_payroll_number_sequence_month_buckets.sql"
)

// migrationSQL reads a migration from the embedded FS, so the bytes under test
// are the bytes that ship.
func readMigration(t *testing.T, name string) string {
	t.Helper()
	b, err := migrations.FS.ReadFile(name)
	if err != nil {
		t.Fatalf("read %s from the embedded FS: %v", name, err)
	}
	return string(b)
}

// scratchSeqSchema creates an isolated schema holding only the two migrations'
// dependency surface, and returns a pool bound to it.
//
// Same pattern as authz_migration_test.go's scratchSchema: DROP first, because a
// run that died before cleanup leaves the schema behind and CREATE would fail
// on the name; and CASCADE, because the tables below are recreated each time.
// Cleanup drops it, so a developer running the suite locally is not left with
// debris, and the developer's own database is never touched.
func scratchSeqSchema(t *testing.T) *pgxpool.Pool {
	t.Helper()
	databaseURL := os.Getenv("DATABASE_URL")
	if databaseURL == "" {
		t.Skip("DATABASE_URL is not set")
	}
	ctx := context.Background()

	admin, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	schema := fmt.Sprintf("seqanchor_%d", time.Now().UnixNano())
	if _, err := admin.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE"); err != nil {
		admin.Close()
		t.Fatal(err)
	}
	if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil {
		admin.Close()
		t.Fatal(err)
	}

	cfg, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		admin.Close()
		t.Fatal(err)
	}
	cfg.ConnConfig.RuntimeParams["search_path"] = schema
	cfg.MaxConns = 2
	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		admin.Close()
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if _, err := admin.Exec(context.Background(), "DROP SCHEMA IF EXISTS "+schema+" CASCADE"); err != nil {
			t.Logf("drop schema %s: %v", schema, err)
		}
		admin.Close()
	})

	for _, ddl := range []string{seqTableDDL, warehousesDDL, payrollsDDL} {
		if _, err := pool.Exec(ctx, ddl); err != nil {
			t.Fatalf("create scratch table: %v", err)
		}
	}
	return pool
}

// applyMigration runs one migration the way internal/dbmigrate does: the whole
// file as a single Exec inside a transaction. Every statement in the file runs,
// including the census SELECTs, so a syntax error or a missing FROM anywhere in
// it fails here exactly as it failed in CI.
func applyMigration(t *testing.T, pool *pgxpool.Pool, name, sql string) {
	t.Helper()
	tx, err := pool.Begin(context.Background())
	if err != nil {
		t.Fatalf("begin for %s: %v", name, err)
	}
	defer func() { _ = tx.Rollback(context.Background()) }()
	if _, err := tx.Exec(context.Background(), sql); err != nil {
		t.Fatalf("apply %s: %v", name, err)
	}
	if err := tx.Commit(context.Background()); err != nil {
		t.Fatalf("commit %s: %v", name, err)
	}
}

// seqCounter reads one bucket, returning ok=false when it does not exist.
func seqCounter(t *testing.T, pool *pgxpool.Pool, prefix string, docDate string) (int64, bool) {
	t.Helper()
	var n int64
	err := pool.QueryRow(context.Background(),
		`SELECT counter FROM document_number_sequences WHERE prefix = $1 AND doc_date = $2::date`,
		prefix, docDate).Scan(&n)
	if err != nil {
		return 0, false
	}
	return n, true
}

func warehouseCodes(t *testing.T, pool *pgxpool.Pool) []string {
	t.Helper()
	rows, err := pool.Query(context.Background(), `SELECT code FROM warehouses ORDER BY code`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var c string
		if err := rows.Scan(&c); err != nil {
			t.Fatal(err)
		}
		out = append(out, c)
	}
	return out
}

func payrollNumbers(t *testing.T, pool *pgxpool.Pool) []string {
	t.Helper()
	rows, err := pool.Query(context.Background(), `SELECT payroll_number FROM payrolls ORDER BY payroll_number`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var c string
		if err := rows.Scan(&c); err != nil {
			t.Fatal(err)
		}
		out = append(out, c)
	}
	return out
}

func insertWarehouse(t *testing.T, pool *pgxpool.Pool, code string) {
	t.Helper()
	if _, err := pool.Exec(context.Background(),
		`INSERT INTO warehouses (code, name) VALUES ($1, $2)`, code, "Fixture "+code); err != nil {
		t.Fatalf("seed warehouse %s: %v", code, err)
	}
}

func insertPayroll(t *testing.T, pool *pgxpool.Pool, number string, employeeID, month, year int) {
	t.Helper()
	if _, err := pool.Exec(context.Background(),
		`INSERT INTO payrolls (payroll_number, employee_id, period_month, period_year)
		 VALUES ($1, $2, $3, $4)`, number, employeeID, month, year); err != nil {
		t.Fatalf("seed payroll %s: %v", number, err)
	}
}

func seedBucket(t *testing.T, pool *pgxpool.Pool, prefix, docDate string, counter int64) {
	t.Helper()
	if _, err := pool.Exec(context.Background(),
		`INSERT INTO document_number_sequences (prefix, doc_date, counter) VALUES ($1, $2::date, $3)`,
		prefix, docDate, counter); err != nil {
		t.Fatalf("seed bucket (%s,%s): %v", prefix, docDate, err)
	}
}

// highestOrdinal is the migration's own rule, restated: the max ordinal in a
// GDG- code, or in a PAY- number for one month. Used to assert the property
// 000107 section 9 and 000108 section 8 report as `ok`, which is exactly
// "counter >= highest ordinal per bucket".
func highestGDGOrdinal(t *testing.T, pool *pgxpool.Pool) int64 {
	t.Helper()
	var n int64
	if err := pool.QueryRow(context.Background(),
		`SELECT coalesce(max(split_part(code, '-', 2)::bigint), 0) FROM warehouses
		 WHERE code ~ '^GDG-[0-9]+$'`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func highestPayOrdinal(t *testing.T, pool *pgxpool.Pool, yyyymm string) int64 {
	t.Helper()
	var n int64
	if err := pool.QueryRow(context.Background(),
		`SELECT coalesce(max(split_part(payroll_number, '-', 3)::bigint), 0) FROM payrolls
		 WHERE payroll_number ~ '^PAY-[0-9]{6}-[0-9]+$' AND split_part(payroll_number, '-', 2) = $1`,
		yyyymm).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

// TestGDGBackfill_LiftsCounterAboveRealData is the case CI cannot reach: the
// anchor bucket does not exist and there are real GDG- codes, so the INSERT is
// not suppressed and the write path runs for the first time.
func TestGDGBackfill_LiftsCounterAboveRealData(t *testing.T) {
	pool := scratchSeqSchema(t)
	ctx := context.Background()

	// Allocator-format codes, which is what a used database looks like.
	insertWarehouse(t, pool, "GDG-001")
	insertWarehouse(t, pool, "GDG-002")
	// A hand-entered code in a different format. The migration's regex guard
	// must ignore it: it must not drag the counter, and it must not be touched.
	insertWarehouse(t, pool, "WH1")
	// A DUPLICATE ORDINAL. warehouses.code is UNIQUE, so two rows cannot share a
	// string, but 'GDG-1' and 'GDG-001' are different strings that split_part
	// maps to the SAME ordinal. This is what 000107 section 4 looks for, and it
	// is reachable through a hand-entered code - it is not dead census output.
	insertWarehouse(t, pool, "GDG-1")
	// A per-date bucket left behind by the old daily allocator. The header
	// promises these are NOT deleted.
	seedBucket(t, pool, "GDG", "2026-03-10", 3)

	before := warehouseCodes(t, pool)
	applyMigration(t, pool, warehouseMigration, readMigration(t, warehouseMigration))

	// THE LOAD-BEARING ASSERTION: the anchor bucket exists, and it is at or
	// above the highest GDG- ordinal in use. With GDG-001, GDG-002 and GDG-1 the
	// highest ordinal is 2.
	got, ok := seqCounter(t, pool, "GDG", "2000-01-01")
	if !ok {
		t.Fatal("no anchor bucket at 2000-01-01: the backfill wrote nothing on populated data")
	}
	if highest := highestGDGOrdinal(t, pool); got < highest {
		t.Errorf("anchor counter = %d, below the highest GDG- ordinal %d: the next create collides", got, highest)
	}
	if got != 2 {
		t.Errorf("anchor counter = %d, want 2 (the max ordinal across GDG-001, GDG-002, GDG-1)", got)
	}

	// The non-matching code did not drag the counter. WH1 and GDG-1 are both
	// outside the plain 1..N range the counter must land on; if either had been
	// counted, the number would be different.
	if got != 2 {
		t.Errorf("anchor counter = %d, want 2: the non-GDG- code and the duplicate ordinal must not inflate it", got)
	}

	// The old per-date bucket is untouched: the header says it is not deleted.
	if n, ok := seqCounter(t, pool, "GDG", "2026-03-10"); !ok || n != 3 {
		t.Errorf("per-date bucket (GDG, 2026-03-10) = %d ok=%v, want it preserved at 3", n, ok)
	}

	// Nothing is renumbered, deleted or disabled. Both files promise this.
	after := warehouseCodes(t, pool)
	if strings.Join(before, ",") != strings.Join(after, ",") {
		t.Errorf("warehouse codes changed: before %v, after %v", before, after)
	}

	// The next allocation must clear every existing code.
	next, err := seq.Next(ctx, pool, "GDG", time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC))
	if err != nil {
		t.Fatal(err)
	}
	if code := fmt.Sprintf("GDG-%03d", next); code == "GDG-001" || code == "GDG-002" || code == "GDG-1" {
		t.Errorf("next allocation would be %s, which already exists", code)
	}
	_ = ctx
}

// TestGDGBackfill_NeverLowersAPreExistingBucket: GREATEST is the entire safety
// property of both files. A counter already ahead of the data must survive.
func TestGDGBackfill_NeverLowersAPreExistingBucket(t *testing.T) {
	pool := scratchSeqSchema(t)
	insertWarehouse(t, pool, "GDG-001")
	insertWarehouse(t, pool, "GDG-002")
	// Deliberately AHEAD of the data: the backfill would compute 2 here.
	seedBucket(t, pool, "GDG", "2000-01-01", 50)

	applyMigration(t, pool, warehouseMigration, readMigration(t, warehouseMigration))

	got, ok := seqCounter(t, pool, "GDG", "2000-01-01")
	if !ok {
		t.Fatal("anchor bucket disappeared")
	}
	if got != 50 {
		t.Errorf("anchor counter = %d, want 50: a pre-existing counter ahead of the data must never be lowered", got)
	}
}

// TestPayBackfill_PerMonthBuckets is the case that separates 000108 from 000107
// and 000106. There is no single counter: every month holding a PAY- number has
// its own bucket, each lifted to ITS OWN maximum.
//
// Seeding only the current month would leave 2026-07 at nothing, and the next
// July run would start at 1 and collide. The earlier month keeping its own value
// is the assertion that catches that mistake.
func TestPayBackfill_PerMonthBuckets(t *testing.T) {
	pool := scratchSeqSchema(t)

	// An earlier period, three payslips.
	insertPayroll(t, pool, "PAY-202607-0001", 1, 7, 2026)
	insertPayroll(t, pool, "PAY-202607-0002", 2, 7, 2026)
	insertPayroll(t, pool, "PAY-202607-0003", 3, 7, 2026)
	// A later period, two payslips, plus a duplicate ordinal ('PAY-202609-2').
	insertPayroll(t, pool, "PAY-202609-0001", 1, 9, 2026)
	insertPayroll(t, pool, "PAY-202609-0002", 2, 9, 2026)
	insertPayroll(t, pool, "PAY-202609-2", 3, 9, 2026)
	// A NUMBER whose month disagrees with its period columns: the number says
	// August, the row is the September period. 000108 section 3 prints exactly
	// this and the header promises it is reported, not corrected.
	insertPayroll(t, pool, "PAY-202608-0001", 4, 9, 2026)
	// A number in a format the allocator never produces. The regex guard must
	// ignore it, and it must not be touched.
	insertPayroll(t, pool, "PR-2026-06-EMP-002", 5, 6, 2026)

	before := payrollNumbers(t, pool)
	applyMigration(t, pool, payrollMigration, readMigration(t, payrollMigration))

	// EACH month lands on ITS OWN maximum. A global counter would put every
	// bucket at 3; a current-month-only seed would leave 2026-07 absent.
	for _, tc := range []struct {
		yyyymm string
		want   int64
	}{
		{"202607", 3},
		{"202609", 2},
		{"202608", 1},
	} {
		got, ok := seqCounter(t, pool, "pay", tc.yyyymm[:4]+"-"+tc.yyyymm[4:6]+"-01")
		if !ok {
			t.Errorf("no bucket for %s: the backfill only seeded some periods", tc.yyyymm)
			continue
		}
		if got != tc.want {
			t.Errorf("bucket %s = %d, want %d (that month's own maximum)", tc.yyyymm, got, got)
		}
	}

	// The earlier month is the load-bearing one: it must be lifted on its own,
	// not because the current month happened to be seeded.
	earlier, ok := seqCounter(t, pool, "pay", "2026-07-01")
	if !ok {
		t.Fatal("2026-07 bucket missing: a migration that seeded only the current month would look identical until the next July")
	}
	if earlier < highestPayOrdinal(t, pool, "202607") {
		t.Errorf("2026-07 counter = %d, below its own maximum %d", earlier, highestPayOrdinal(t, pool, "202607"))
	}

	// The non-matching format seeded no bucket, and nothing was renumbered.
	if n, ok := seqCounter(t, pool, "pay", "2026-06-01"); ok {
		t.Errorf("a bucket was created for 2026-06 from a PR- number: %d", n)
	}
	after := payrollNumbers(t, pool)
	if strings.Join(before, ",") != strings.Join(after, ",") {
		t.Errorf("payroll numbers changed: before %v, after %v", before, after)
	}
	// The month-disagreement row is still exactly as it was, uncorrected.
	var month, year int
	if err := pool.QueryRow(context.Background(),
		`SELECT period_month, period_year FROM payrolls WHERE payroll_number = 'PAY-202608-0001'`).
		Scan(&month, &year); err != nil {
		t.Fatal(err)
	}
	if month != 9 || year != 2026 {
		t.Errorf("the disagreeing payslip was corrected to %d-%d, want 9-2026 unchanged", year, month)
	}
}

// TestBackfill_Section8VerdictIsOK asserts the property 000108 section 8 and
// 000107 section 9 report as `ok`: for every bucket, the counter is at or above
// the highest ordinal issued for that bucket's period. The verdict is a pure
// function of that comparison, so asserting the comparison IS asserting `ok`,
// without this file having to parse the migration's SQL to find the query.
func TestBackfill_Section8VerdictIsOK(t *testing.T) {
	pool := scratchSeqSchema(t)
	insertWarehouse(t, pool, "GDG-001")
	insertWarehouse(t, pool, "GDG-009")
	insertPayroll(t, pool, "PAY-202605-0001", 1, 5, 2026)
	insertPayroll(t, pool, "PAY-202605-0004", 2, 5, 2026)
	seedBucket(t, pool, "GDG", "2000-01-01", 100) // already ahead

	applyMigration(t, pool, warehouseMigration, readMigration(t, warehouseMigration))
	applyMigration(t, pool, payrollMigration, readMigration(t, payrollMigration))

	rows, err := pool.Query(context.Background(),
		`SELECT s.doc_date, s.counter,
		        coalesce(max(split_part(w.code, '-', 2)::bigint), 0)
		 FROM document_number_sequences s
		 LEFT JOIN warehouses w
		   ON s.prefix = 'GDG' AND w.code ~ '^GDG-[0-9]+$'
		 WHERE s.prefix = 'GDG'
		 GROUP BY s.doc_date, s.counter`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	checked := 0
	for rows.Next() {
		var d time.Time
		var counter, highest int64
		if err := rows.Scan(&d, &counter, &highest); err != nil {
			t.Fatal(err)
		}
		checked++
		if counter < highest {
			t.Errorf("bucket %s would report STILL BELOW DATA: counter %d < highest %d",
				d.Format("2006-01-02"), counter, highest)
		}
	}
	if checked == 0 {
		t.Fatal("no GDG bucket to check")
	}

	// Same property for every pay bucket, each against ITS OWN month.
	prows, err := pool.Query(context.Background(),
		`SELECT s.doc_date, s.counter,
		        coalesce((SELECT max(split_part(p.payroll_number, '-', 3)::bigint)
		                    FROM payrolls p
		                   WHERE p.payroll_number ~ '^PAY-[0-9]{6}-[0-9]+$'
		                     AND to_date(split_part(p.payroll_number, '-', 2), 'YYYYMM') = s.doc_date), 0)
		 FROM document_number_sequences s
		 WHERE s.prefix = 'pay'
		 ORDER BY s.doc_date`)
	if err != nil {
		t.Fatal(err)
	}
	defer prows.Close()
	pchecked := 0
	for prows.Next() {
		var d time.Time
		var counter, highest int64
		if err := prows.Scan(&d, &counter, &highest); err != nil {
			t.Fatal(err)
		}
		pchecked++
		if counter < highest {
			t.Errorf("pay bucket %s would report STILL BELOW DATA: counter %d < highest %d",
				d.Format("2006-01-02"), counter, highest)
		}
	}
	if pchecked != 1 {
		t.Errorf("checked %d pay buckets, want 1", pchecked)
	}
}

// TestBackfill_EmptyDatabaseInsertsNothingAndFirstAllocationIsOne: the fresh
// case CI has always exercised, now asserted on outcomes rather than assumed.
// Nothing to lift, so nothing is written, and the live allocator opens each
// bucket at 1.
func TestBackfill_EmptyDatabaseInsertsNothingAndFirstAllocationIsOne(t *testing.T) {
	pool := scratchSeqSchema(t)
	ctx := context.Background()

	applyMigration(t, pool, warehouseMigration, readMigration(t, warehouseMigration))
	applyMigration(t, pool, payrollMigration, readMigration(t, payrollMigration))

	if n, ok := seqCounter(t, pool, "GDG", "2000-01-01"); ok {
		t.Errorf("a GDG anchor bucket was written on an empty database: %d", n)
	}
	for _, d := range []string{"2000-01-01", "2026-09-01"} {
		if n, ok := seqCounter(t, pool, "pay", d); ok {
			t.Errorf("a pay bucket was written for %s on an empty database: %d", d, n)
		}
	}

	// The real allocator, against the post-migration state.
	gdg, err := seq.Next(ctx, pool, "GDG", time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC))
	if err != nil {
		t.Fatal(err)
	}
	if code := fmt.Sprintf("GDG-%03d", gdg); code != "GDG-001" {
		t.Errorf("first warehouse code = %s, want GDG-001", code)
	}
	pay, err := seq.Next(ctx, pool, "pay", time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC))
	if err != nil {
		t.Fatal(err)
	}
	if number := fmt.Sprintf("PAY-%04d-%04d", 202609, pay); number != "PAY-202609-0001" {
		t.Errorf("first payroll number = %s, want PAY-202609-0001", number)
	}
}

// TestPayBackfill_SeedingOnlyTheLatestMonthIsCaught is the second non-vacuity
// proof, for the property that separates 000108 from the other two backfills.
//
// 000108's whole difficulty is that there is no single counter: every month
// holding a PAY- number has its own bucket. A migration that looked correct and
// passed CI could seed only the current month, because on an empty database
// every variant writes nothing. This runs such a variant - the real 000108 with
// its GROUP BY restricted to the latest yyyymm - and asserts that the earlier
// month's bucket is then MISSING and below its own maximum, which is exactly
// what TestPayBackfill_PerMonthBuckets forbids.
//
// migrations/ is never touched; the variant is a string transform of the
// embedded file, so it cannot drift from what ships.
func TestPayBackfill_SeedingOnlyTheLatestMonthIsCaught(t *testing.T) {
	real := readMigration(t, payrollMigration)
	const groupBy = "GROUP BY p.yyyymm\nHAVING count(*) > 0"
	if !strings.Contains(real, groupBy) {
		t.Fatalf("000108 no longer contains %q, so this non-vacuity proof no longer applies; update it", groupBy)
	}
	latestOnly := `GROUP BY p.yyyymm
HAVING count(*) > 0
  AND p.yyyymm = (SELECT max(split_part(payroll_number, '-', 2)) FROM payrolls
                   WHERE payroll_number ~ '^PAY-[0-9]{6}-[0-9]+$')`
	broken := strings.Replace(real, groupBy, latestOnly, 1)
	if broken == real {
		t.Fatal("the transform did not change the SQL")
	}

	pool := scratchSeqSchema(t)
	insertPayroll(t, pool, "PAY-202607-0001", 1, 7, 2026)
	insertPayroll(t, pool, "PAY-202607-0002", 2, 7, 2026)
	insertPayroll(t, pool, "PAY-202607-0003", 3, 7, 2026)
	insertPayroll(t, pool, "PAY-202609-0001", 4, 9, 2026)

	applyMigration(t, pool, payrollMigration+" (latest-month-only, scratch copy only)", broken)

	// The current month is seeded, so the migration "works" and CI would be
	// satisfied.
	if n, ok := seqCounter(t, pool, "pay", "2026-09-01"); !ok || n != 1 {
		t.Errorf("latest month bucket = %d ok=%v, want 1: the variant should still seed the current month", n, ok)
	}
	// The earlier month is not, and that is what the next July run would
	// collide against: its counter would reopen at 1 over PAY-202607-0001.
	earlier, ok := seqCounter(t, pool, "pay", "2026-07-01")
	if ok && earlier >= highestPayOrdinal(t, pool, "202607") {
		t.Fatalf("the latest-month-only variant lifted 2026-07 to %d, so this test is not exercising anything", earlier)
	}
	t.Logf("latest-month-only variant left 2026-07 at %d (ok=%v) against a highest ordinal of %d; "+
		"TestPayBackfill_PerMonthBuckets requires it lifted, so it would fail here",
		earlier, ok, highestPayOrdinal(t, pool, "202607"))
}

// TestBackfill_GreatestIsLoadBearing is the non-vacuity proof, and it is
// deliberately NOT a test of the migration: it runs a COPY of 000107 with
// GREATEST relaxed to a plain assignment, against a counter seeded ahead of the
// data, and asserts the counter comes back LOWERED.
//
// migrations/ is never touched. The copy is a string transform of the real
// embedded file, so if the migration ever changes shape this test fails loudly
// rather than quietly testing a stale copy.
//
// The point: TestGDGBackfill_NeverLowersAPreExistingBucket asserts the counter
// survives at 50. This test shows that with GREATEST removed it would be 2, so
// that assertion is a real check on a real property and not a tautology.
func TestBackfill_GreatestIsLoadBearing(t *testing.T) {
	real := readMigration(t, warehouseMigration)
	const greatest = `GREATEST(document_number_sequences.counter, EXCLUDED.counter)`
	if !strings.Contains(real, greatest) {
		t.Fatalf("000107 no longer contains %q, so this non-vacuity proof no longer applies; update it", greatest)
	}
	broken := strings.Replace(real, greatest, "EXCLUDED.counter", 1)
	if broken == real {
		t.Fatal("the transform did not change the SQL")
	}

	pool := scratchSeqSchema(t)
	insertWarehouse(t, pool, "GDG-001")
	insertWarehouse(t, pool, "GDG-002")
	seedBucket(t, pool, "GDG", "2000-01-01", 50)

	applyMigration(t, pool, warehouseMigration+" (GREATEST relaxed, scratch copy only)", broken)

	got, ok := seqCounter(t, pool, "GDG", "2000-01-01")
	if !ok {
		t.Fatal("no anchor bucket")
	}
	// With GREATEST the counter is 50. Without it, the plain assignment wins
	// and the counter is CLOBBERED DOWN to 2 - which is what would happen to a
	// counter that had run ahead of the data, and exactly the data loss
	// GREATEST exists to prevent.
	if got != 2 {
		t.Fatalf("the relaxed copy left the counter at %d, want 2: GREATEST is not what is holding this property up", got)
	}
	if got == 50 {
		t.Fatal("the relaxed copy did not lower the counter, so this test is not exercising anything")
	}
}