package bootstate

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/shared/pkg/constants"
	contractcontrolplane "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

const (
	bootstrapStateTable                            = "_System_BootstrapState"
	schemaBootstrapStateKey                        = "schema_initialize"
	controlPlaneApplyAllStateKey                   = "control_plane_apply_all"
	controlPlaneApplyAllFinalizationReadyPhase     = "final-verified"
	controlPlaneApplyAllFinalizationReadyStatus    = "ready"
	controlPlaneApplyAllFinalLockFailureMessage    = "control-plane apply lock final gate: control-plane apply lock scope is no longer active"
	connectorPushLegacyMirrorBackfillStateKey      = "connector_push_legacy_mirror_backfill"
	webhookSecretKeyReadProtectionBackfillStateKey = "webhook_secretkey_read_protection_backfill"
	connectorEventEnqueueOrdinalBackfillStateKey   = "connector_event_enqueue_ordinal_backfill"
	connectorEventSourceLaneBackfillStateKey       = "connector_event_source_lane_backfill"
	connectorEventExecutionOwnerFenceStateKey      = "connector_event_execution_owner_fence"
	listViewLegacyCutoverStateKey                  = "listview_legacy_cutover"
	controlPlaneApplyAllRunningPhase               = "apply-all"
	recordEventOutboxObjectAPIName                 = "_System_RecordEventOutbox"
	blockedObjectPermissionValue                   = "BLOCKED"
	runtimeTokenObjectPermissionsNotReadyReason    = "live _System_Object metadata for _System_FileUploadOperation and " +
		"_System_RecordEventOutbox must set RequiredReadPermission=BLOCKED and RequiredWritePermission=BLOCKED; " +
		"resolve customized permission drift and rerun setup apply"
)

type bootstrapStateSnapshot struct {
	RunID  string
	Phase  string
	Status string
}

// ControlPlaneAutoStartAction re-exports the contract type so callers of this
// package keep the same symbol surface.
type ControlPlaneAutoStartAction = contractcontrolplane.ControlPlaneAutoStartAction

const (
	ControlPlaneAutoStartApply            = contractcontrolplane.ControlPlaneAutoStartApply
	ControlPlaneAutoStartSkip             = contractcontrolplane.ControlPlaneAutoStartSkip
	ControlPlaneAutoStartHaltForMigration = contractcontrolplane.ControlPlaneAutoStartHaltForMigration
)

// ControlPlaneAutoStartDecision re-exports the contract type so callers of this
// package keep the same symbol surface.
type ControlPlaneAutoStartDecision = contractcontrolplane.ControlPlaneAutoStartDecision

