package packs

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/binary"
	"fmt"
	"math"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/kernel/pkg/utils"
	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

type managedLifecycleColumn struct {
	name     string
	binary   bool
	nullable bool
}

type managedLifecycleTableSchema struct {
	columns    []managedLifecycleColumn
	uniqueKeys [][]string
}

type managedLifecycleCatalog interface {
	TableSchema(context.Context, string) (managedLifecycleTableSchema, error)
}

type sqlManagedLifecycleCatalog struct{ exec componentSQLExecutor }

func (c sqlManagedLifecycleCatalog) TableSchema(ctx context.Context, table string) (managedLifecycleTableSchema, error) {
	if err := validateComponentSQLIdentifier(table); err != nil {
		return managedLifecycleTableSchema{}, err
	}
	if c.exec == nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("managed lifecycle catalog SQL executor is required")
	}
	rows, err := c.exec.QueryContext(ctx, `
SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = ?
ORDER BY ORDINAL_POSITION`, table)
	if err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("query managed lifecycle columns for %s: %w", table, err)
	}
	schema := managedLifecycleTableSchema{}
	for rows.Next() {
		var name, dataType, isNullable string
		if err := rows.Scan(&name, &dataType, &isNullable); err != nil {
			rows.Close()
			return managedLifecycleTableSchema{}, fmt.Errorf("scan managed lifecycle column for %s: %w", table, err)
		}
		schema.columns = append(schema.columns, managedLifecycleColumn{
			name: name, binary: managedLifecycleBinaryType(dataType), nullable: strings.EqualFold(isNullable, "YES"),
		})
	}
	if err := rows.Close(); err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("close managed lifecycle columns for %s: %w", table, err)
	}
	if err := rows.Err(); err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("iterate managed lifecycle columns for %s: %w", table, err)
	}
	indexRows, err := c.exec.QueryContext(ctx, `
SELECT INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = ?
  AND NON_UNIQUE = 0
ORDER BY CASE WHEN INDEX_NAME = 'PRIMARY' THEN 0 ELSE 1 END, INDEX_NAME, SEQ_IN_INDEX`, table)
	if err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("query managed lifecycle unique keys for %s: %w", table, err)
	}
	var current string
	for indexRows.Next() {
		var indexName, column string
		var sequence int
		if err := indexRows.Scan(&indexName, &column, &sequence); err != nil {
			indexRows.Close()
			return managedLifecycleTableSchema{}, fmt.Errorf("scan managed lifecycle unique key for %s: %w", table, err)
		}
		if indexName != current {
			schema.uniqueKeys = append(schema.uniqueKeys, nil)
			current = indexName
		}
		schema.uniqueKeys[len(schema.uniqueKeys)-1] = append(schema.uniqueKeys[len(schema.uniqueKeys)-1], column)
	}
	if err := indexRows.Close(); err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("close managed lifecycle unique keys for %s: %w", table, err)
	}
	if err := indexRows.Err(); err != nil {
		return managedLifecycleTableSchema{}, fmt.Errorf("iterate managed lifecycle unique keys for %s: %w", table, err)
	}
	return schema, nil
}

func managedLifecycleBinaryType(dataType string) bool {
	switch strings.ToLower(strings.TrimSpace(dataType)) {
	case "binary", "varbinary", "tinyblob", "blob", "mediumblob", "longblob", "bit":
		return true
	default:
		return false
	}
}

type managedLifecycleRowStore struct {
	boundCtx context.Context
	exec     componentSQLExecutor
	catalog  managedLifecycleCatalog
	read     map[string]struct{}
	write    map[string]struct{}
	budget   cp.ComponentManagedBudget
	schemas  map[string]managedLifecycleTableSchema
	schemaMu sync.Mutex
	started  time.Time
	meter    *managedLifecycleMeter
	failure  error
}

var _ cp.ComponentManagedReadStore = (*managedLifecycleRowStore)(nil)
var _ cp.ComponentManagedTransactionReadStore = (*managedLifecycleRowStore)(nil)

func newManagedLifecycleReadStore(boundCtx context.Context, exec componentSQLExecutor, catalog managedLifecycleCatalog, writeSet []cp.ComponentWriteSurface, readSet []cp.ComponentReadSurface, budget cp.ComponentManagedBudget) (*managedLifecycleRowStore, error) {
	if boundCtx == nil {
		return nil, fmt.Errorf("managed lifecycle bound context is required")
	}
	if exec == nil {
		return nil, fmt.Errorf("managed lifecycle SQL executor is required")
	}
	if catalog == nil {
		return nil, fmt.Errorf("managed lifecycle catalog is required")
	}
	read, write := make(map[string]struct{}, len(readSet)+len(writeSet)), make(map[string]struct{}, len(writeSet))
	for _, surface := range readSet {
		if err := validateComponentSQLIdentifier(surface.Table); err != nil {
			return nil, fmt.Errorf("managed lifecycle read set: %w", err)
		}
		read[surface.Table] = struct{}{}
	}
	for _, surface := range writeSet {
		if err := validateComponentSQLIdentifier(surface.Table); err != nil {
			return nil, fmt.Errorf("managed lifecycle write set: %w", err)
		}
		read[surface.Table] = struct{}{}
		write[surface.Table] = struct{}{}
	}
	return &managedLifecycleRowStore{
		boundCtx: boundCtx, exec: exec, catalog: catalog, read: read, write: write, budget: budget,
		schemas: make(map[string]managedLifecycleTableSchema), started: time.Now(),
	}, nil
}

func newManagedLifecycleBoundedReadStore(boundCtx context.Context, exec componentSQLExecutor, catalog managedLifecycleCatalog, writeSet []cp.ComponentWriteSurface, readSet []cp.ComponentReadSurface, budget cp.ComponentManagedBudget) (*managedLifecycleRowStore, error) {
	store, err := newManagedLifecycleReadStore(boundCtx, exec, catalog, writeSet, readSet, budget)
	if err != nil {
		return nil, err
	}
	store.meter = &managedLifecycleMeter{budget: budget}
	return store, nil
}

