package packs

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"sync"
	"time"

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

const (
	managedLifecycleMaximumProjectionIdentityBytes = 1024
	managedLifecycleMaximumEffectReceiptBytes      = 64 << 10
	managedLifecycleMaximumEvidenceReceiptBytes    = 1 << 20
)

type managedLifecyclePreflight struct {
	plan cp.ComponentManagedPlan
}

// ManagedLifecycleHost is the single Plan/Mutate/Project execution boundary.
// Registration, transactions, target serialization, stores, effects,
// evidence, retry, and verification remain host-owned.
type ManagedLifecycleHost struct {
	db           *sql.DB
	catalog      managedLifecycleCatalog
	registration cp.ComponentKindRegistration
	lifecycle    cp.ComponentManagedLifecycleRegistration
	effects      map[cp.ComponentManagedEffectClass]ManagedLifecycleEffectAdapter

	preflightMu sync.Mutex
	preflights  map[[sha256.Size]byte][]managedLifecyclePreflight
	ddlGate     chan struct{}
	ddlReady    bool
}

// ApplyItem executes one preflighted item through the host-owned transaction
// protocol.
func (h *ManagedLifecycleHost) ApplyItem(ctx context.Context, request cp.ComponentManagedRequest) (ManagedLifecycleOperationResult, error) {
	if err := h.validateItemRequest(request); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	return h.apply(ctx, cp.ComponentManagedScopeItem, cp.NewComponentManagedItemLifecycleRequest(request), true)
}

// ReconcileSet executes one preflighted complete desired set.
func (h *ManagedLifecycleHost) ReconcileSet(ctx context.Context, request cp.ComponentManagedSetRequest, desiredBusinessKeys []string) (ManagedLifecycleOperationResult, error) {
	if err := h.validateSetRequest(request); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	if err := h.validateDesiredSetItems(desiredBusinessKeys); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	lifecycleRequest, err := cp.NewComponentManagedSetLifecycleRequest(request, desiredBusinessKeys)
	if err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	return h.apply(ctx, cp.ComponentManagedScopeSet, lifecycleRequest, true)
}

func (h *ManagedLifecycleHost) applyItemWithoutVerification(ctx context.Context, request cp.ComponentManagedRequest) (ManagedLifecycleOperationResult, error) {
	if err := h.validateItemRequest(request); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	return h.apply(ctx, cp.ComponentManagedScopeItem, cp.NewComponentManagedItemLifecycleRequest(request), false)
}

func (h *ManagedLifecycleHost) reconcileSetWithoutVerification(ctx context.Context, request cp.ComponentManagedSetRequest, desiredBusinessKeys []string) (ManagedLifecycleOperationResult, error) {
	if err := h.validateSetRequest(request); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	if err := h.validateDesiredSetItems(desiredBusinessKeys); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	lifecycleRequest, err := cp.NewComponentManagedSetLifecycleRequest(request, desiredBusinessKeys)
	if err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	return h.apply(ctx, cp.ComponentManagedScopeSet, lifecycleRequest, false)
}