func ensureBootstrapStateTable(rawDB *sql.DB) error {
	ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
		`+"`StateKey` VARCHAR(128) NOT NULL PRIMARY KEY,"+`
		`+"`RunID` VARCHAR(64) NOT NULL,"+`
		`+"`Phase` VARCHAR(128) NOT NULL,"+`
		`+"`Status` VARCHAR(32) NOT NULL,"+`
		`+"`LastError` TEXT NULL,"+`
		`+"`CreatedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,"+`
		`+"`LastModifiedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"+`
	) %s`, bootstrapStateTable, constants.DefaultTableOptions)
	_, err := rawDB.Exec(ddl)
	return err
}

func bootstrapStateTablePresent(rawDB *sql.DB) (bool, error) {
	var present int
	if err := rawDB.QueryRow(
		"SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?",
		bootstrapStateTable,
	).Scan(&present); err != nil {
		return false, err
	}
	return present > 0, nil
}

func loadBootstrapStateForKey(rawDB *sql.DB, stateKey string) (*bootstrapStateSnapshot, error) {
	row := rawDB.QueryRow(
		fmt.Sprintf("SELECT `RunID`, `Phase`, `Status` FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		stateKey,
	)
	var s bootstrapStateSnapshot
	if err := row.Scan(&s.RunID, &s.Phase, &s.Status); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return &s, nil
}

func loadBootstrapState(rawDB *sql.DB) (*bootstrapStateSnapshot, error) {
	return loadBootstrapStateForKey(rawDB, schemaBootstrapStateKey)
}

func bootstrapStateCompleted(state *bootstrapStateSnapshot) bool {
	return state != nil && strings.EqualFold(state.Phase, "completed") && strings.EqualFold(state.Status, "completed")
}

// IsSchemaBootstrapped reports whether the L0 schema bootstrap has completed
// successfully on this database. It is a read-only probe over
// _System_BootstrapState. This is intentionally narrower than full setup apply
// completion; container first-boot gates must use IsControlPlaneApplyAllCompleted
// instead. It returns (false, nil) for:
//   - a greenfield database (state table absent),
//   - an in-progress, failed, or partial bootstrap (Phase/Status != "completed").
//
// It returns (false, err) only on an unexpected database error.
func IsSchemaBootstrapped(rawDB *sql.DB) (bool, error) {
	// Detect greenfield explicitly: on a fresh database the state table does not
	// exist yet, which would otherwise surface as a "table doesn't exist" query
	// error rather than sql.ErrNoRows.
	present, err := bootstrapStateTablePresent(rawDB)
	if err != nil {
		return false, err
	}
	if !present {
		return false, nil
	}
	state, err := loadBootstrapState(rawDB)
	if err != nil {
		return false, err
	}
	if state == nil {
		return false, nil
	}
	return bootstrapStateCompleted(state), nil
}

// IsControlPlaneApplyAllCompleted reports whether the full control-plane apply
// sequence and the release-scoped warm-upgrade repairs have completed
// successfully. This is the only projection that container auto startup may
// use to skip first-boot apply; schema-only, L0-only, partial bootstrap,
// completed markers missing release-scoped repair receipts, or missing required
// live schema or metadata deliberately do not satisfy it.
func IsControlPlaneApplyAllCompleted(rawDB *sql.DB) (bool, error) {
	present, err := bootstrapStateTablePresent(rawDB)
	if err != nil {
		return false, err
	}
	if !present {
		return false, nil
	}
	state, err := loadBootstrapStateForKey(rawDB, controlPlaneApplyAllStateKey)
	if err != nil {
		return false, err
	}
	if state == nil {
		return false, nil
	}
	if !bootstrapStateCompleted(state) {
		return false, nil
	}
	connectorBackfillCompleted, err := IsConnectorPushLegacyMirrorBackfillCompleted(rawDB)
	if err != nil || !connectorBackfillCompleted {
		return connectorBackfillCompleted, err
	}
	webhookBackfillCompleted, err := IsWebhookSecretKeyReadProtectionBackfillCompleted(rawDB)
	if err != nil || !webhookBackfillCompleted {
		return webhookBackfillCompleted, err
	}
	connectorEventOrdinalBackfillCompleted, err := IsConnectorEventEnqueueOrdinalBackfillCompleted(rawDB)
	if err != nil || !connectorEventOrdinalBackfillCompleted {
		return connectorEventOrdinalBackfillCompleted, err
	}
	connectorEventSourceLaneBackfillCompleted, err := IsConnectorEventSourceLaneBackfillCompleted(rawDB)
	if err != nil || !connectorEventSourceLaneBackfillCompleted {
		return connectorEventSourceLaneBackfillCompleted, err
	}
	connectorEventExecutionOwnerFenceCompleted, err := IsConnectorEventExecutionOwnerFenceCompleted(rawDB)
	if err != nil || !connectorEventExecutionOwnerFenceCompleted {
		return connectorEventExecutionOwnerFenceCompleted, err
	}
	listViewLegacyCutoverCompleted, err := IsListViewLegacyCutoverCompleted(rawDB)
	if err != nil || !listViewLegacyCutoverCompleted {
		return listViewLegacyCutoverCompleted, err
	}
	sessionCurrentBearerReady, err := sessionCurrentBearerTokenColumnReady(context.Background(), rawDB)
	if err != nil || !sessionCurrentBearerReady {
		return sessionCurrentBearerReady, err
	}
	outboundGenerationReady, err := runtimeOutboundIntentGenerationReady(context.Background(), rawDB)
	if err != nil || !outboundGenerationReady {
		return outboundGenerationReady, err
	}
	runtimeTokenPermissionsReady, err := runtimeTokenObjectPermissionsReady(context.Background(), rawDB)
	if err != nil || !runtimeTokenPermissionsReady {
		return runtimeTokenPermissionsReady, err
	}
	return true, nil
}

// runtimeOutboundIntentGenerationReady is the live warm-upgrade fence for the
// durable outbound wake cursor. An older completed ApplyAll marker predates the
// table, so startup must force ApplyAll until the canonical one-row-per-scope
// primary key exists. Runtime performs the complete column/index verification
// again before enabling the consumer.
func runtimeOutboundIntentGenerationReady(ctx context.Context, rawDB *sql.DB) (bool, error) {
	var indexColumns int
	var exactColumns int
	err := rawDB.QueryRowContext(ctx, `
