package packs

import (
	"bytes"
	"context"
	"errors"
	"math"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

type staticManagedLifecycleCatalog map[string]managedLifecycleTableSchema

func (c staticManagedLifecycleCatalog) TableSchema(_ context.Context, table string) (managedLifecycleTableSchema, error) {
	return c[table], nil
}

func TestManagedLifecycleRowEncodingIsBinarySafeAndRejectsUnstableValues(t *testing.T) {
	left, err := encodeManagedLifecycleRow(cp.ComponentRow{"A": "a\x00b", "B": "c"})
	if err != nil {
		t.Fatal(err)
	}
	right, err := encodeManagedLifecycleRow(cp.ComponentRow{"A": "a", "B": "b\x00c"})
	if err != nil {
		t.Fatal(err)
	}
	if bytes.Equal(left, right) {
		t.Fatal("length-prefixed row encoding collided across zero-byte boundaries")
	}
	if _, err := encodeManagedLifecycleRowBounded(cp.ComponentRow{"A": "a\x00b", "B": "c"}, len(left)-1); err == nil {
		t.Fatal("row encoding exceeded its allocation bound")
	}
	exact, err := encodeManagedLifecycleRowBounded(cp.ComponentRow{"A": "a\x00b", "B": "c"}, len(left))
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(left, exact) {
		t.Fatal("bounded row encoding changed the stable representation")
	}
	type namedInteger int
	plain, err := encodeManagedLifecycleRow(cp.ComponentRow{"ID": int(7)})
	if err != nil {
		t.Fatal(err)
	}
	named, err := encodeManagedLifecycleRow(cp.ComponentRow{"ID": namedInteger(7)})
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(plain, named) {
		t.Fatal("equivalent integer values did not receive one stable encoding")
	}
	if _, err := encodeManagedLifecycleRow(cp.ComponentRow{"ID": map[string]string{"unstable": "value"}}); err == nil {
		t.Fatal("map value received an address-dependent fallback encoding")
	}
	if _, err := encodeManagedLifecycleRow(cp.ComponentRow{"ID": uint64(math.MaxUint64)}); err == nil {
		t.Fatal("unsigned value outside the SQL driver range was accepted")
	}
}

func TestManagedLifecycleReadStoreBoundsRowsAndPreservesBinaryColumns(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "Owner"}, {name: "Label"}, {name: "BinaryKey", binary: true}},
		uniqueKeys: [][]string{{"BinaryKey"}},
	}}
	store, err := newManagedLifecycleReadStore(context.Background(), db, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		testManagedLifecycleBudget(),
	)
	if err != nil {
		t.Fatal(err)
	}
	query := cp.ComponentManagedRowQuery{
		Table: "ManagedRows", Columns: []string{"Label", "BinaryKey"}, Equal: cp.ComponentRow{"Owner": "pack"},
		OrderBy: []cp.ComponentManagedOrder{{Column: "Label", Direction: cp.ComponentManagedOrderAscending}}, MaximumRows: 1,
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Label`, `BinaryKey` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `Label` ASC LIMIT 2")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"Label", "BinaryKey"}).AddRow([]byte("open"), []byte{'a', 0, 'b'}))
	rows, err := store.ReadRows(context.Background(), query)
	if err != nil {
		t.Fatal(err)
	}
	if label, ok := rows[0]["Label"].(string); !ok || label != "open" {
		t.Fatalf("text value = %#v", rows[0]["Label"])
	}
	if binary, ok := rows[0]["BinaryKey"].([]byte); !ok || string(binary) != "a\x00b" {
		t.Fatalf("binary value = %#v", rows[0]["BinaryKey"])
	}

	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Label`, `BinaryKey` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `Label` ASC LIMIT 2")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"Label", "BinaryKey"}).AddRow("a", []byte("a")).AddRow("b", []byte("b")))
	if _, err := store.ReadRows(context.Background(), query); err == nil {
		t.Fatal("maximum-plus-one read was accepted")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStorePagesByLogicalKeyWithoutSilentCap(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "Owner"}, {name: "Sequence"}, {name: "ID"}},
		uniqueKeys: [][]string{{"Owner", "Sequence", "ID"}},
	}}
	store, err := newManagedLifecycleReadStore(context.Background(), db, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		testManagedLifecycleBudget(),
	)
	if err != nil {
		t.Fatal(err)
	}
	query := cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"Sequence", "ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy: []cp.ComponentManagedOrder{
				{Column: "Sequence", Direction: cp.ComponentManagedOrderDescending},
				{Column: "ID", Direction: cp.ComponentManagedOrderAscending},
			},
			LogicalKeyColumns: []string{"Sequence", "ID"}, MaximumRows: 2,
		},
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Sequence`, `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `Sequence` DESC, `ID` ASC LIMIT 2")).
		WithArgs("pack").
		WillReturnRows(sqlmock.NewRows([]string{"Sequence", "ID"}).AddRow(9, "a").AddRow(9, "b"))
	first, err := store.ReadRowsPage(context.Background(), query)
	if err != nil {
		t.Fatal(err)
	}
	if len(first) != 2 {
		t.Fatalf("first page rows = %d, want 2", len(first))
	}
	query.ContinueAfter = cp.ComponentRow{"Sequence": 9, "ID": "b"}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Sequence`, `ID` FROM `ManagedRows` WHERE `Owner` = ? AND ((`Sequence` < ?) OR (`Sequence` = ? AND `ID` > ?)) ORDER BY `Sequence` DESC, `ID` ASC LIMIT 2")).
		WithArgs("pack", 9, 9, "b").
		WillReturnRows(sqlmock.NewRows([]string{"Sequence", "ID"}).AddRow(8, "c"))
	second, err := store.ReadRowsPage(context.Background(), query)
	if err != nil {
		t.Fatal(err)
	}
	if len(second) != 1 || second[0]["ID"] != "c" {
		t.Fatalf("second page = %#v", second)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreRejectsNullKeysetValues(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := newManagedLifecycleReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow(nil))
	_, err = store.ReadRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err == nil || !strings.Contains(err.Error(), `logical key column "ID" cannot be NULL`) {
		t.Fatalf("nullable keyset error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreRejectsNullableDescendingKeysetSchema(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := newManagedLifecycleReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID", nullable: true}, {name: "Owner"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.ReadRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderDescending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err == nil || !strings.Contains(err.Error(), `logical key column "ID" must be NOT NULL`) {
		t.Fatalf("nullable descending keyset schema error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreRejectsNonUniqueKeysetOrder(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := newManagedLifecycleReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Sequence"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.ReadRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"Sequence"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "Sequence", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"Sequence"}, MaximumRows: 1,
		},
	})
	if err == nil || !strings.Contains(err.Error(), "must cover a live unique key") {
		t.Fatalf("non-unique keyset error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreRejectsMismatchedKeysetOrder(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := newManagedLifecycleReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Sequence"}},
		uniqueKeys: [][]string{{"Owner", "ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.ReadRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID", "Sequence"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "Sequence", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err == nil || !strings.Contains(err.Error(), "must match logical key column") {
		t.Fatalf("mismatched keyset error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreDoesNotExposeBudgetLimitedShortPage(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	budget := testManagedLifecycleBudget()
	budget.MaxCurrentItems = 1
	store, err := newManagedLifecycleBoundedReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, budget)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1"))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `ManagedRows` WHERE `Owner` = ? AND ((`ID` > ?)) ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack", "1").WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	_, err = store.ReadRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 2,
		},
	})
	if err == nil || !strings.Contains(err.Error(), "exceeds current-item budget 1") {
		t.Fatalf("budget-limited keyset error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreProvesExactBudgetExhaustion(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	budget := testManagedLifecycleBudget()
	budget.MaxCurrentItems = 1
	store, err := newManagedLifecycleBoundedReadStore(context.Background(), db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, budget)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1"))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `ManagedRows` WHERE `Owner` = ? AND ((`ID` > ?)) ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack", "1").WillReturnRows(sqlmock.NewRows([]string{"1"}))
	page, err := store.ReadRowsKeysetPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err != nil || len(page.Rows) != 1 || !page.Exhausted {
		t.Fatalf("exact-budget page = %#v, error = %v", page, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreRejectsDescendingLockPages(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(
		context.Background(),
		tx,
		staticManagedLifecycleCatalog{"ManagedRows": {
			columns:    []managedLifecycleColumn{{name: "Owner"}, {name: "ID"}},
			uniqueKeys: [][]string{{"Owner", "ID"}},
		}},
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		testManagedLifecycleBudget(),
		0,
		1,
	)
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.LockRowsPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderDescending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err == nil || !strings.Contains(err.Error(), "locking keyset page requires logical-key ascending order") {
		t.Fatalf("descending locking page error = %v", err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreProvesExactLockBudgetExhaustion(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	budget.MaxCurrentItems = 1
	store, err := newManagedLifecycleMutationStore(
		context.Background(),
		tx,
		staticManagedLifecycleCatalog{"ManagedRows": {
			columns:    []managedLifecycleColumn{{name: "Owner"}, {name: "ID"}, {name: "Label"}},
			uniqueKeys: [][]string{{"ID"}},
		}},
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		budget,
		0,
		1,
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 1 FOR UPDATE")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1"))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `ManagedRows` WHERE `Owner` = ? AND ((`ID` > ?)) ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack", "1").WillReturnRows(sqlmock.NewRows([]string{"1"}))
	page, err := store.LockRowsKeysetPage(context.Background(), cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		},
	})
	if err != nil || len(page.Rows) != 1 || !page.Exhausted || store.ActualReceipt().Usage.RowsLocked != 1 {
		t.Fatalf("exact lock-budget page = %#v, receipt = %#v, error = %v", page, store.ActualReceipt(), err)
	}
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `Owner` = ? LIMIT 2")).
		WithArgs("changed", "pack").
		WillReturnResult(sqlmock.NewResult(0, 1))
	affected, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"},
		Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 1,
	})
	if err != nil || affected != 1 {
		t.Fatalf("exact lock-budget update = %d, %v", affected, err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreDoesNotCacheExhaustedKeysetSuffixAsCompletePredicate(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(
		context.Background(),
		tx,
		staticManagedLifecycleCatalog{"ManagedRows": {
			columns:    []managedLifecycleColumn{{name: "Owner"}, {name: "ID"}, {name: "Label"}},
			uniqueKeys: [][]string{{"ID"}},
		}},
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		testManagedLifecycleBudget(),
		0,
		1,
	)
	if err != nil {
		t.Fatal(err)
	}
	page := cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"Owner": "pack"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 2,
		},
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 2 FOR UPDATE")).
		WithArgs("pack").
		WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1").AddRow("2"))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `ManagedRows` WHERE `Owner` = ? AND ((`ID` > ?)) ORDER BY `ID` ASC LIMIT 1")).
		WithArgs("pack", "2").
		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	first, err := store.LockRowsKeysetPage(context.Background(), page)
	if err != nil || first.Exhausted {
		t.Fatalf("first lock page = %#v, error = %v", first, err)
	}

	page.ContinueAfter = cp.ComponentRow{"ID": "2"}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? AND ((`ID` > ?)) ORDER BY `ID` ASC LIMIT 2 FOR UPDATE")).
		WithArgs("pack", "2").
		WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("3"))
	second, err := store.LockRowsKeysetPage(context.Background(), page)
	if err != nil || !second.Exhausted {
		t.Fatalf("second lock page = %#v, error = %v", second, err)
	}

	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 4 FOR UPDATE")).
		WithArgs("pack").
		WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1").AddRow("2").AddRow("3"))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `Owner` = ? LIMIT 4")).
		WithArgs("changed", "pack").
		WillReturnResult(sqlmock.NewResult(0, 3))
	affected, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"},
		Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 3,
	})
	if err != nil || affected != 3 {
		t.Fatalf("broad update after keyset suffix = %d, %v", affected, err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleReadStoreHonorsBoundContextWhenCallerReplacesContext(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	boundCtx, cancel := context.WithCancel(context.Background())
	store, err := newManagedLifecycleReadStore(boundCtx, db, staticManagedLifecycleCatalog{"ManagedRows": {
		columns: []managedLifecycleColumn{{name: "ID"}},
	}}, nil, []cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	cancel()
	_, err = store.ReadRows(context.Background(), cp.ComponentManagedRowQuery{
		Table: "ManagedRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"ID": "one"}, MaximumRows: 1,
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("read after bound context cancellation = %v, want context canceled", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreEnforcesOwnElapsedBudgetAndReportsUsage(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, staticManagedLifecycleCatalog{}, nil, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	store.started = time.Now().Add(-2 * budget.MaxElapsed)
	if _, err := store.NewID(); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("NewID() after elapsed budget = %v, want deadline exceeded", err)
	}
	if receipt := store.ActualReceipt(); receipt.Usage.Elapsed <= budget.MaxElapsed {
		t.Fatalf("receipt elapsed = %s, want greater than %s", receipt.Usage.Elapsed, budget.MaxElapsed)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreRejectsAffectedRowOverflowBeforeDML(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Label"}},
		uniqueKeys: [][]string{{"ID"}},
	}}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}},
		[]cp.ComponentReadSurface{{Table: "ManagedRows", Fact: "rows"}},
		testManagedLifecycleBudget(), 1, 1,
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 2 FOR UPDATE")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1").AddRow("2"))
	if _, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"}, Values: cp.ComponentRow{"Label": "updated"}, MaximumAffectedRows: 1,
	}); err == nil {
		t.Fatal("maximum-plus-one update candidates were accepted")
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	receipt := store.ActualReceipt()
	if receipt.Usage.RowsLocked != 2 || receipt.Usage.CurrentItems != 2 || receipt.Usage.MutationStatements != 0 {
		t.Fatalf("overflow receipt = %#v", receipt)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreDoesNotTreatNullableUniqueKeyAsOneRow(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "NullableKey"}, {name: "Label"}},
		uniqueKeys: [][]string{{"NullableKey"}},
	}}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil,
		testManagedLifecycleBudget(), 1, 1,
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `NullableKey` FROM `ManagedRows` WHERE `NullableKey` IS NULL ORDER BY `NullableKey` ASC LIMIT 2 FOR UPDATE")).
		WillReturnRows(sqlmock.NewRows([]string{"NullableKey"}).AddRow(nil).AddRow(nil))
	if _, err := store.DeleteRows(context.Background(), cp.ComponentManagedDelete{
		Table: "ManagedRows", Match: cp.ComponentRow{"NullableKey": nil}, MaximumAffectedRows: 1,
	}); err == nil || !strings.Contains(err.Error(), "matched 2 rows") {
		t.Fatalf("nullable unique delete error = %v", err)
	}
	if store.failureErr() == nil {
		t.Fatal("nullable unique overflow did not poison the attempt")
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStorePoisonsPostDMLOverflow(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Label"}},
		uniqueKeys: [][]string{{"ID"}},
	}}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil,
		testManagedLifecycleBudget(), 1, 1,
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `ID` = ? LIMIT 2")).
		WithArgs("changed", "one").WillReturnResult(sqlmock.NewResult(0, 2))
	if _, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 1,
	}); err == nil || !strings.Contains(err.Error(), "affected 2 rows") {
		t.Fatalf("post-DML overflow error = %v", err)
	}
	if store.failureErr() == nil {
		t.Fatal("post-DML overflow did not poison the attempt")
	}
	if _, err := store.NewID(); err == nil || !strings.Contains(err.Error(), "affected 2 rows") {
		t.Fatalf("operation after poison error = %v", err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreBoundsDMLByRemainingLockWait(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	budget.MaxLockWait = 10 * time.Millisecond
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, staticManagedLifecycleCatalog{"ManagedRows": {
		columns: []managedLifecycleColumn{{name: "ID"}, {name: "Label"}}, uniqueKeys: [][]string{{"ID"}},
	}}, []cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `ID` = ? LIMIT 2")).
		WithArgs("changed", "one").WillDelayFor(100 * time.Millisecond).WillReturnResult(sqlmock.NewResult(0, 1))
	started := time.Now()
	_, err = store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 1,
	})
	if err == nil {
		t.Fatal("lock-wait budget did not cancel delayed DML")
	}
	if elapsed := time.Since(started); elapsed >= 50*time.Millisecond {
		t.Fatalf("delayed DML returned after %s, want lock-wait cancellation", elapsed)
	}
	if receipt := store.ActualReceipt(); receipt.Usage.LockWait <= 0 {
		t.Fatalf("receipt lock wait = %s, want observed DML wait", receipt.Usage.LockWait)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreReusesCoveringLockWithoutHiddenProbe(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Label"}},
		uniqueKeys: [][]string{{"ID"}},
	}}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil,
		testManagedLifecycleBudget(), 1, 1,
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID`, `Owner`, `Label` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 3 FOR UPDATE")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID", "Owner", "Label"}).AddRow("1", "pack", "old"))
	if _, err := store.LockRows(context.Background(), cp.ComponentManagedRowQuery{
		Table: "ManagedRows", Columns: []string{"ID", "Owner", "Label"}, Equal: cp.ComponentRow{"Owner": "pack"},
		LogicalKeyColumns: []string{"ID"}, MaximumRows: 2,
	}); err != nil {
		t.Fatal(err)
	}
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `Label` = ? AND `Owner` = ? LIMIT 3")).
		WithArgs("changed", "old", "pack").WillReturnResult(sqlmock.NewResult(0, 1))
	affected, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack", "Label": "old"},
		Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 2,
	})
	if err != nil || affected != 1 {
		t.Fatalf("covered update = %d, %v", affected, err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	receipt := store.ActualReceipt()
	if receipt.Usage.RowsLocked != 1 || receipt.Usage.RowsUpdated != 1 || receipt.Usage.MutationStatements != 1 {
		t.Fatalf("covered-lock receipt = %#v", receipt)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreReusesHiddenBoundedProbe(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	budget.MaxCurrentItems = 1
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Label"}},
		uniqueKeys: [][]string{{"ID"}},
	}}, []cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 2 FOR UPDATE")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1"))
	for _, label := range []string{"first", "second"} {
		mock.ExpectExec(regexp.QuoteMeta("UPDATE `ManagedRows` SET `Label` = ? WHERE `Owner` = ? LIMIT 2")).
			WithArgs(label, "pack").WillReturnResult(sqlmock.NewResult(0, 1))
		affected, err := store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
			Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"},
			Values: cp.ComponentRow{"Label": label}, MaximumAffectedRows: 1,
		})
		if err != nil || affected != 1 {
			t.Fatalf("bounded update %q = %d, %v", label, affected, err)
		}
	}
	if receipt := store.ActualReceipt(); receipt.Usage.RowsLocked != 1 || receipt.Usage.CurrentItems != 1 {
		t.Fatalf("hidden-probe receipt = %#v", receipt)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

type recordingManagedLifecycleCatalog struct {
	calls  int
	schema managedLifecycleTableSchema
}

func (c *recordingManagedLifecycleCatalog) TableSchema(context.Context, string) (managedLifecycleTableSchema, error) {
	c.calls++
	return c.schema, nil
}

func TestManagedLifecycleStoreRejectsUnauthorizedTableBeforeCatalogOrSQL(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := &recordingManagedLifecycleCatalog{}
	store, err := newManagedLifecycleReadStore(context.Background(), db, catalog, nil,
		[]cp.ComponentReadSurface{{Table: "AllowedRows", Fact: "rows"}}, testManagedLifecycleBudget())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.ReadRows(context.Background(), cp.ComponentManagedRowQuery{
		Table: "ForbiddenRows", Columns: []string{"ID"}, Equal: cp.ComponentRow{"ID": "one"}, MaximumRows: 1,
	}); err == nil {
		t.Fatal("unauthorized read was accepted")
	}
	if catalog.calls != 0 {
		t.Fatalf("unauthorized read reached catalog %d times", catalog.calls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreRejectsMutationBudgetBeforeCatalogOrProbe(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	catalog := &recordingManagedLifecycleCatalog{schema: managedLifecycleTableSchema{
		columns: []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Label"}}, uniqueKeys: [][]string{{"ID"}},
	}}
	budget := testManagedLifecycleBudget()
	budget.MaxDeltaRows = 1
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"}, Values: cp.ComponentRow{"Label": "updated"}, MaximumAffectedRows: 2,
	})
	if err == nil || !strings.Contains(err.Error(), "actual delta rows 2 exceeds budget 1") {
		t.Fatalf("over-budget update error = %v", err)
	}
	if catalog.calls != 0 {
		t.Fatalf("over-budget update reached catalog %d times", catalog.calls)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreRejectsOversizedValuesBeforeCatalog(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	catalog := &recordingManagedLifecycleCatalog{schema: managedLifecycleTableSchema{
		columns: []managedLifecycleColumn{{name: "ID"}, {name: "Label"}}, uniqueKeys: [][]string{{"ID"}},
	}}
	budget := testManagedLifecycleBudget()
	budget.MaxMutationBytes = 64
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	_, err = store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"Label": strings.Repeat("x", 1024)}, MaximumAffectedRows: 1,
	})
	if err == nil || !strings.Contains(err.Error(), "maximum 64 bytes") {
		t.Fatalf("oversized update error = %v", err)
	}
	if catalog.calls != 0 {
		t.Fatalf("oversized update reached catalog %d times", catalog.calls)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationProbeUsesRemainingCurrentItemBudget(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	budget.MaxCurrentItems = 1
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, staticManagedLifecycleCatalog{"ManagedRows": {
		columns: []managedLifecycleColumn{{name: "ID"}, {name: "Owner"}, {name: "Label"}}, uniqueKeys: [][]string{{"ID"}},
	}}, []cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID` FROM `ManagedRows` WHERE `Owner` = ? ORDER BY `ID` ASC LIMIT 2 FOR UPDATE")).
		WithArgs("pack").WillReturnRows(sqlmock.NewRows([]string{"ID"}).AddRow("1").AddRow("2"))
	_, err = store.UpdateRows(context.Background(), cp.ComponentManagedUpdate{
		Table: "ManagedRows", Match: cp.ComponentRow{"Owner": "pack"}, Values: cp.ComponentRow{"Label": "updated"}, MaximumAffectedRows: 10,
	})
	if err == nil || !strings.Contains(err.Error(), "actual current items 2 exceeds budget 1") {
		t.Fatalf("current-item probe error = %v", err)
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleMutationStoreBoundsIDsAndRejectsDuplicateInsertKeys(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	catalog := staticManagedLifecycleCatalog{"ManagedRows": {
		columns:    []managedLifecycleColumn{{name: "ID"}, {name: "Label"}},
		uniqueKeys: [][]string{{"ID"}},
	}}
	mock.ExpectBegin()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	budget := testManagedLifecycleBudget()
	budget.MaxDeltaRows = 2
	store, err := newManagedLifecycleMutationStore(context.Background(), tx, catalog,
		[]cp.ComponentWriteSurface{{Table: "ManagedRows", Fact: "rows"}}, nil, budget, 1, 1)
	if err != nil {
		t.Fatal(err)
	}
	for index := 0; index < budget.MaxDeltaRows; index++ {
		if id, err := store.NewID(); err != nil || id == "" {
			t.Fatalf("ID %d = %q, %v", index, id, err)
		}
	}
	if _, err := store.NewID(); err == nil {
		t.Fatal("ID allocation exceeded the delta-row budget")
	}
	if _, err := store.InsertRows(context.Background(), cp.ComponentManagedInsertBatch{
		Table: "ManagedRows",
		Rows: []cp.ComponentManagedInsertRow{
			{LogicalKey: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"ID": "one", "Label": "first"}},
			{LogicalKey: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"ID": "one", "Label": "second"}},
		},
	}); err == nil {
		t.Fatal("duplicate logical insert keys were accepted")
	}
	mock.ExpectRollback()
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func testManagedLifecycleBudget() cp.ComponentManagedBudget {
	return cp.ComponentManagedBudget{
		MaxTargets: 10, MaxCurrentItems: 100, MaxDesiredItems: 100, MaxDeltaRows: 100,
		MaxMutationBytes: 1 << 20, MaxMutationStatements: 100, MaxBatchRows: 10,
		MaxEvidenceWrites: 1, MaxEffectCalls: 10,
		MaxLockWait: time.Second, MaxElapsed: time.Second, MaxAttempts: 3,
	}
}