func (h *ManagedLifecycleHost) apply(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, verifyAfterCommit bool) (ManagedLifecycleOperationResult, error) {
	desiredItems := managedLifecycleDesiredItems(scope, request)
	if err := validateManagedLifecycleUsage(h.lifecycle.Budget(), cp.ComponentManagedUsage{DesiredItems: desiredItems}); err != nil {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, err
	}
	preflight, ok := h.takePreflight(scope, request)
	if !ok {
		return ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure}, fmt.Errorf("managed lifecycle %s has no complete preflight for this request", h.lifecycle.Kind())
	}
	budget := h.lifecycle.Budget()
	operationCtx, cancel := context.WithTimeout(ctx, budget.MaxElapsed)
	defer cancel()
	started := time.Now()
	operationID := []byte(utils.GenerateID())
	result := ManagedLifecycleOperationResult{Disposition: ManagedLifecycleOperationNoMutationFailure, OperationID: append([]byte(nil), operationID...)}
	if err := h.ensureCoordinationTables(operationCtx); err != nil {
		return result, err
	}
	lifecycle := h.lifecycle.Lifecycle()
	provider := managedLifecycleAttemptProvider{
		replan: func(ctx context.Context, store cp.ComponentManagedTransactionReadStore) (cp.ComponentManagedPlan, error) {
			return lifecycle.Plan(ctx, scope, request.Clone(), managedLifecycleReadView{delegate: store})
		},
		mutate: func(ctx context.Context, plan cp.ComponentManagedPlan, store cp.ComponentManagedTransactionMutationStore) (cp.ComponentManagedMutationResult, error) {
			return lifecycle.Mutate(ctx, scope, request.Clone(), plan, store)
		},
	}
	ambientPlan := preflight.plan
	for attempt := 1; attempt <= budget.MaxAttempts; attempt++ {
		result.Attempts = attempt
		preparationWait, err := h.prepareManagedLifecycleAttemptTargets(operationCtx, ambientPlan.Targets(), started)
		if err != nil {
			return result, err
		}
		transaction, err := beginManagedLifecycleSemanticTransaction(operationCtx, h.db)
		if err != nil {
			return result, err
		}
		catalog := managedLifecycleCatalogForTransaction(h.catalog, transaction.tx)
		attemptResult, attemptErr := h.runManagedLifecycleAttempt(operationCtx, managedLifecycleAttemptRequest{
			tx: transaction.tx, catalog: catalog, operationID: operationID, attempt: attempt, scope: scope,
			desiredItems: desiredItems, ambientPlan: ambientPlan, preparationWait: preparationWait,
			started: started, provider: provider,
		})
		if attemptErr != nil {
			if !transaction.rollback() {
				result.Disposition = ManagedLifecycleOperationIndeterminate
				return result, fmt.Errorf("managed lifecycle attempt failed and rollback was not confirmed: %w", attemptErr)
			}
			if errors.Is(attemptErr, errManagedLifecycleTargetDrift) {
				ambientPlan = attemptResult.lockedPlan
				continue
			}
			if managedLifecycleAttemptCanRetry(attemptErr) && attempt < budget.MaxAttempts {
				if attemptResult.lockedPlanReady {
					ambientPlan = attemptResult.lockedPlan
				}
				continue
			}
			result.Disposition = ManagedLifecycleOperationRolledBackFailure
			return result, attemptErr
		}
		commitErr, closeErr := transaction.commit()
		if commitErr != nil {
			if managedLifecycleDeterministicTransactionError(commitErr) {
				result.Disposition = ManagedLifecycleOperationRolledBackFailure
				return result, errors.Join(fmt.Errorf("commit managed lifecycle transaction: %w", commitErr), closeErr)
			}
			proof, proofErr := h.proveCommitOutcome(
				operationCtx,
				operationID,
				attempt,
				scope,
				request,
				attemptResult.lockedPlan,
				attemptResult.mutation,
				attemptResult.effectReceipts,
				attemptResult.semanticMutationRows,
				verifyAfterCommit,
			)
			if proof.evidenceFound {
				result.Verification = proof.verification
				if proofErr != nil || !proof.proved {
					result.Disposition = ManagedLifecycleOperationIndeterminate
					return result, errors.Join(fmt.Errorf("commit response failed after durable managed lifecycle evidence: %w", commitErr), proofErr, closeErr)
				}
				switch attemptResult.mutation.Disposition {
				case cp.ComponentManagedMutationNoChange:
					result.Disposition = ManagedLifecycleOperationNoChange
				case cp.ComponentManagedMutationDesiredProjectionStaged:
					result.Disposition = ManagedLifecycleOperationChanged
				case cp.ComponentManagedMutationSafeProjectionStaged:
					result.Disposition = ManagedLifecycleOperationSafe
				}
				if closeErr != nil {
					return result, fmt.Errorf("managed lifecycle committed but session cleanup failed: %w", closeErr)
				}
				return result, nil
			}
			if proofErr != nil {
				result.Disposition = ManagedLifecycleOperationIndeterminate
				return result, errors.Join(fmt.Errorf("commit managed lifecycle transaction: %w", commitErr), proofErr, closeErr)
			}
			result.Disposition = ManagedLifecycleOperationIndeterminate
			return result, errors.Join(fmt.Errorf("commit managed lifecycle transaction without durable evidence: %w", commitErr), closeErr)
		}
		if !verifyAfterCommit {
			switch attemptResult.mutation.Disposition {
			case cp.ComponentManagedMutationNoChange:
				result.Disposition = ManagedLifecycleOperationNoChange
			case cp.ComponentManagedMutationDesiredProjectionStaged:
				result.Disposition = ManagedLifecycleOperationChanged
			case cp.ComponentManagedMutationSafeProjectionStaged:
				result.Disposition = ManagedLifecycleOperationSafe
			}
			if closeErr != nil {
				return result, fmt.Errorf("managed lifecycle committed but session cleanup failed: %w", closeErr)
			}
			return result, nil
		}
		verification, err := h.verify(operationCtx, scope, request)
		result.Verification = verification
		if err != nil || !verification.Matches || !bytes.Equal(verification.Projection.Identity, attemptResult.lockedPlan.ProjectionIdentity()) {
			result.Disposition = ManagedLifecycleOperationVerificationFailure
			if err == nil && !bytes.Equal(verification.Projection.Identity, attemptResult.lockedPlan.ProjectionIdentity()) {
				err = fmt.Errorf("managed lifecycle projection identity changed after commit")
			}
			if err == nil {
				err = fmt.Errorf("managed lifecycle committed projection does not match expected bytes")
			}
			return result, errors.Join(err, closeErr)
		}
		switch attemptResult.mutation.Disposition {
		case cp.ComponentManagedMutationNoChange:
			result.Disposition = ManagedLifecycleOperationNoChange
		case cp.ComponentManagedMutationDesiredProjectionStaged:
			result.Disposition = ManagedLifecycleOperationChanged
		case cp.ComponentManagedMutationSafeProjectionStaged:
			result.Disposition = ManagedLifecycleOperationSafe
		}
		if closeErr != nil {
			return result, fmt.Errorf("managed lifecycle committed but session cleanup failed: %w", closeErr)
		}
		return result, nil
	}
	result.Disposition = ManagedLifecycleOperationRolledBackFailure
	return result, fmt.Errorf("managed lifecycle attempts exhausted after target drift")
}