func (s *managedLifecycleRowStore) ReadRows(ctx context.Context, query cp.ComponentManagedRowQuery) (rows []cp.ComponentRow, err error) {
	if failure := s.failureErr(); failure != nil {
		return nil, failure
	}
	defer func() { s.recordFailure(err) }()
	rows, _, err = s.readRows(ctx, query, nil, false, false, false)
	return rows, err
}

func (s *managedLifecycleRowStore) ReadRowsPage(ctx context.Context, page cp.ComponentManagedKeysetRowQuery) (rows []cp.ComponentRow, err error) {
	if failure := s.failureErr(); failure != nil {
		return nil, failure
	}
	defer func() { s.recordFailure(err) }()
	if err := validateManagedLifecycleKeysetPage(page, s.budget.MaxBatchRows); err != nil {
		return nil, err
	}
	rows, _, err = s.readRows(ctx, page.Query, page.ContinueAfter, true, false, false)
	return rows, err
}

func (s *managedLifecycleRowStore) ReadRowsKeysetPage(ctx context.Context, page cp.ComponentManagedKeysetRowQuery) (result cp.ComponentManagedKeysetRowPage, err error) {
	if failure := s.failureErr(); failure != nil {
		return result, failure
	}
	defer func() { s.recordFailure(err) }()
	if err := validateManagedLifecycleKeysetPage(page, s.budget.MaxBatchRows); err != nil {
		return result, err
	}
	result.Rows, result.Exhausted, err = s.readRows(ctx, page.Query, page.ContinueAfter, true, false, true)
	return result, err
}

func validateManagedLifecycleKeysetPage(page cp.ComponentManagedKeysetRowQuery, maxBatchRows int) error {
	if err := page.Validate(maxBatchRows); err != nil {
		return err
	}
	if len(page.Query.OrderBy) != len(page.Query.LogicalKeyColumns) {
		return fmt.Errorf("managed lifecycle keyset order must exactly match logical key columns")
	}
	for index, order := range page.Query.OrderBy {
		if order.Column != page.Query.LogicalKeyColumns[index] {
			return fmt.Errorf("managed lifecycle keyset order must exactly match logical key columns")
		}
	}
	return nil
}

func (s *managedLifecycleRowStore) failureErr() error {
	if s.failure != nil {
		return s.failure
	}
	if err := s.boundCtx.Err(); err != nil {
		return err
	}
	if !time.Now().Before(s.started.Add(s.budget.MaxElapsed)) {
		return context.DeadlineExceeded
	}
	return nil
}

func (s *managedLifecycleRowStore) recordFailure(err error) {
	if err != nil && s.failure == nil {
		s.failure = err
	}
}