SELECT COUNT(*),
       COALESCE(SUM(CASE
         WHEN NON_UNIQUE = 0
          AND SEQ_IN_INDEX = 1
          AND LOWER(COLUMN_NAME) = LOWER(?)
          AND SUB_PART IS NULL
         THEN 1 ELSE 0 END), 0)
  FROM INFORMATION_SCHEMA.STATISTICS
 WHERE TABLE_SCHEMA = DATABASE()
   AND LOWER(TABLE_NAME) = LOWER(?)
   AND INDEX_NAME = 'PRIMARY'`,
		"ScopeDigest", constants.TableRuntimeOutboundIntentGeneration,
	).Scan(&indexColumns, &exactColumns)
	if err != nil {
		return false, fmt.Errorf("runtime outbound intent generation readiness: %w", err)
	}
	return indexColumns == 1 && exactColumns == 1, nil
}

// sessionCurrentBearerTokenColumnReady reports whether the physical session
// column required by every SessionRepository read and write has its exact TiDB
// shape. The live shape, rather than a historical receipt, is the convergence
// proof for this additive nullable column.
func sessionCurrentBearerTokenColumnReady(ctx context.Context, rawDB *sql.DB) (bool, error) {
	var columns int
	if err := rawDB.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE()
   AND TABLE_NAME = ?
   AND LOWER(COLUMN_NAME) = LOWER(?)
   AND LOWER(COLUMN_TYPE) = 'varbinary(3000)'`,
		constants.TableSession,
		constants.FieldSysSession_CurrentBearerToken,
	).Scan(&columns); err != nil {
		return false, fmt.Errorf("session current bearer column readiness: %w", err)
	}
	return columns == 1, nil
}

// runtimeTokenObjectPermissionsReady verifies the live metadata that the
// permission service actually consumes. Catalog declarations alone are not a
// convergence proof for warm upgrades: an older completed ApplyAll marker can
// predate the declarations, and customized object rows preserve both permission
// fields during managed-schema reconciliation.
func runtimeTokenObjectPermissionsReady(ctx context.Context, rawDB *sql.DB) (bool, error) {
	rows, err := rawDB.QueryContext(ctx, `
SELECT ApiName, RequiredReadPermission, RequiredWritePermission
  FROM _System_Object
 WHERE ApiName IN (?, ?)
   AND COALESCE(IsDeleted, 0) = 0`,
		constants.TableFileUploadOperation,
		recordEventOutboxObjectAPIName,
	)
	if err != nil {
		return false, fmt.Errorf("runtime token object permission readiness: %w", err)
	}
	defer rows.Close()

	required := map[string]struct{}{
		constants.TableFileUploadOperation: {},
		recordEventOutboxObjectAPIName:     {},
	}
	seen := make(map[string]struct{}, len(required))
	for rows.Next() {
		var apiName string
		var readPermission sql.NullString
		var writePermission sql.NullString
		if err := rows.Scan(&apiName, &readPermission, &writePermission); err != nil {
			return false, fmt.Errorf("runtime token object permission readiness: %w", err)
		}
		if _, ok := required[apiName]; !ok {
			return false, nil
		}
		if _, duplicate := seen[apiName]; duplicate {
			return false, nil
		}
		if !readPermission.Valid || readPermission.String != blockedObjectPermissionValue ||
			!writePermission.Valid || writePermission.String != blockedObjectPermissionValue {
			return false, nil
		}
		seen[apiName] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("runtime token object permission readiness: %w", err)
	}
	return len(seen) == len(required), nil
}