func validateManagedLifecycleStagedMutation(budget cp.ComponentManagedBudget, plan cp.ComponentManagedPlan, mutation cp.ComponentManagedMutationResult, receipt cp.ComponentManagedMutationReceipt) error {
	if err := mutation.Validate(); err != nil {
		return fmt.Errorf("managed lifecycle mutation result: %w", err)
	}
	usage := receipt.Usage
	if usage.RowsRead > usage.CurrentItems || usage.RowsLocked != usage.CurrentItems-usage.RowsRead {
		return fmt.Errorf("managed lifecycle current items must equal rows read plus rows locked")
	}
	if usage.RowsInserted > usage.DeltaRows || usage.RowsUpdated > usage.DeltaRows-usage.RowsInserted || usage.RowsDeleted != usage.DeltaRows-usage.RowsInserted-usage.RowsUpdated {
		return fmt.Errorf("managed lifecycle delta rows must equal rows inserted, updated, and deleted")
	}
	if err := validateManagedLifecycleUsage(budget, usage); err != nil {
		return err
	}
	matchingEffects := 0
	for _, effect := range plan.Effects() {
		for _, trigger := range effect.Triggers {
			if trigger == mutation.Disposition {
				matchingEffects++
				break
			}
		}
	}
	if matchingEffects > budget.MaxEffectCalls {
		return fmt.Errorf("managed lifecycle required effect calls %d exceed budget %d", matchingEffects, budget.MaxEffectCalls)
	}
	if budget.MaxEvidenceWrites < 1 {
		return fmt.Errorf("managed lifecycle evidence-write budget is exhausted")
	}
	switch mutation.Disposition {
	case cp.ComponentManagedMutationNoChange:
		if usage.DeltaRows != 0 || usage.MutationStatements != 0 || matchingEffects != 0 {
			return fmt.Errorf("managed lifecycle mutation disposition %q contradicts actual semantic usage", mutation.Disposition)
		}
	case cp.ComponentManagedMutationDesiredProjectionStaged, cp.ComponentManagedMutationSafeProjectionStaged:
		if usage.DeltaRows == 0 && matchingEffects == 0 {
			return fmt.Errorf("managed lifecycle mutation disposition %q requires an actual row or effect change", mutation.Disposition)
		}
	}
	return nil
}

type managedLifecycleReadView struct {
	delegate cp.ComponentManagedTransactionReadStore
}

var _ cp.ComponentManagedKeysetTransactionReadStore = managedLifecycleReadView{}

func (v managedLifecycleReadView) ReadRows(ctx context.Context, query cp.ComponentManagedRowQuery) ([]cp.ComponentRow, error) {
	return v.delegate.ReadRows(ctx, query)
}

func (v managedLifecycleReadView) ReadRowsPage(ctx context.Context, query cp.ComponentManagedKeysetRowQuery) ([]cp.ComponentRow, error) {
	return v.delegate.ReadRowsPage(ctx, query)
}

func (v managedLifecycleReadView) ReadRowsKeysetPage(ctx context.Context, query cp.ComponentManagedKeysetRowQuery) (cp.ComponentManagedKeysetRowPage, error) {
	delegate, ok := v.delegate.(cp.ComponentManagedKeysetTransactionReadStore)
	if !ok {
		return cp.ComponentManagedKeysetRowPage{}, fmt.Errorf("managed lifecycle keyset exhaustion is unavailable")
	}
	return delegate.ReadRowsKeysetPage(ctx, query)
}

func (h *ManagedLifecycleHost) takePreflight(scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest) (managedLifecyclePreflight, bool) {
	key := managedLifecycleRequestKey(scope, request)
	h.preflightMu.Lock()
	defer h.preflightMu.Unlock()
	queue := h.preflights[key]
	if len(queue) == 0 {
		return managedLifecyclePreflight{}, false
	}
	value := queue[0]
	if len(queue) == 1 {
		delete(h.preflights, key)
	} else {
		h.preflights[key] = queue[1:]
	}
	return value, true
}