func (s *managedLifecycleRowStore) readRows(
	ctx context.Context,
	query cp.ComponentManagedRowQuery,
	continueAfter cp.ComponentRow,
	keysetPagination bool,
	lock bool,
	proveExhaustion bool,
) ([]cp.ComponentRow, bool, error) {
	ctx, operationCancel := s.operationContext(ctx)
	defer operationCancel()
	if err := ctx.Err(); err != nil {
		return nil, false, err
	}
	if err := query.Validate(s.budget.MaxBatchRows); err != nil {
		return nil, false, err
	}
	if _, ok := s.read[query.Table]; !ok {
		return nil, false, fmt.Errorf("table %q is not in the managed lifecycle read set", query.Table)
	}
	columns := append(append([]string(nil), query.Columns...), sortedComponentRowKeys(query.Equal)...)
	for _, order := range query.OrderBy {
		columns = append(columns, order.Column)
	}
	columns = append(columns, query.LogicalKeyColumns...)
	schema, err := s.allowColumns(ctx, query.Table, columns)
	if err != nil {
		return nil, false, err
	}
	if keysetPagination {
		keyColumns := make([]string, len(query.OrderBy))
		for index, order := range query.OrderBy {
			keyColumns[index] = order.Column
			if managedLifecycleColumnIsNullable(schema, order.Column) {
				return nil, false, fmt.Errorf("managed lifecycle keyset logical key column %q must be NOT NULL", order.Column)
			}
		}
		for _, column := range sortedComponentRowKeys(query.Equal) {
			if query.Equal[column] == nil {
				return nil, false, fmt.Errorf("managed lifecycle keyset equality column %q cannot be NULL", column)
			}
			keyColumns = append(keyColumns, column)
		}
		if !managedLifecycleColumnsCoverUniqueKey(schema, keyColumns) {
			return nil, false, fmt.Errorf("managed lifecycle keyset equality and logical key columns must cover a live unique key")
		}
	}
	if lock && len(query.LogicalKeyColumns) == 0 {
		return nil, false, fmt.Errorf("managed lifecycle locking read requires logical key columns")
	}
	if lock && !managedLifecycleColumnsCoverUniqueKey(schema, query.LogicalKeyColumns) {
		return nil, false, fmt.Errorf("managed lifecycle locking read logical key columns must cover a live unique key")
	}
	effectiveMaximum := query.MaximumRows
	if s.meter != nil {
		remaining := s.budget.MaxCurrentItems - s.meter.usage.CurrentItems
		if remaining <= 0 {
			return nil, false, fmt.Errorf("managed lifecycle current-item budget is exhausted")
		}
		if effectiveMaximum > remaining {
			effectiveMaximum = remaining
		}
	}
	where, args := componentPredicates(query.Equal, sortedComponentRowKeys(query.Equal))
	if keysetPagination && len(continueAfter) != 0 {
		continuation, continuationArgs := managedLifecycleKeysetPredicate(query.OrderBy, continueAfter)
		where += " AND " + continuation
		args = append(args, continuationArgs...)
	}
	statement := "SELECT " + quoteComponentIdentifiers(query.Columns) + " FROM " + quoteComponentIdentifier(query.Table) + " WHERE " + where
	orders := append([]cp.ComponentManagedOrder(nil), query.OrderBy...)
	if lock && !keysetPagination {
		orders = make([]cp.ComponentManagedOrder, len(query.LogicalKeyColumns))
		for index, column := range query.LogicalKeyColumns {
			orders[index] = cp.ComponentManagedOrder{Column: column, Direction: cp.ComponentManagedOrderAscending}
		}
	}
	if len(orders) != 0 {
		parts := make([]string, len(orders))
		for index, order := range orders {
			direction := "ASC"
			if order.Direction == cp.ComponentManagedOrderDescending {
				direction = "DESC"
			}
			parts[index] = quoteComponentIdentifier(order.Column) + " " + direction
		}
		statement += " ORDER BY " + strings.Join(parts, ", ")
	}
	limit := effectiveMaximum + 1
	if keysetPagination {
		limit = effectiveMaximum
	}
	statement += fmt.Sprintf(" LIMIT %d", limit)
	if lock {
		statement += " FOR UPDATE"
	}
	queryCtx := ctx
	cancel := func() {}
	if lock && s.meter != nil {
		remaining := s.budget.MaxLockWait - s.meter.usage.LockWait
		if remaining <= 0 {
			return nil, false, fmt.Errorf("managed lifecycle lock-wait budget is exhausted")
		}
		queryCtx, cancel = context.WithTimeout(ctx, remaining)
	}
	defer cancel()
	started := time.Now()
	rows, err := s.exec.QueryContext(queryCtx, statement, args...)
	if err != nil {
		return nil, false, err
	}
	defer rows.Close()
	binaryColumns := make(map[string]bool, len(schema.columns))
	for _, column := range schema.columns {
		binaryColumns[column.name] = column.binary
	}
	out := make([]cp.ComponentRow, 0, effectiveMaximum)
	for rows.Next() {
		values := make([]any, len(query.Columns))
		destinations := make([]any, len(values))
		for index := range values {
			destinations[index] = &values[index]
		}
		if err := rows.Scan(destinations...); err != nil {
			return nil, false, err
		}
		row := make(cp.ComponentRow, len(values))
		for index, column := range query.Columns {
			value := values[index]
			if raw, ok := value.([]byte); ok {
				if binaryColumns[column] {
					value = append([]byte(nil), raw...)
				} else {
					value = string(raw)
				}
			}
			row[column] = value
		}
		if keysetPagination {
			for _, order := range query.OrderBy {
				column := order.Column
				if row[column] == nil {
					return nil, false, fmt.Errorf("managed lifecycle keyset logical key column %q cannot be NULL", column)
				}
			}
		}
		out = append(out, row)
		if len(out) > effectiveMaximum {
			if s.meter != nil {
				s.meter.recordRead(len(out), lock, time.Since(started))
			}
			return nil, false, fmt.Errorf("managed lifecycle read returned more than effective maximum %d rows", effectiveMaximum)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, false, err
	}
	if s.meter != nil {
		if err := s.meter.recordRead(len(out), lock, time.Since(started)); err != nil {
			return nil, false, err
		}
	}
	exhausted := !keysetPagination || len(out) < effectiveMaximum
	atItemBudget := s.meter != nil && s.meter.usage.CurrentItems >= s.budget.MaxCurrentItems
	if keysetPagination && len(out) == effectiveMaximum && (proveExhaustion || atItemBudget) {
		exhausted, err = s.probeManagedLifecycleKeysetExhaustion(ctx, query, out[len(out)-1])
		if err != nil {
			return nil, false, err
		}
		if !exhausted && atItemBudget {
			return nil, false, fmt.Errorf("managed lifecycle keyset page exceeds current-item budget %d", s.budget.MaxCurrentItems)
		}
	}
	if lock && s.meter != nil {
		s.meter.locks = append(s.meter.locks, managedLifecycleLockedQuery{
			table: query.Table, equal: cloneComponentRow(query.Equal), maximum: query.MaximumRows,
			complete: exhausted && len(continueAfter) == 0,
		})
	}
	return out, exhausted, nil
}

func (s *managedLifecycleRowStore) probeManagedLifecycleKeysetExhaustion(
	ctx context.Context,
	query cp.ComponentManagedRowQuery,
	after cp.ComponentRow,
) (bool, error) {
	// The sentinel is an existence proof, not a provider-visible current item.
	// Keep it non-locking so LockRowsKeysetPage neither locks nor meters a row
	// that belongs to the next page; the enclosing managed target lock provides
	// cross-operation serialization.
	where, args := componentPredicates(query.Equal, sortedComponentRowKeys(query.Equal))
	continuation, continuationArgs := managedLifecycleKeysetPredicate(query.OrderBy, after)
	where += " AND " + continuation
	args = append(args, continuationArgs...)
	parts := make([]string, len(query.OrderBy))
	for index, order := range query.OrderBy {
		direction := "ASC"
		if order.Direction == cp.ComponentManagedOrderDescending {
			direction = "DESC"
		}
		parts[index] = quoteComponentIdentifier(order.Column) + " " + direction
	}
	statement := "SELECT 1 FROM " + quoteComponentIdentifier(query.Table) + " WHERE " + where +
		" ORDER BY " + strings.Join(parts, ", ") + " LIMIT 1"
	rows, err := s.exec.QueryContext(ctx, statement, args...)
	if err != nil {
		return false, err
	}
	defer rows.Close()
	var marker int
	found := rows.Next()
	if found {
		err = rows.Scan(&marker)
	} else {
		err = rows.Err()
	}
	if err != nil {
		return false, err
	}
	return !found, nil
}

func managedLifecycleKeysetPredicate(orders []cp.ComponentManagedOrder, after cp.ComponentRow) (string, []any) {
	alternatives := make([]string, len(orders))
	args := make([]any, 0, len(orders)*(len(orders)+1)/2)
	for index, order := range orders {
		parts := make([]string, 0, index+1)
		for prefix := 0; prefix < index; prefix++ {
			parts = append(parts, quoteComponentIdentifier(orders[prefix].Column)+" = ?")
			args = append(args, after[orders[prefix].Column])
		}
		operator := ">"
		if order.Direction == cp.ComponentManagedOrderDescending {
			operator = "<"
		}
		parts = append(parts, quoteComponentIdentifier(order.Column)+" "+operator+" ?")
		args = append(args, after[order.Column])
		alternatives[index] = "(" + strings.Join(parts, " AND ") + ")"
	}
	return "(" + strings.Join(alternatives, " OR ") + ")", args
}

func (s *managedLifecycleRowStore) operationContext(callerCtx context.Context) (context.Context, context.CancelFunc) {
	deadline := s.started.Add(s.budget.MaxElapsed)
	if boundDeadline, ok := s.boundCtx.Deadline(); ok && boundDeadline.Before(deadline) {
		deadline = boundDeadline
	}
	ctx, cancel := context.WithDeadline(callerCtx, deadline)
	stop := context.AfterFunc(s.boundCtx, cancel)
	if s.boundCtx.Err() != nil {
		cancel()
	}
	return ctx, func() {
		stop()
		cancel()
	}
}

func (s *managedLifecycleRowStore) allowColumns(ctx context.Context, table string, columns []string) (managedLifecycleTableSchema, error) {
	s.schemaMu.Lock()
	schema, ok := s.schemas[table]
	if !ok {
		var err error
		schema, err = s.catalog.TableSchema(ctx, table)
		if err != nil {
			s.schemaMu.Unlock()
			return managedLifecycleTableSchema{}, fmt.Errorf("resolve managed lifecycle schema for %s: %w", table, err)
		}
		for _, column := range schema.columns {
			if err := validateComponentSQLIdentifier(column.name); err != nil {
				s.schemaMu.Unlock()
				return managedLifecycleTableSchema{}, fmt.Errorf("managed lifecycle catalog column for %s: %w", table, err)
			}
		}
		s.schemas[table] = schema
	}
	s.schemaMu.Unlock()
	allowed := make(map[string]struct{}, len(schema.columns))
	for _, column := range schema.columns {
		allowed[column.name] = struct{}{}
	}
	for _, column := range columns {
		if err := validateComponentSQLIdentifier(column); err != nil {
			return managedLifecycleTableSchema{}, err
		}
		if _, ok := allowed[column]; !ok {
			return managedLifecycleTableSchema{}, fmt.Errorf("column %q is not present in the live schema for table %q", column, table)
		}
	}
	return schema, nil
}

func managedLifecycleUniqueKeyForMatch(schema managedLifecycleTableSchema, match cp.ComponentRow) ([]string, bool) {
	for _, key := range schema.uniqueKeys {
		covered := len(key) != 0
		for _, column := range key {
			value, ok := match[column]
			if !ok || value == nil {
				covered = false
				break
			}
		}
		if covered {
			return append([]string(nil), key...), true
		}
	}
	return nil, false
}

func managedLifecycleColumnsCoverUniqueKey(schema managedLifecycleTableSchema, columns []string) bool {
	set := make(map[string]struct{}, len(columns))
	for _, column := range columns {
		set[column] = struct{}{}
	}
	for _, key := range schema.uniqueKeys {
		covered := len(key) != 0
		for _, column := range key {
			if _, ok := set[column]; !ok {
				covered = false
				break
			}
		}
		if covered {
			return true
		}
	}
	return false
}

func managedLifecycleColumnIsNullable(schema managedLifecycleTableSchema, name string) bool {
	for _, column := range schema.columns {
		if column.name == name {
			return column.nullable
		}
	}
	return false
}

func managedLifecycleStableKey(schema managedLifecycleTableSchema) ([]string, error) {
	if len(schema.uniqueKeys) == 0 || len(schema.uniqueKeys[0]) == 0 {
		return nil, fmt.Errorf("managed lifecycle bounded mutation requires a catalog unique key")
	}
	return append([]string(nil), schema.uniqueKeys[0]...), nil
}

type managedLifecycleLockedQuery struct {
	table    string
	equal    cp.ComponentRow
	maximum  int
	complete bool
}

type managedLifecycleMeter struct {
	budget cp.ComponentManagedBudget
	usage  cp.ComponentManagedUsage
	locks  []managedLifecycleLockedQuery
}

func (m *managedLifecycleMeter) recordRead(rows int, locked bool, wait time.Duration) error {
	if locked {
		m.usage.RowsLocked += rows
		m.usage.LockWait += wait
	} else {
		m.usage.RowsRead += rows
	}
	m.usage.CurrentItems += rows
	if rows > m.usage.BatchRows {
		m.usage.BatchRows = rows
	}
	return validateManagedLifecycleUsage(m.budget, m.usage)
}

func (m *managedLifecycleMeter) recordMutation(inserted, updated, deleted, statements, mutationBytes, batchRows int) error {
	m.usage.RowsInserted += inserted
	m.usage.RowsUpdated += updated
	m.usage.RowsDeleted += deleted
	m.usage.DeltaRows += inserted + updated + deleted
	m.usage.MutationStatements += statements
	m.usage.MutationBytes += mutationBytes
	if batchRows > m.usage.BatchRows {
		m.usage.BatchRows = batchRows
	}
	return validateManagedLifecycleUsage(m.budget, m.usage)
}

func validateManagedLifecycleUsage(budget cp.ComponentManagedBudget, usage cp.ComponentManagedUsage) error {
	limits := []struct {
		name          string
		actual, limit int
	}{
		{"current items", usage.CurrentItems, budget.MaxCurrentItems},
		{"desired items", usage.DesiredItems, budget.MaxDesiredItems},
		{"delta rows", usage.DeltaRows, budget.MaxDeltaRows},
		{"mutation bytes", usage.MutationBytes, budget.MaxMutationBytes},
		{"mutation statements", usage.MutationStatements, budget.MaxMutationStatements},
		{"batch rows", usage.BatchRows, budget.MaxBatchRows},
		{"attempts", usage.Attempts, budget.MaxAttempts},
	}
	for _, limit := range limits {
		if limit.actual > limit.limit {
			return fmt.Errorf("managed lifecycle actual %s %d exceeds budget %d", limit.name, limit.actual, limit.limit)
		}
	}
	if usage.LockWait > budget.MaxLockWait {
		return fmt.Errorf("managed lifecycle actual lock wait %s exceeds budget %s", usage.LockWait, budget.MaxLockWait)
	}
	if usage.Elapsed > budget.MaxElapsed {
		return fmt.Errorf("managed lifecycle actual elapsed %s exceeds budget %s", usage.Elapsed, budget.MaxElapsed)
	}
	return nil
}

type managedLifecycleMutationStore struct {
	*managedLifecycleRowStore
	meter        *managedLifecycleMeter
	idsAllocated int
}

var _ cp.ComponentManagedMutationStore = (*managedLifecycleMutationStore)(nil)
var _ cp.ComponentManagedTransactionMutationStore = (*managedLifecycleMutationStore)(nil)
var _ cp.ComponentManagedKeysetTransactionMutationStore = (*managedLifecycleMutationStore)(nil)

func (s *managedLifecycleMutationStore) execMutation(ctx context.Context, statement string, args ...any) (sql.Result, error) {
	remaining := s.budget.MaxLockWait - s.meter.usage.LockWait
	if remaining <= 0 {
		return nil, fmt.Errorf("managed lifecycle lock-wait budget is exhausted")
	}
	statementCtx, cancel := context.WithTimeout(ctx, remaining)
	defer cancel()
	started := time.Now()
	result, err := s.exec.ExecContext(statementCtx, statement, args...)
	s.meter.usage.LockWait += time.Since(started)
	if err != nil {
		return result, err
	}
	if err := validateManagedLifecycleUsage(s.budget, s.meter.usage); err != nil {
		return result, err
	}
	return result, nil
}

func newManagedLifecycleMutationStore(boundCtx context.Context, tx *sql.Tx, catalog managedLifecycleCatalog, writeSet []cp.ComponentWriteSurface, readSet []cp.ComponentReadSurface, budget cp.ComponentManagedBudget, desiredItems, attempt int) (*managedLifecycleMutationStore, error) {
	if tx == nil {
		return nil, fmt.Errorf("managed lifecycle transaction is required")
	}
	store, err := newManagedLifecycleReadStore(boundCtx, tx, catalog, writeSet, readSet, budget)
	if err != nil {
		return nil, err
	}
	meter := &managedLifecycleMeter{budget: budget, usage: cp.ComponentManagedUsage{DesiredItems: desiredItems, Attempts: attempt}}
	if err := validateManagedLifecycleUsage(budget, meter.usage); err != nil {
		return nil, err
	}
	store.meter = meter
	return &managedLifecycleMutationStore{managedLifecycleRowStore: store, meter: meter}, nil
}

func (s *managedLifecycleMutationStore) LockRows(ctx context.Context, query cp.ComponentManagedRowQuery) (rows []cp.ComponentRow, err error) {
	if failure := s.failureErr(); failure != nil {
		return nil, failure
	}
	defer func() { s.recordFailure(err) }()
	rows, _, err = s.readRows(ctx, query, nil, false, true, false)
	return rows, err
}

func (s *managedLifecycleMutationStore) LockRowsPage(ctx context.Context, page cp.ComponentManagedKeysetRowQuery) (rows []cp.ComponentRow, err error) {
	if failure := s.failureErr(); failure != nil {
		return nil, failure
	}
	defer func() { s.recordFailure(err) }()
	if err := validateManagedLifecycleKeysetPage(page, s.budget.MaxBatchRows); err != nil {
		return nil, err
	}
	for _, order := range page.Query.OrderBy {
		if order.Direction != cp.ComponentManagedOrderAscending {
			return nil, fmt.Errorf("managed lifecycle locking keyset page requires logical-key ascending order")
		}
	}
	rows, _, err = s.readRows(ctx, page.Query, page.ContinueAfter, true, true, false)
	return rows, err
}

func (s *managedLifecycleMutationStore) LockRowsKeysetPage(ctx context.Context, page cp.ComponentManagedKeysetRowQuery) (result cp.ComponentManagedKeysetRowPage, err error) {
	if failure := s.failureErr(); failure != nil {
		return result, failure
	}
	defer func() { s.recordFailure(err) }()
	if err := validateManagedLifecycleKeysetPage(page, s.budget.MaxBatchRows); err != nil {
		return result, err
	}
	for _, order := range page.Query.OrderBy {
		if order.Direction != cp.ComponentManagedOrderAscending {
			return result, fmt.Errorf("managed lifecycle locking keyset page requires logical-key ascending order")
		}
	}
	result.Rows, result.Exhausted, err = s.readRows(ctx, page.Query, page.ContinueAfter, true, true, true)
	return result, err
}

func (s *managedLifecycleMutationStore) NewID() (id string, err error) {
	if failure := s.failureErr(); failure != nil {
		return "", failure
	}
	defer func() { s.recordFailure(err) }()
	if s.idsAllocated >= s.budget.MaxDeltaRows {
		return "", fmt.Errorf("managed lifecycle allocated ID count exceeds delta-row budget %d", s.budget.MaxDeltaRows)
	}
	s.idsAllocated++
	return utils.GenerateID(), nil
}

func (s *managedLifecycleMutationStore) InsertRows(ctx context.Context, batch cp.ComponentManagedInsertBatch) (affectedRows int64, err error) {
	if failure := s.failureErr(); failure != nil {
		return 0, failure
	}
	defer func() { s.recordFailure(err) }()
	ctx, cancel := s.operationContext(ctx)
	defer cancel()
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	if err := batch.Validate(s.budget.MaxBatchRows); err != nil {
		return 0, err
	}
	if _, ok := s.write[batch.Table]; !ok {
		return 0, fmt.Errorf("table %q is not in the managed lifecycle write set", batch.Table)
	}
	type encodedInsertRow struct {
		row        cp.ComponentManagedInsertRow
		logicalKey []byte
	}
	rows := make([]encodedInsertRow, len(batch.Rows))
	mutationBytes := 0
	remainingMutationBytes := s.budget.MaxMutationBytes - s.meter.usage.MutationBytes
	for index, row := range batch.Rows {
		logicalKey, err := encodeManagedLifecycleRowBounded(row.LogicalKey, remainingMutationBytes)
		if err != nil {
			return 0, fmt.Errorf("managed lifecycle insert row %d logical key exceeds mutation-byte budget maximum %d bytes: %w", index, s.budget.MaxMutationBytes, err)
		}
		remainingMutationBytes -= len(logicalKey)
		values, err := encodeManagedLifecycleRowBounded(row.Values, remainingMutationBytes)
		if err != nil {
			return 0, fmt.Errorf("managed lifecycle insert row %d values exceed mutation-byte budget maximum %d bytes: %w", index, s.budget.MaxMutationBytes, err)
		}
		remainingMutationBytes -= len(values)
		mutationBytes += len(logicalKey) + len(values)
		rows[index] = encodedInsertRow{row: row, logicalKey: logicalKey}
	}
	prospective := s.meter.usage
	prospective.RowsInserted += len(rows)
	prospective.DeltaRows += len(rows)
	prospective.MutationStatements++
	prospective.MutationBytes += mutationBytes
	if len(rows) > prospective.BatchRows {
		prospective.BatchRows = len(rows)
	}
	if err := validateManagedLifecycleUsage(s.budget, prospective); err != nil {
		return 0, err
	}
	sort.Slice(rows, func(i, j int) bool { return bytes.Compare(rows[i].logicalKey, rows[j].logicalKey) < 0 })
	for index := 1; index < len(rows); index++ {
		if bytes.Equal(rows[index-1].logicalKey, rows[index].logicalKey) {
			return 0, fmt.Errorf("managed lifecycle insert batch contains duplicate logical key at sorted index %d", index)
		}
	}
	valueColumns := sortedComponentRowKeys(rows[0].row.Values)
	columns := append(append([]string(nil), valueColumns...), batch.CurrentTimestampColumns...)
	for index := range rows {
		if !reflect.DeepEqual(sortedComponentRowKeys(rows[index].row.Values), valueColumns) {
			return 0, fmt.Errorf("managed lifecycle insert batch row %d has a different column set", index)
		}
	}
	schema, err := s.allowColumns(ctx, batch.Table, columns)
	if err != nil {
		return 0, err
	}
	for index, encoded := range rows {
		if _, exact := managedLifecycleUniqueKeyForMatch(schema, encoded.row.LogicalKey); !exact {
			return 0, fmt.Errorf("managed lifecycle insert row %d logical key does not cover a live unique key", index)
		}
	}
	valuesSQL := make([]string, len(rows))
	args := make([]any, 0, len(rows)*len(valueColumns))
	for index, encoded := range rows {
		expressions, rowArgs := componentMutationExpressions(encoded.row.Values, valueColumns, batch.CurrentTimestampColumns, false)
		valuesSQL[index] = "(" + strings.Join(expressions, ", ") + ")"
		args = append(args, rowArgs...)
	}
	result, err := s.execMutation(ctx, "INSERT INTO "+quoteComponentIdentifier(batch.Table)+" ("+quoteComponentIdentifiers(columns)+") VALUES "+strings.Join(valuesSQL, ", "), args...)
	affected, err := componentRowsAffected(result, err)
	if err != nil {
		return 0, err
	}
	if affected < 0 || affected > int64(len(rows)) {
		return affected, fmt.Errorf("managed lifecycle insert affected %d rows; batch contains %d", affected, len(rows))
	}
	if err := s.meter.recordMutation(int(affected), 0, 0, 1, mutationBytes, len(rows)); err != nil {
		return affected, err
	}
	return affected, nil
}

func (s *managedLifecycleMutationStore) UpdateRows(ctx context.Context, update cp.ComponentManagedUpdate) (affectedRows int64, err error) {
	if failure := s.failureErr(); failure != nil {
		return 0, failure
	}
	defer func() { s.recordFailure(err) }()
	ctx, cancel := s.operationContext(ctx)
	defer cancel()
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	if err := update.Validate(s.budget.MaxBatchRows); err != nil {
		return 0, err
	}
	if _, ok := s.write[update.Table]; !ok {
		return 0, fmt.Errorf("table %q is not in the managed lifecycle write set", update.Table)
	}
	remainingMutationBytes := s.budget.MaxMutationBytes - s.meter.usage.MutationBytes
	encodedMatch, err := encodeManagedLifecycleRowBounded(update.Match, remainingMutationBytes)
	if err != nil {
		return 0, fmt.Errorf("managed lifecycle update match exceeds mutation-byte budget maximum %d bytes: %w", s.budget.MaxMutationBytes, err)
	}
	remainingMutationBytes -= len(encodedMatch)
	encodedValues, err := encodeManagedLifecycleRowBounded(update.Values, remainingMutationBytes)
	if err != nil {
		return 0, fmt.Errorf("managed lifecycle update values exceed mutation-byte budget maximum %d bytes: %w", s.budget.MaxMutationBytes, err)
	}
	valueColumns, matchColumns := sortedComponentRowKeys(update.Values), sortedComponentRowKeys(update.Match)
	columns := append(append(append([]string(nil), valueColumns...), update.CurrentTimestampColumns...), matchColumns...)
	mutationBytes := len(encodedMatch) + len(encodedValues)
	prospective := s.meter.usage
	prospective.RowsUpdated += update.MaximumAffectedRows
	prospective.DeltaRows += update.MaximumAffectedRows
	prospective.MutationStatements++
	prospective.MutationBytes += mutationBytes
	if update.MaximumAffectedRows > prospective.BatchRows {
		prospective.BatchRows = update.MaximumAffectedRows
	}
	if err := validateManagedLifecycleUsage(s.budget, prospective); err != nil {
		return 0, err
	}
	schema, err := s.allowColumns(ctx, update.Table, columns)
	if err != nil {
		return 0, err
	}
	if err := s.ensureAffectedRowsBound(ctx, update.Table, update.Match, update.MaximumAffectedRows, schema); err != nil {
		return 0, err
	}
	expressions, args := componentMutationExpressions(update.Values, valueColumns, update.CurrentTimestampColumns, true)
	where, matchArgs := componentPredicates(update.Match, matchColumns)
	statement := "UPDATE " + quoteComponentIdentifier(update.Table) + " SET " + strings.Join(expressions, ", ") + " WHERE " + where + " LIMIT " + strconv.Itoa(update.MaximumAffectedRows+1)
	result, err := s.execMutation(ctx, statement, append(args, matchArgs...)...)
	affected, err := componentRowsAffected(result, err)
	if err != nil {
		return 0, err
	}
	if affected > int64(update.MaximumAffectedRows) {
		return affected, fmt.Errorf("managed lifecycle update affected %d rows; maximum is %d", affected, update.MaximumAffectedRows)
	}
	if err := s.meter.recordMutation(0, int(affected), 0, 1, mutationBytes, int(affected)); err != nil {
		return affected, err
	}
	return affected, nil
}

func (s *managedLifecycleMutationStore) DeleteRows(ctx context.Context, deletion cp.ComponentManagedDelete) (affectedRows int64, err error) {
	if failure := s.failureErr(); failure != nil {
		return 0, failure
	}
	defer func() { s.recordFailure(err) }()
	ctx, cancel := s.operationContext(ctx)
	defer cancel()
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	if err := deletion.Validate(s.budget.MaxBatchRows); err != nil {
		return 0, err
	}
	if _, ok := s.write[deletion.Table]; !ok {
		return 0, fmt.Errorf("table %q is not in the managed lifecycle write set", deletion.Table)
	}
	remainingMutationBytes := s.budget.MaxMutationBytes - s.meter.usage.MutationBytes
	encodedMatch, err := encodeManagedLifecycleRowBounded(deletion.Match, remainingMutationBytes)
	if err != nil {
		return 0, fmt.Errorf("managed lifecycle delete match exceeds mutation-byte budget maximum %d bytes: %w", s.budget.MaxMutationBytes, err)
	}
	matchColumns := sortedComponentRowKeys(deletion.Match)
	mutationBytes := len(encodedMatch)
	prospective := s.meter.usage
	prospective.RowsDeleted += deletion.MaximumAffectedRows
	prospective.DeltaRows += deletion.MaximumAffectedRows
	prospective.MutationStatements++
	prospective.MutationBytes += mutationBytes
	if deletion.MaximumAffectedRows > prospective.BatchRows {
		prospective.BatchRows = deletion.MaximumAffectedRows
	}
	if err := validateManagedLifecycleUsage(s.budget, prospective); err != nil {
		return 0, err
	}
	schema, err := s.allowColumns(ctx, deletion.Table, matchColumns)
	if err != nil {
		return 0, err
	}
	if err := s.ensureAffectedRowsBound(ctx, deletion.Table, deletion.Match, deletion.MaximumAffectedRows, schema); err != nil {
		return 0, err
	}
	where, args := componentPredicates(deletion.Match, matchColumns)
	statement := "DELETE FROM " + quoteComponentIdentifier(deletion.Table) + " WHERE " + where + " LIMIT " + strconv.Itoa(deletion.MaximumAffectedRows+1)
	result, err := s.execMutation(ctx, statement, args...)
	affected, err := componentRowsAffected(result, err)
	if err != nil {
		return 0, err
	}
	if affected > int64(deletion.MaximumAffectedRows) {
		return affected, fmt.Errorf("managed lifecycle delete affected %d rows; maximum is %d", affected, deletion.MaximumAffectedRows)
	}
	if err := s.meter.recordMutation(0, 0, int(affected), 1, mutationBytes, int(affected)); err != nil {
		return affected, err
	}
	return affected, nil
}

func (s *managedLifecycleMutationStore) ActualReceipt() cp.ComponentManagedMutationReceipt {
	usage := s.meter.usage
	usage.Elapsed = time.Since(s.started)
	return cp.ComponentManagedMutationReceipt{Usage: usage}
}

func (s *managedLifecycleMutationStore) ensureAffectedRowsBound(ctx context.Context, table string, match cp.ComponentRow, maximum int, schema managedLifecycleTableSchema) error {
	if _, unique := managedLifecycleUniqueKeyForMatch(schema, match); unique {
		return nil
	}
	for _, locked := range s.meter.locks {
		if locked.complete && locked.table == table && locked.maximum <= maximum && managedLifecyclePredicateContains(match, locked.equal) {
			return nil
		}
	}
	stableKey, err := managedLifecycleStableKey(schema)
	if err != nil {
		return err
	}
	effectiveMaximum := maximum
	remainingItems := s.budget.MaxCurrentItems - s.meter.usage.CurrentItems
	if remainingItems <= 0 {
		return fmt.Errorf("managed lifecycle current-item budget is exhausted")
	}
	if effectiveMaximum > remainingItems {
		effectiveMaximum = remainingItems
	}
	where, args := componentPredicates(match, sortedComponentRowKeys(match))
	statement := "SELECT " + quoteComponentIdentifiers(stableKey) + " FROM " + quoteComponentIdentifier(table) + " WHERE " + where + " ORDER BY " + managedLifecycleAscendingColumns(stableKey) + " LIMIT " + strconv.Itoa(effectiveMaximum+1) + " FOR UPDATE"
	remaining := s.budget.MaxLockWait - s.meter.usage.LockWait
	if remaining <= 0 {
		return fmt.Errorf("managed lifecycle lock-wait budget is exhausted")
	}
	lockCtx, cancel := context.WithTimeout(ctx, remaining)
	defer cancel()
	started := time.Now()
	rows, err := s.exec.QueryContext(lockCtx, statement, args...)
	if err != nil {
		return err
	}
	defer rows.Close()
	count := 0
	for rows.Next() {
		values := make([]any, len(stableKey))
		destinations := make([]any, len(values))
		for index := range values {
			destinations[index] = &values[index]
		}
		if err := rows.Scan(destinations...); err != nil {
			return err
		}
		count++
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if err := s.meter.recordRead(count, true, time.Since(started)); err != nil {
		return err
	}
	if count > effectiveMaximum {
		return fmt.Errorf("managed lifecycle mutation matched %d rows; maximum is %d", count, maximum)
	}
	s.meter.locks = append(s.meter.locks, managedLifecycleLockedQuery{
		table: table, equal: cloneComponentRow(match), maximum: maximum, complete: true,
	})
	return nil
}

func managedLifecyclePredicateContains(candidate, required cp.ComponentRow) bool {
	for column, expected := range required {
		actual, ok := candidate[column]
		if !ok || !reflect.DeepEqual(actual, expected) {
			return false
		}
	}
	return true
}

func managedLifecycleAscendingColumns(columns []string) string {
	out := make([]string, len(columns))
	for index, column := range columns {
		out[index] = quoteComponentIdentifier(column) + " ASC"
	}
	return strings.Join(out, ", ")
}

func cloneComponentRow(row cp.ComponentRow) cp.ComponentRow {
	out := make(cp.ComponentRow, len(row))
	for key, value := range row {
		if raw, ok := value.([]byte); ok {
			value = append([]byte(nil), raw...)
		}
		out[key] = value
	}
	return out
}

func encodeManagedLifecycleRow(row cp.ComponentRow) ([]byte, error) {
	return encodeManagedLifecycleRowBounded(row, math.MaxInt)
}

func encodeManagedLifecycleRowBounded(row cp.ComponentRow, maximum int) ([]byte, error) {
	size, err := managedLifecycleEncodedRowSize(row, maximum)
	if err != nil {
		return nil, err
	}
	buffer := bytes.NewBuffer(make([]byte, 0, size))
	for _, column := range sortedComponentRowKeys(row) {
		writeManagedLifecycleBytes(buffer, []byte(column))
		if err := writeManagedLifecycleValue(buffer, row[column]); err != nil {
			return nil, fmt.Errorf("column %q: %w", column, err)
		}
	}
	return buffer.Bytes(), nil
}

func managedLifecycleEncodedRowSize(row cp.ComponentRow, maximum int) (int, error) {
	if maximum < 0 {
		return 0, fmt.Errorf("managed lifecycle encoded row exceeds maximum %d bytes", maximum)
	}
	total := 0
	add := func(part int) error {
		if part > maximum || total > maximum-part {
			return fmt.Errorf("managed lifecycle encoded row exceeds maximum %d bytes", maximum)
		}
		total += part
		return nil
	}
	for _, column := range sortedComponentRowKeys(row) {
		if err := add(8); err != nil {
			return 0, err
		}
		if err := add(len(column)); err != nil {
			return 0, err
		}
		if err := managedLifecycleEncodedValueSize(row[column], add); err != nil {
			return 0, fmt.Errorf("column %q: %w", column, err)
		}
	}
	return total, nil
}

func managedLifecycleEncodedValueSize(value any, add func(int) error) error {
	if err := add(1); err != nil {
		return err
	}
	if value == nil {
		return nil
	}
	addBytes := func(length int) error {
		if err := add(8); err != nil {
			return err
		}
		return add(length)
	}
	switch typed := value.(type) {
	case []byte:
		return addBytes(len(typed))
	case string:
		return addBytes(len(typed))
	case bool:
		return add(1)
	case time.Time:
		return addBytes(len(typed.UTC().Format(time.RFC3339Nano)))
	}
	reflected := reflect.ValueOf(value)
	switch reflected.Kind() {
	case reflect.String:
		return addBytes(reflected.Len())
	case reflect.Bool:
		return add(1)
	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
		return addBytes(len(strconv.FormatInt(reflected.Int(), 10)))
	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
		if reflected.Uint() > math.MaxInt64 {
			return fmt.Errorf("unsigned integer %d exceeds the SQL driver range", reflected.Uint())
		}
		return addBytes(len(strconv.FormatUint(reflected.Uint(), 10)))
	case reflect.Float32, reflect.Float64:
		return add(8)
	case reflect.Slice:
		if reflected.Type().Elem().Kind() != reflect.Uint8 {
			return fmt.Errorf("value type %T is unsupported", value)
		}
		return addBytes(reflected.Len())
	default:
		return fmt.Errorf("value type %T is unsupported", value)
	}
}

func writeManagedLifecycleBytes(buffer *bytes.Buffer, value []byte) {
	var size [8]byte
	binary.BigEndian.PutUint64(size[:], uint64(len(value)))
	buffer.Write(size[:])
	buffer.Write(value)
}

func writeManagedLifecycleValue(buffer *bytes.Buffer, value any) error {
	if value == nil {
		buffer.WriteByte(0)
		return nil
	}
	switch typed := value.(type) {
	case []byte:
		buffer.WriteByte(1)
		writeManagedLifecycleBytes(buffer, typed)
		return nil
	case string:
		buffer.WriteByte(2)
		writeManagedLifecycleBytes(buffer, []byte(typed))
		return nil
	case bool:
		buffer.WriteByte(3)
		if typed {
			buffer.WriteByte(1)
		} else {
			buffer.WriteByte(0)
		}
		return nil
	case time.Time:
		buffer.WriteByte(8)
		writeManagedLifecycleBytes(buffer, []byte(typed.UTC().Format(time.RFC3339Nano)))
		return nil
	}
	reflected := reflect.ValueOf(value)
	switch reflected.Kind() {
	case reflect.String:
		buffer.WriteByte(2)
		writeManagedLifecycleBytes(buffer, []byte(reflected.String()))
	case reflect.Bool:
		buffer.WriteByte(3)
		if reflected.Bool() {
			buffer.WriteByte(1)
		} else {
			buffer.WriteByte(0)
		}
	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
		buffer.WriteByte(4)
		writeManagedLifecycleBytes(buffer, []byte(strconv.FormatInt(reflected.Int(), 10)))
	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
		if reflected.Uint() > math.MaxInt64 {
			return fmt.Errorf("unsigned integer %d exceeds the SQL driver range", reflected.Uint())
		}
		buffer.WriteByte(5)
		writeManagedLifecycleBytes(buffer, []byte(strconv.FormatUint(reflected.Uint(), 10)))
	case reflect.Float32, reflect.Float64:
		buffer.WriteByte(6)
		var bits [8]byte
		binary.BigEndian.PutUint64(bits[:], math.Float64bits(reflected.Convert(reflect.TypeOf(float64(0))).Float()))
		buffer.Write(bits[:])
	case reflect.Slice:
		if reflected.Type().Elem().Kind() != reflect.Uint8 {
			return fmt.Errorf("value type %T is unsupported", value)
		}
		buffer.WriteByte(1)
		writeManagedLifecycleBytes(buffer, reflected.Bytes())
	default:
		return fmt.Errorf("value type %T is unsupported", value)
	}
	return nil
}