// IsConnectorPushLegacyMirrorBackfillCompleted reports whether this release's
// warm-upgrade materialization of legacy PushToConnector mirror evidence has run
// at least once. It is a separate receipt so already-bootstrapped databases from
// a prior head cannot take the start-only fast path and skip the backfill.
func IsConnectorPushLegacyMirrorBackfillCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, connectorPushLegacyMirrorBackfillStateKey)
	if err != nil {
		return false, err
	}
	return bootstrapStateCompleted(state), nil
}

// IsWebhookSecretKeyReadProtectionBackfillCompleted reports whether this
// release's warm-upgrade protection of legacy _System_Webhook.SecretKey field
// metadata has run at least once. It gates the start-only path so databases
// with an older completed control_plane_apply_all marker cannot skip the
// backfill that replaces the old runtime hard block.
func IsWebhookSecretKeyReadProtectionBackfillCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, webhookSecretKeyReadProtectionBackfillStateKey)
	if err != nil {
		return false, err
	}
	return bootstrapStateCompleted(state), nil
}

// ClassifyControlPlaneAutoStart is the single fail-closed classifier for the
// container auto-start apply decision. Callers that may mutate on "not
// completed" must use this under the control-plane apply lock rather than
// composing independent shell probes:
//   - completed full ApplyAll marker with release receipts and required live
//     schema: skip mutation and start;
//   - current-release running/failed/incomplete marker, greenfield, or partial
//     schema state: run/retry ApplyAll;
//   - completed schema bootstrap with no full ApplyAll marker: halt for an
//     explicit one-time migration apply.
//
// Any read error is returned as an error, never folded into a mutating default.
func ClassifyControlPlaneAutoStart(rawDB *sql.DB) (ControlPlaneAutoStartDecision, error) {
	present, err := bootstrapStateTablePresent(rawDB)
	if err != nil {
		return ControlPlaneAutoStartDecision{}, err
	}
	if !present {
		return ControlPlaneAutoStartDecision{
			Action: ControlPlaneAutoStartApply,
			Reason: "bootstrap state table absent (greenfield)",
		}, nil
	}

	applyAll, err := loadBootstrapStateForKey(rawDB, controlPlaneApplyAllStateKey)
	if err != nil {
		return ControlPlaneAutoStartDecision{}, err
	}
	if applyAll != nil {
		if bootstrapStateCompleted(applyAll) {
			backfillCompleted, err := IsConnectorPushLegacyMirrorBackfillCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !backfillCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "connector push legacy mirror backfill marker is incomplete",
				}, nil
			}
			webhookBackfillCompleted, err := IsWebhookSecretKeyReadProtectionBackfillCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !webhookBackfillCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "webhook SecretKey read-protection backfill marker is incomplete",
				}, nil
			}
			connectorEventOrdinalBackfillCompleted, err := IsConnectorEventEnqueueOrdinalBackfillCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !connectorEventOrdinalBackfillCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "connector event enqueue ordinal backfill marker is incomplete",
				}, nil
			}
			connectorEventSourceLaneBackfillCompleted, err := IsConnectorEventSourceLaneBackfillCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !connectorEventSourceLaneBackfillCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "connector event source-lane backfill marker is incomplete",
				}, nil
			}
			connectorEventExecutionOwnerFenceCompleted, err := IsConnectorEventExecutionOwnerFenceCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !connectorEventExecutionOwnerFenceCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "connector event execution-owner fence is incomplete",
				}, nil
			}
			listViewLegacyCutoverCompleted, err := IsListViewLegacyCutoverCompleted(rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !listViewLegacyCutoverCompleted {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "list-view legacy cutover marker is incomplete",
				}, nil
			}
			sessionCurrentBearerReady, err := sessionCurrentBearerTokenColumnReady(context.Background(), rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !sessionCurrentBearerReady {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "session CurrentBearerToken live column is missing or incompatible",
				}, nil
			}
			outboundGenerationReady, err := runtimeOutboundIntentGenerationReady(context.Background(), rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !outboundGenerationReady {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: "Runtime outbound-intent generation table is missing or incompatible",
				}, nil
			}
			runtimeTokenPermissionsReady, err := runtimeTokenObjectPermissionsReady(context.Background(), rawDB)
			if err != nil {
				return ControlPlaneAutoStartDecision{}, err
			}
			if !runtimeTokenPermissionsReady {
				return ControlPlaneAutoStartDecision{
					Action: ControlPlaneAutoStartApply,
					Reason: runtimeTokenObjectPermissionsNotReadyReason,
				}, nil
			}
			return ControlPlaneAutoStartDecision{
				Action: ControlPlaneAutoStartSkip,
				Reason: "full ApplyAll marker, release repairs, and required live invariants satisfied",
			}, nil
		}
		return ControlPlaneAutoStartDecision{
			Action: ControlPlaneAutoStartApply,
			Reason: fmt.Sprintf("full ApplyAll marker is %s/%s", applyAll.Phase, applyAll.Status),
		}, nil
	}

	schemaState, err := loadBootstrapState(rawDB)
	if err != nil {
		return ControlPlaneAutoStartDecision{}, err
	}
	if bootstrapStateCompleted(schemaState) {
		return ControlPlaneAutoStartDecision{
			Action: ControlPlaneAutoStartHaltForMigration,
			Reason: "schema bootstrap completed but full ApplyAll marker is absent",
		}, nil
	}
	if schemaState == nil {
		return ControlPlaneAutoStartDecision{
			Action: ControlPlaneAutoStartApply,
			Reason: "no full ApplyAll marker and no schema bootstrap marker",
		}, nil
	}
	return ControlPlaneAutoStartDecision{
		Action: ControlPlaneAutoStartApply,
		Reason: fmt.Sprintf("no full ApplyAll marker and schema bootstrap is %s/%s", schemaState.Phase, schemaState.Status),
	}, nil
}