func validateManagedLifecycleLockedEnvelope(ambient, locked cp.ComponentManagedPlan) error {
	if !reflect.DeepEqual(ambient.RequiredCapabilities(), locked.RequiredCapabilities()) {
		return fmt.Errorf("managed lifecycle locked plan changed required capabilities")
	}
	if !reflect.DeepEqual(ambient.Effects(), locked.Effects()) {
		return fmt.Errorf("managed lifecycle locked plan changed required effects")
	}
	if !bytes.Equal(ambient.ProjectionIdentity(), locked.ProjectionIdentity()) {
		return fmt.Errorf("managed lifecycle locked plan changed projection identity")
	}
	return nil
}

type managedLifecycleEffectReceipt struct {
	Class   cp.ComponentManagedEffectClass `json:"class"`
	Receipt []byte                         `json:"receipt"`
}

func (h *ManagedLifecycleHost) applyEffects(ctx context.Context, tx *sql.Tx, operationID []byte, attempt int, scope cp.ComponentManagedScope, mutation cp.ComponentManagedMutationResult, plan cp.ComponentManagedPlan) ([]managedLifecycleEffectReceipt, []cp.ComponentManagedEffectClass, error) {
	receipts := make([]managedLifecycleEffectReceipt, 0)
	classes := make([]cp.ComponentManagedEffectClass, 0)
	for _, requirement := range plan.Effects() {
		triggered := false
		for _, trigger := range requirement.Triggers {
			triggered = triggered || trigger == mutation.Disposition
		}
		if !triggered {
			continue
		}
		adapter, ok := h.effects[requirement.Class]
		if !ok {
			return nil, nil, fmt.Errorf("managed lifecycle effect %q is not configured", requirement.Class)
		}
		request := ManagedLifecycleEffectRequest{
			OperationID: append([]byte(nil), operationID...), Attempt: attempt, Kind: h.lifecycle.Kind(), Scope: scope, Disposition: mutation.Disposition,
		}
		receipt, err := adapter.Apply(ctx, tx, request)
		if err != nil {
			return nil, nil, fmt.Errorf("apply managed lifecycle effect %q: %w", requirement.Class, err)
		}
		if len(receipt) == 0 || len(receipt) > managedLifecycleMaximumEffectReceiptBytes {
			return nil, nil, fmt.Errorf("managed lifecycle effect %q receipt must contain 1..%d bytes", requirement.Class, managedLifecycleMaximumEffectReceiptBytes)
		}
		receipts = append(receipts, managedLifecycleEffectReceipt{Class: requirement.Class, Receipt: append([]byte(nil), receipt...)})
		classes = append(classes, requirement.Class)
	}
	return receipts, classes, nil
}

func insertManagedLifecycleEvidence(ctx context.Context, tx *sql.Tx, operationID []byte, attempt int, kind cp.ComponentKind, scope cp.ComponentManagedScope, plan cp.ComponentManagedPlan, mutation cp.ComponentManagedMutationResult, receipts []managedLifecycleEffectReceipt, semanticMutationRows int) error {
	encoded, err := json.Marshal(receipts)
	if err != nil {
		return err
	}
	if len(encoded) > managedLifecycleMaximumEvidenceReceiptBytes {
		return fmt.Errorf("managed lifecycle encoded effect receipts contain %d bytes; maximum is %d", len(encoded), managedLifecycleMaximumEvidenceReceiptBytes)
	}
	projectionIdentityHash := sha256.Sum256(plan.ProjectionIdentity())
	var effectReceiptHash []byte
	if len(receipts) != 0 {
		hash := sha256.Sum256(encoded)
		effectReceiptHash = hash[:]
	}
	_, err = tx.ExecContext(ctx, "INSERT INTO `_System_ManagedComponentOperationEvidence` (`OperationId`, `Attempt`, `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
		operationID, attempt, string(kind), string(scope), plan.Fingerprint(), projectionIdentityHash[:], string(mutation.Disposition), semanticMutationRows, effectReceiptHash)
	if err != nil {
		return fmt.Errorf("insert managed lifecycle evidence: %w", err)
	}
	return nil
}

type managedLifecycleCommitProof struct {
	evidenceFound bool
	proved        bool
	verification  ManagedLifecycleVerification
}