// IsLegacyUnmigrated reports whether the database was prepared under a PRIOR
// release that predates the control_plane_apply_all marker and has not yet been
// migrated to this release. It is read-only and true iff:
//   - the control_plane_apply_all marker row is ABSENT (under the current release
//     every ApplyAll stamps it - running - BEFORE doing any work, so an absent
//     row means no current-release apply has run on this database; a partial or
//     failed current-release apply leaves a running/failed row, not an absent
//     one), AND
//   - the schema bootstrap (schema_initialize) is completed (so the database
//     holds real prepared state - it is not greenfield).
//
// Such a database must neither auto-start start-only nor be auto-force-applied on
// restart:
//   - auto start-only is unsafe: the current release's apply may include repairs
//     or convergence this database has never run, and pod-readiness
//     controlplane.Verify is not a complete substitute for apply (e.g. profile
//     permission coverage is deep-only), so trusting it could serve stale state;
//   - auto force-apply is also wrong: re-coupling restart to a full ApplyAll for
//     an already-serving deployment is exactly what #1590 removes.
//
// The resolution is a one-time explicit operator migration: the operator runs
// `setup apply` once (which stamps the marker), and thereafter restarts are
// start-only. This helper is a diagnostic projection of
// ClassifyControlPlaneAutoStart; auto-start enforcement happens under the
// apply-if-needed lock.
func IsLegacyUnmigrated(rawDB *sql.DB) (bool, error) {
	decision, err := ClassifyControlPlaneAutoStart(rawDB)
	if err != nil {
		return false, err
	}
	return decision.Action == ControlPlaneAutoStartHaltForMigration, nil
}