func (h *ManagedLifecycleHost) proveCommitOutcome(
	ctx context.Context,
	operationID []byte,
	attempt int,
	scope cp.ComponentManagedScope,
	request cp.ComponentManagedLifecycleRequest,
	plan cp.ComponentManagedPlan,
	mutation cp.ComponentManagedMutationResult,
	receipts []managedLifecycleEffectReceipt,
	semanticMutationRows int,
	verifyProjection bool,
) (managedLifecycleCommitProof, error) {
	row := h.db.QueryRowContext(ctx, "SELECT `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash` FROM `_System_ManagedComponentOperationEvidence` WHERE `OperationId` = ? AND `Attempt` = ? LIMIT 1", operationID, attempt)
	var kind, actualScope, planFingerprint, projectionIdentityHash, disposition, effectReceiptHash []byte
	var actualSemanticMutationRows int
	if err := row.Scan(&kind, &actualScope, &planFingerprint, &projectionIdentityHash, &disposition, &actualSemanticMutationRows, &effectReceiptHash); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return managedLifecycleCommitProof{}, nil
		}
		return managedLifecycleCommitProof{}, fmt.Errorf("read managed lifecycle commit evidence: %w", err)
	}
	proof := managedLifecycleCommitProof{evidenceFound: true}
	expectedProjectionIdentityHash := sha256.Sum256(plan.ProjectionIdentity())
	encodedReceipts, err := json.Marshal(receipts)
	if err != nil {
		return proof, fmt.Errorf("encode managed lifecycle effect receipts: %w", err)
	}
	var expectedEffectReceiptHash []byte
	if len(receipts) != 0 {
		hash := sha256.Sum256(encodedReceipts)
		expectedEffectReceiptHash = hash[:]
	}
	if string(kind) != string(h.lifecycle.Kind()) || string(actualScope) != string(scope) || !bytes.Equal(planFingerprint, plan.Fingerprint()) || !bytes.Equal(projectionIdentityHash, expectedProjectionIdentityHash[:]) || string(disposition) != string(mutation.Disposition) || actualSemanticMutationRows != semanticMutationRows || !bytes.Equal(effectReceiptHash, expectedEffectReceiptHash) {
		return proof, fmt.Errorf("managed lifecycle commit evidence does not match the attempted operation")
	}
	if !verifyProjection {
		proof.proved = true
		return proof, nil
	}
	verification, err := h.verify(ctx, scope, request)
	proof.verification = verification
	if err != nil {
		return proof, err
	}
	if !verification.Matches || !bytes.Equal(verification.Projection.Identity, plan.ProjectionIdentity()) {
		return proof, fmt.Errorf("managed lifecycle projection does not prove the committed attempt")
	}
	expectedClasses := make([]cp.ComponentManagedEffectClass, 0)
	for _, requirement := range plan.Effects() {
		for _, trigger := range requirement.Triggers {
			if trigger == mutation.Disposition {
				expectedClasses = append(expectedClasses, requirement.Class)
				break
			}
		}
	}
	if len(receipts) != len(expectedClasses) {
		return proof, fmt.Errorf("managed lifecycle effect receipts do not prove the committed attempt")
	}
	for index, class := range expectedClasses {
		if receipts[index].Class != class {
			return proof, fmt.Errorf("managed lifecycle effect receipt class %q does not match %q", receipts[index].Class, class)
		}
		adapter := h.effects[class]
		effectRequest := ManagedLifecycleEffectRequest{
			OperationID: append([]byte(nil), operationID...), Attempt: attempt, Kind: h.lifecycle.Kind(), Scope: scope, Disposition: mutation.Disposition,
		}
		proved, err := adapter.Prove(ctx, h.db, effectRequest, append([]byte(nil), receipts[index].Receipt...))
		if err != nil {
			return proof, fmt.Errorf("prove managed lifecycle effect %q: %w", class, err)
		}
		if !proved {
			return proof, fmt.Errorf("managed lifecycle effect %q does not prove the committed attempt", class)
		}
	}
	proof.proved = true
	return proof, nil
}

func (h *ManagedLifecycleHost) ensureCoordinationTables(ctx context.Context) error {
	select {
	case <-ctx.Done():
		return fmt.Errorf("ensure managed lifecycle coordination tables: %w", ctx.Err())
	case <-h.ddlGate:
	}
	defer func() { h.ddlGate <- struct{}{} }()
	if h.ddlReady {
		return nil
	}
	statements := []string{
		"CREATE TABLE IF NOT EXISTS `_System_ManagedComponentTargetLock` (`TargetDomain` VARBINARY(255) NOT NULL, `PhysicalKey` VARBINARY(1024) NOT NULL, `CreatedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), `LastModifiedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), PRIMARY KEY (`TargetDomain`, `PhysicalKey`))",
		"CREATE TABLE IF NOT EXISTS `_System_ManagedComponentOperationEvidence` (`OperationId` VARBINARY(64) NOT NULL, `Attempt` INT NOT NULL, `ComponentKind` VARCHAR(255) NOT NULL, `Scope` VARCHAR(16) NOT NULL, `PlanFingerprint` BINARY(32) NOT NULL, `ProjectionIdentityHash` BINARY(32) NOT NULL, `MutationDisposition` VARCHAR(32) NOT NULL, `SemanticMutationRows` INT NOT NULL, `EffectReceiptHash` BINARY(32) NULL, `CreatedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), `LastModifiedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), PRIMARY KEY (`OperationId`, `Attempt`))",
	}
	for _, statement := range statements {
		if _, err := h.db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("ensure managed lifecycle coordination tables: %w", err)
		}
	}
	if err := validateManagedLifecycleTargetTable(ctx, h.db); err != nil {
		return err
	}
	if err := validateManagedLifecycleEvidenceTable(ctx, h.db); err != nil {
		return err
	}
	h.ddlReady = true
	return nil
}

// NewManagedLifecycleHost validates and snapshots one additive lifecycle.
func NewManagedLifecycleHost(db *sql.DB, registration cp.ComponentKindRegistration, lifecycle cp.ComponentManagedLifecycleRegistration, effects []ManagedLifecycleEffectAdapter) (*ManagedLifecycleHost, error) {
	return newManagedLifecycleHost(db, sqlManagedLifecycleCatalog{exec: db}, registration, lifecycle, effects)
}

func newManagedLifecycleHost(db *sql.DB, catalog managedLifecycleCatalog, registration cp.ComponentKindRegistration, lifecycle cp.ComponentManagedLifecycleRegistration, effects []ManagedLifecycleEffectAdapter) (*ManagedLifecycleHost, error) {
	if db == nil {
		return nil, fmt.Errorf("managed lifecycle database is required")
	}
	if catalog == nil {
		return nil, fmt.Errorf("managed lifecycle catalog is required")
	}
	registration = cloneComponentKindRegistration(registration)
	registration.ManagedProvider = nil
	if err := validateManagedLifecycleBaseRegistration(registration, lifecycle); err != nil {
		return nil, err
	}
	if len(registration.Contract.Kind) > 255 {
		return nil, fmt.Errorf("managed lifecycle component kind exceeds 255 bytes")
	}
	if err := ValidateManagedLifecycleEffectAdapters(effects); err != nil {
		return nil, err
	}
	adapters := make(map[cp.ComponentManagedEffectClass]ManagedLifecycleEffectAdapter, len(effects))
	for _, adapter := range effects {
		adapters[adapter.Class] = adapter
	}
	declared := make(map[cp.ComponentManagedEffectClass]struct{}, len(lifecycle.EffectClasses()))
	for _, class := range lifecycle.EffectClasses() {
		declared[class] = struct{}{}
	}
	selected := make(map[cp.ComponentManagedEffectClass]ManagedLifecycleEffectAdapter, len(declared))
	for class, adapter := range adapters {
		if _, ok := declared[class]; ok {
			selected[class] = adapter
		}
	}
	host := &ManagedLifecycleHost{
		db: db, catalog: catalog, registration: registration, lifecycle: lifecycle, effects: selected,
		preflights: make(map[[sha256.Size]byte][]managedLifecyclePreflight), ddlGate: make(chan struct{}, 1),
	}
	host.ddlGate <- struct{}{}
	return host, nil
}

// PreflightItem plans one item through a fresh bounded reader and retains the
// immutable envelope for the later locked attempt.
func (h *ManagedLifecycleHost) PreflightItem(ctx context.Context, request cp.ComponentManagedRequest) error {
	if err := h.validateItemRequest(request); err != nil {
		return err
	}
	lifecycleRequest := cp.NewComponentManagedItemLifecycleRequest(request)
	return h.preflight(ctx, cp.ComponentManagedScopeItem, lifecycleRequest)
}

// PreflightSet plans one complete owner-scoped desired set without mutation.
func (h *ManagedLifecycleHost) PreflightSet(ctx context.Context, request cp.ComponentManagedSetRequest, desiredBusinessKeys []string) error {
	if err := h.validateSetRequest(request); err != nil {
		return err
	}
	if err := h.validateDesiredSetItems(desiredBusinessKeys); err != nil {
		return err
	}
	lifecycleRequest, err := cp.NewComponentManagedSetLifecycleRequest(request, desiredBusinessKeys)
	if err != nil {
		return err
	}
	return h.preflight(ctx, cp.ComponentManagedScopeSet, lifecycleRequest)
}

func (h *ManagedLifecycleHost) preflight(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest) error {
	budget := h.lifecycle.Budget()
	if err := validateManagedLifecycleUsage(budget, cp.ComponentManagedUsage{DesiredItems: managedLifecycleDesiredItems(scope, request)}); err != nil {
		return err
	}
	boundedCtx, cancel := context.WithTimeout(ctx, budget.MaxElapsed)
	defer cancel()
	reader, err := newManagedLifecycleBoundedReadStore(boundedCtx, h.db, h.catalog, h.registration.Contract.WriteSet, h.lifecycle.ReadSet(), budget)
	if err != nil {
		return err
	}
	plan, err := h.lifecycle.Lifecycle().Plan(boundedCtx, scope, request.Clone(), reader)
	if err != nil {
		return fmt.Errorf("preflight managed lifecycle %s: %w", h.lifecycle.Kind(), err)
	}
	if err := reader.failureErr(); err != nil {
		return fmt.Errorf("preflight managed lifecycle %s ignored a structured-store failure: %w", h.lifecycle.Kind(), err)
	}
	if err := boundedCtx.Err(); err != nil {
		return fmt.Errorf("preflight managed lifecycle %s exceeded elapsed budget: %w", h.lifecycle.Kind(), err)
	}
	if err := h.validateManagedLifecyclePlan(plan); err != nil {
		return fmt.Errorf("preflight managed lifecycle %s plan: %w", h.lifecycle.Kind(), err)
	}
	key := managedLifecycleRequestKey(scope, request)
	h.preflightMu.Lock()
	h.preflights[key] = append(h.preflights[key], managedLifecyclePreflight{plan: plan})
	h.preflightMu.Unlock()
	return nil
}