// MarkControlPlaneApplyAllRunning unconditionally stamps the full-ApplyAll marker
// to running, overwriting any prior state - including a completed marker. Every
// MUTATING ApplyAll attempt calls this before doing work, so the marker reflects
// THIS attempt and a failed mutating apply leaves a failed (never a stale
// completed) marker. Preserving a prior completed marker across a mutating
// reapply would be unsafe: a re-apply that fails after partial state changes
// would keep the completed marker, and a later auto restart would skip apply as
// if the database were clean and serve partial/stale state (readiness verify is
// not a complete substitute for apply). The auto first-boot path skips BEFORE
// calling this when the marker is already completed, so it never overwrites a
// completed marker with running on a no-op restart.
func MarkControlPlaneApplyAllRunning(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("apply-all-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, controlPlaneApplyAllStateKey, runID, controlPlaneApplyAllRunningPhase, "running", nil)
}

// MarkControlPlaneApplyAllInvalidated downgrades the full-ApplyAll completion
// marker before a non-ApplyAll control-plane mutator writes. A completed marker
// is a restart convergence proof for the last full ApplyAll only; once an
// operator runs a narrower mutator (apply-source-convergence, install-pack,
// repair, etc.), the next auto restart must run a full ApplyAll before trusting
// the marker again. Only MarkControlPlaneApplyAllCompleted may restore completed.
func MarkControlPlaneApplyAllInvalidated(rawDB *sql.DB, operation string) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	operation = strings.TrimSpace(operation)
	if operation == "" {
		operation = "control-plane mutation"
	}
	runID := fmt.Sprintf("apply-all-invalidated-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(
		rawDB,
		controlPlaneApplyAllStateKey,
		runID,
		"invalidated",
		"stale",
		fmt.Errorf("full ApplyAll completion marker invalidated by %s; run setup apply to restore completed", operation),
	)
}

func MarkControlPlaneApplyAllFailed(rawDB *sql.DB, applyErr error) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("apply-all-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, controlPlaneApplyAllStateKey, runID, "failed", "failed", applyErr)
}