func managedLifecycleDesiredItems(scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest) int {
	if scope == cp.ComponentManagedScopeSet {
		return len(request.DesiredBusinessKeys)
	}
	return 1
}

func validateManagedLifecycleHostPlan(plan cp.ComponentManagedPlan) error {
	if size := len(plan.ProjectionIdentity()); size > managedLifecycleMaximumProjectionIdentityBytes {
		return fmt.Errorf("managed lifecycle projection identity contains %d bytes; maximum is %d", size, managedLifecycleMaximumProjectionIdentityBytes)
	}
	return nil
}

func (h *ManagedLifecycleHost) validatePlanEffectAdapters(plan cp.ComponentManagedPlan) error {
	for _, requirement := range plan.Effects() {
		if _, ok := h.effects[requirement.Class]; !ok {
			return fmt.Errorf("managed lifecycle effect %q is not configured", requirement.Class)
		}
	}
	return nil
}

// VerifyItem projects one item from a fresh bounded reader.
func (h *ManagedLifecycleHost) VerifyItem(ctx context.Context, request cp.ComponentManagedRequest) (ManagedLifecycleVerification, error) {
	if err := h.validateItemRequest(request); err != nil {
		return ManagedLifecycleVerification{}, err
	}
	return h.verify(ctx, cp.ComponentManagedScopeItem, cp.NewComponentManagedItemLifecycleRequest(request))
}

// VerifySet projects one complete desired set from a fresh bounded reader.
func (h *ManagedLifecycleHost) VerifySet(ctx context.Context, request cp.ComponentManagedSetRequest, desiredBusinessKeys []string) (ManagedLifecycleVerification, error) {
	if err := h.validateSetRequest(request); err != nil {
		return ManagedLifecycleVerification{}, err
	}
	if err := h.validateDesiredSetItems(desiredBusinessKeys); err != nil {
		return ManagedLifecycleVerification{}, err
	}
	lifecycleRequest, err := cp.NewComponentManagedSetLifecycleRequest(request, desiredBusinessKeys)
	if err != nil {
		return ManagedLifecycleVerification{}, err
	}
	return h.verify(ctx, cp.ComponentManagedScopeSet, lifecycleRequest)
}

func (h *ManagedLifecycleHost) verify(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest) (ManagedLifecycleVerification, error) {
	budget := h.lifecycle.Budget()
	if err := validateManagedLifecycleUsage(budget, cp.ComponentManagedUsage{DesiredItems: managedLifecycleDesiredItems(scope, request)}); err != nil {
		return ManagedLifecycleVerification{}, err
	}
	boundedCtx, cancel := context.WithTimeout(ctx, budget.MaxElapsed)
	defer cancel()
	// TiDB does not support the read-only transaction characteristic.
	readTx, err := h.db.BeginTx(boundedCtx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
	if err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("begin fresh verification transaction for managed lifecycle %s: %w", h.lifecycle.Kind(), err)
	}
	defer readTx.Rollback()
	catalog := h.catalog
	if _, ok := catalog.(sqlManagedLifecycleCatalog); ok {
		catalog = sqlManagedLifecycleCatalog{exec: readTx}
	}
	reader, err := newManagedLifecycleBoundedReadStore(boundedCtx, readTx, catalog, h.registration.Contract.WriteSet, h.lifecycle.ReadSet(), budget)
	if err != nil {
		return ManagedLifecycleVerification{}, err
	}
	projection, err := h.lifecycle.Lifecycle().Project(boundedCtx, scope, request.Clone(), managedLifecycleReadView{delegate: reader})
	if err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("verify managed lifecycle %s: %w", h.lifecycle.Kind(), err)
	}
	if err := reader.failureErr(); err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("verify managed lifecycle %s ignored a structured-store failure: %w", h.lifecycle.Kind(), err)
	}
	if err := boundedCtx.Err(); err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("verify managed lifecycle %s exceeded elapsed budget: %w", h.lifecycle.Kind(), err)
	}
	if err := projection.Validate(); err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("verify managed lifecycle %s projection: %w", h.lifecycle.Kind(), err)
	}
	if err := readTx.Rollback(); err != nil {
		return ManagedLifecycleVerification{}, fmt.Errorf("close fresh verification transaction for managed lifecycle %s: %w", h.lifecycle.Kind(), err)
	}
	return ManagedLifecycleVerification{Projection: cloneManagedLifecycleProjection(projection), Matches: bytes.Equal(projection.Expected, projection.Live)}, nil
}