// IsControlPlaneApplyAllFinalizationRecoverable reports whether a prior full
// ApplyAll reached its terminal verification boundary but could not stamp the
// completed marker. It deliberately recognizes only two control-plane-owned
// proofs:
//   - the typed final-verified/ready receipt written by current releases; or
//   - the exact failed marker emitted by the immediately preceding release when
//     its session-scoped advisory-lock connection was retired at the final gate.
//
// A generic failed/running/invalidated marker is never recoverable. Callers must
// still reacquire the apply lock and re-verify the exact immutable candidate and
// every terminal readiness gate before marking ApplyAll completed.
func IsControlPlaneApplyAllFinalizationRecoverable(rawDB *sql.DB) (bool, error) {
	present, err := bootstrapStateTablePresent(rawDB)
	if err != nil || !present {
		return false, err
	}
	row := rawDB.QueryRow(
		fmt.Sprintf("SELECT `Phase`, `Status`, `LastError` FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		controlPlaneApplyAllStateKey,
	)
	var phase string
	var status string
	var lastError sql.NullString
	if err := row.Scan(&phase, &status, &lastError); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	if strings.EqualFold(phase, controlPlaneApplyAllFinalizationReadyPhase) &&
		strings.EqualFold(status, controlPlaneApplyAllFinalizationReadyStatus) {
		return true, nil
	}
	if !strings.EqualFold(phase, "failed") || !strings.EqualFold(status, "failed") || !lastError.Valid {
		return false, nil
	}
	var payload struct {
		Message string `json:"message"`
	}
	if err := json.Unmarshal([]byte(lastError.String), &payload); err != nil {
		return false, nil
	}
	return payload.Message == controlPlaneApplyAllFinalLockFailureMessage, nil
}

// MarkControlPlaneApplyAllFinalizationReady writes the durable boundary between
// a successfully verified full ApplyAll and its short completion commit. It is
// not a serving-ready marker: Runtime continues to fail closed until the caller
// reacquires the apply lock, re-verifies the exact candidate and terminal gates,
// and calls MarkControlPlaneApplyAllCompleted.
func MarkControlPlaneApplyAllFinalizationReady(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	if err := requireControlPlaneApplyAllCompletionLiveInvariants(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("apply-all-final-verified-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(
		rawDB,
		controlPlaneApplyAllStateKey,
		runID,
		controlPlaneApplyAllFinalizationReadyPhase,
		controlPlaneApplyAllFinalizationReadyStatus,
		nil,
	)
}

func MarkControlPlaneApplyAllCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	if err := requireControlPlaneApplyAllCompletionLiveInvariants(rawDB); err != nil {
		return err
	}
	if err := markConnectorEventExecutionOwnerFenceCompleted(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("apply-all-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, controlPlaneApplyAllStateKey, runID, "completed", "completed", nil)
}

func requireControlPlaneApplyAllCompletionLiveInvariants(rawDB *sql.DB) error {
	ready, err := ConnectorEventExecutionOwnerColumnReady(context.Background(), rawDB)
	if err != nil {
		return err
	}
	if !ready {
		return fmt.Errorf("connector event execution-owner fence is not ready")
	}
	sessionCurrentBearerReady, err := sessionCurrentBearerTokenColumnReady(context.Background(), rawDB)
	if err != nil {
		return err
	}
	if !sessionCurrentBearerReady {
		return fmt.Errorf("session CurrentBearerToken live column is missing or incompatible")
	}
	outboundGenerationReady, err := runtimeOutboundIntentGenerationReady(context.Background(), rawDB)
	if err != nil {
		return err
	}
	if !outboundGenerationReady {
		return fmt.Errorf("Runtime outbound-intent generation table is missing or incompatible")
	}
	runtimeTokenPermissionsReady, err := runtimeTokenObjectPermissionsReady(context.Background(), rawDB)
	if err != nil {
		return err
	}
	if !runtimeTokenPermissionsReady {
		return errors.New(runtimeTokenObjectPermissionsNotReadyReason)
	}
	return nil
}

func MarkConnectorPushLegacyMirrorBackfillCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("connector-push-legacy-mirror-backfill-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, connectorPushLegacyMirrorBackfillStateKey, runID, "completed", "completed", nil)
}

func MarkWebhookSecretKeyReadProtectionBackfillCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("webhook-secretkey-read-protection-backfill-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, webhookSecretKeyReadProtectionBackfillStateKey, runID, "completed", "completed", nil)
}

// IsConnectorEventEnqueueOrdinalBackfillCompleted reports whether this
// release's connector event queue ordinal repair has run at least once. It gates
// the start-only path so databases with an older completed control_plane_apply_all
// marker cannot start workers with legacy EnqueueOrdinal=0 rows.
func IsConnectorEventEnqueueOrdinalBackfillCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, connectorEventEnqueueOrdinalBackfillStateKey)
	if err != nil {
		return false, err
	}
	if !bootstrapStateCompleted(state) {
		return false, nil
	}
	ready, err := ConnectorEventEnqueueOrdinalBackfillColumnsReady(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	if !ready {
		return false, nil
	}
	remaining, err := ConnectorEventEnqueueOrdinalLiveZeroCount(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	return remaining == 0, nil
}

// MarkConnectorEventEnqueueOrdinalBackfillCompleted stamps the bootstrap-state
// receipt after MigrateConnectorEventEnqueueOrdinalBackfill completes.
func MarkConnectorEventEnqueueOrdinalBackfillCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("connector-event-enqueue-ordinal-backfill-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, connectorEventEnqueueOrdinalBackfillStateKey, runID, "completed", "completed", nil)
}

// IsConnectorEventSourceLaneBackfillCompleted reports whether this release's
// connector event source-lane materialization has run at least once and the
// backlog no longer has payload-only source lanes.
func IsConnectorEventSourceLaneBackfillCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, connectorEventSourceLaneBackfillStateKey)
	if err != nil {
		return false, err
	}
	if !bootstrapStateCompleted(state) {
		return false, nil
	}
	ready, err := ConnectorEventSourceLaneColumnsReady(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	if !ready {
		return false, nil
	}
	remaining, err := ConnectorEventSourceLanePayloadOnlyCount(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	return remaining == 0, nil
}

// MarkConnectorEventSourceLaneBackfillCompleted stamps the bootstrap-state
// receipt after BackfillConnectorEventSourceLanes completes.
func MarkConnectorEventSourceLaneBackfillCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("connector-event-source-lane-backfill-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, connectorEventSourceLaneBackfillStateKey, runID, "completed", "completed", nil)
}

// IsConnectorEventExecutionOwnerFenceCompleted reports whether this release
// applied the mixed-version ownership fence and the column remains live.
func IsConnectorEventExecutionOwnerFenceCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, connectorEventExecutionOwnerFenceStateKey)
	if err != nil {
		return false, err
	}
	if !bootstrapStateCompleted(state) {
		return false, nil
	}
	return ConnectorEventExecutionOwnerColumnReady(context.Background(), rawDB)
}

// MarkConnectorEventExecutionOwnerFenceCompleted stamps the release receipt
// after the live schema readiness check succeeds.
func MarkConnectorEventExecutionOwnerFenceCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	return markConnectorEventExecutionOwnerFenceCompleted(rawDB)
}