func (h *ManagedLifecycleHost) validateItemRequest(request cp.ComponentManagedRequest) error {
	if err := validateCanonicalComponentIdentity("pack name", request.PackName); err != nil {
		return err
	}
	if request.Component.Kind != h.registration.Contract.Kind {
		return fmt.Errorf("managed lifecycle component kind %q does not match host kind %q", request.Component.Kind, h.registration.Contract.Kind)
	}
	if err := validateCanonicalComponentIdentity("business key", request.Component.BusinessKey); err != nil {
		return err
	}
	if err := h.validateOwner(request.SourceOwner); err != nil {
		return err
	}
	if _, ok := h.registration.Contract.Manageabilities[request.Component.Manageability]; !ok {
		return fmt.Errorf("managed lifecycle %s does not support manageability %q", h.lifecycle.Kind(), request.Component.Manageability)
	}
	if len(request.Payload) == 0 || (len(request.Component.Payload) != 0 && !bytes.Equal(request.Component.Payload, request.Payload)) {
		return fmt.Errorf("managed lifecycle payload is required; an inline manifest payload must match it")
	}
	if len(request.ExpectedProjection) == 0 {
		return fmt.Errorf("managed lifecycle expected projection is required")
	}
	return nil
}

func (h *ManagedLifecycleHost) validateSetRequest(request cp.ComponentManagedSetRequest) error {
	if err := validateCanonicalComponentIdentity("pack name", request.PackName); err != nil {
		return err
	}
	if request.Kind != h.registration.Contract.Kind {
		return fmt.Errorf("managed lifecycle set kind %q does not match host kind %q", request.Kind, h.registration.Contract.Kind)
	}
	for _, capability := range h.lifecycle.Capabilities() {
		if capability == cp.ComponentManagedCapabilityExactSet {
			return h.validateOwner(request.SourceOwner)
		}
	}
	return fmt.Errorf("managed lifecycle %s does not declare capability %q required for set operations", h.lifecycle.Kind(), cp.ComponentManagedCapabilityExactSet)
}

func (h *ManagedLifecycleHost) validateDesiredSetItems(desiredBusinessKeys []string) error {
	return validateManagedLifecycleUsage(h.lifecycle.Budget(), cp.ComponentManagedUsage{DesiredItems: len(desiredBusinessKeys)})
}

func (h *ManagedLifecycleHost) validateOwner(owner string) error {
	switch h.registration.Contract.OwnerDiscriminator {
	case cp.ComponentOwnerStrategyOwnerPack:
		if owner != "" {
			return fmt.Errorf("pack-owned managed lifecycle cannot carry source owner %q", owner)
		}
	case cp.ComponentOwnerStrategySource:
		return validateCanonicalComponentIdentity("source owner", owner)
	default:
		return fmt.Errorf("managed lifecycle has unsupported owner discriminator %q", h.registration.Contract.OwnerDiscriminator)
	}
	return nil
}

func managedLifecycleRequestKey(scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest) [sha256.Size]byte {
	hash := sha256.New()
	write := func(value []byte) {
		var size [8]byte
		binary.BigEndian.PutUint64(size[:], uint64(len(value)))
		_, _ = hash.Write(size[:])
		_, _ = hash.Write(value)
	}
	write([]byte(scope))
	if request.Item != nil {
		write([]byte(request.Item.PackName))
		write([]byte(request.Item.SourceOwner))
		write([]byte(request.Item.Component.Kind))
		write([]byte(request.Item.Component.BusinessKey))
		write([]byte(request.Item.Component.Manageability))
		write([]byte(request.Item.Component.Path))
		write(request.Item.Component.Payload)
		for _, dependency := range request.Item.Component.DependsOn {
			write([]byte(dependency.Kind))
			write([]byte(dependency.BusinessKey))
			if dependency.Required {
				write([]byte{1})
			} else {
				write([]byte{0})
			}
		}
		write(request.Item.Payload)
		write(request.Item.ExpectedProjection)
	}
	if request.Set != nil {
		write([]byte(request.Set.PackName))
		write([]byte(request.Set.SourceOwner))
		write([]byte(request.Set.Kind))
	}
	for _, key := range request.DesiredBusinessKeys {
		write([]byte(key))
	}
	var out [sha256.Size]byte
	copy(out[:], hash.Sum(nil))
	return out
}

func cloneManagedLifecycleProjection(projection cp.ComponentManagedProjection) cp.ComponentManagedProjection {
	return cp.ComponentManagedProjection{
		Identity: append([]byte(nil), projection.Identity...),
		Expected: append([]byte(nil), projection.Expected...),
		Live:     append([]byte(nil), projection.Live...),
	}
}

func sortedManagedLifecycleEffectClasses(values map[cp.ComponentManagedEffectClass]ManagedLifecycleEffectAdapter) []cp.ComponentManagedEffectClass {
	out := make([]cp.ComponentManagedEffectClass, 0, len(values))
	for value := range values {
		out = append(out, value)
	}
	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	return out
}