func markConnectorEventExecutionOwnerFenceCompleted(rawDB *sql.DB) error {
	runID := fmt.Sprintf("connector-event-execution-owner-fence-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, connectorEventExecutionOwnerFenceStateKey, runID, "completed", "completed", nil)
}

// IsListViewLegacyCutoverCompleted reports whether this release's private
// list-view ownership cutover has run and the physical OwnerId relationship is
// closed.
func IsListViewLegacyCutoverCompleted(rawDB *sql.DB) (bool, error) {
	state, err := loadBootstrapStateForKey(rawDB, listViewLegacyCutoverStateKey)
	if err != nil {
		return false, err
	}
	if !bootstrapStateCompleted(state) {
		return false, nil
	}
	ready, err := ListViewLegacyCutoverColumnsReady(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	if !ready {
		return false, nil
	}
	closed, err := ListViewLegacyCutoverClosed(context.Background(), rawDB)
	if err != nil {
		return false, err
	}
	return closed, nil
}

// MarkListViewLegacyCutoverCompleted stamps the bootstrap-state receipt after
// BackfillListViewLegacyCutover completes and the OwnerId FK has been ensured.
func MarkListViewLegacyCutoverCompleted(rawDB *sql.DB) error {
	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return err
	}
	runID := fmt.Sprintf("listview-legacy-cutover-%d", time.Now().UTC().UnixNano())
	return persistBootstrapStateForKey(rawDB, listViewLegacyCutoverStateKey, runID, "completed", "completed", nil)
}

func persistBootstrapStateForKey(rawDB *sql.DB, stateKey, runID, phase, status string, lastErr error) error {
	payload := ""
	if lastErr != nil {
		serialized, marshalErr := json.Marshal(map[string]string{
			"message":   lastErr.Error(),
			"timestamp": time.Now().UTC().Format(time.RFC3339),
		})
		if marshalErr != nil {
			return marshalErr
		}
		payload = string(serialized)
	}
	_, err := rawDB.Exec(
		fmt.Sprintf(
			"INSERT INTO `%s` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) VALUES (?,?,?,?,?,NOW(),NOW()) "+
				"ON DUPLICATE KEY UPDATE `RunID`=VALUES(`RunID`),`Phase`=VALUES(`Phase`),`Status`=VALUES(`Status`),`LastError`=VALUES(`LastError`),`LastModifiedDate`=NOW()",
			bootstrapStateTable,
		),
		stateKey, runID, phase, status, payload,
	)
	return err
}