package bootstrapimpl

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
)

// VerifySchemaReadiness checks managed schema registry health without mutating schema.
func VerifySchemaReadiness(db *database.TiDBConnection) error {
	repo := persistence.NewSchemaRepository(db.DB(), logging.Default)
	health, err := repo.ValidateSchemaRegistry()
	if err != nil {
		return fmt.Errorf("validate schema registry: %w", err)
	}
	if health == nil {
		return fmt.Errorf("schema registry health check returned nil")
	}
	if health.Status != persistence.StatusHealthy {
		return fmt.Errorf("schema registry unhealthy: expected=%d actual=%d missing=%v",
			health.ExpectedCount, health.ActualCount, health.MissingTables)
	}
	if err := verifyCriticalManagedColumns(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	if err := verifyConnectorWorkItemProjectionOrderIndex(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	if err := verifyPackGenerationAuthorityIndexes(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	if err := verifyPackGenerationConnectorBindingDefinitionAuthoritySchema(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	if err := verifyConnectorServingEpochReproducibilityAuthoritySchema(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	if err := verifyConnectorV2IngressReceiptAuthoritySchema(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	return nil
}

type requiredManagedIndex struct {
	tableName string
	indexName string
	columns   []string
	unique    bool
}

var packGenerationAuthorityRequiredIndexes = []requiredManagedIndex{
	{
		tableName: "_System_PackGeneration",
		indexName: "uq_pack_generation_install",
		columns:   []string{"PackInstallId"},
		unique:    true,
	},
	{
		tableName: "_System_PackActivation",
		indexName: "uq_pack_activation_name",
		columns:   []string{"PackName"},
		unique:    true,
	},
	{
		tableName: "_System_PackGenerationConnectorBindingDefinitionSet",
		indexName: "PRIMARY",
		columns:   []string{"PackGenerationId"},
		unique:    true,
	},
	{
		tableName: "_System_PackGenerationConnectorBindingDefinition",
		indexName: "PRIMARY",
		columns:   []string{"Id"},
		unique:    true,
	},
	{
		tableName: "_System_PackGenerationConnectorBindingDefinition",
		indexName: "ux_pack_generation_binding_definition_ordinal",
		columns:   []string{"PackGenerationId", "DefinitionOrdinal"},
		unique:    true,
	},
	{
		tableName: "_System_PackGenerationConnectorBindingDefinition",
		indexName: "ux_pack_generation_binding_definition_identity",
		columns:   []string{"PackGenerationId", "Kind", "BusinessKey"},
		unique:    true,
	},
}

// verifyPackGenerationAuthorityIndexes fails startup before any source-backed
// lifecycle can publish or select a generation without its physical singleton
// constraints. The managed catalog is the schema authority; this check only
// proves that InitL0/apply materialized the required catalog shape.
func verifyPackGenerationAuthorityIndexes(rawDB *sql.DB) error {
	for _, required := range packGenerationAuthorityRequiredIndexes {
		// Use the shared physical-index verifier so every readiness consumer has
		// one definition of an exact index. In particular, it excludes any named
		// index with a non-NULL SUB_PART: a prefix unique index does not enforce
		// uniqueness over the complete immutable authority identity.
		if err := persistence.VerifyExactTableIndex(
			context.Background(),
			rawDB,
			required.tableName,
			required.indexName,
			required.unique,
			required.columns,
		); err != nil {
			return fmt.Errorf(
				"schema readiness missing required full-column unique index %s.%s(%s): %w; run 'cmd/setup apply' to reconcile schema",
				required.tableName,
				required.indexName,
				strings.Join(required.columns, ","),
				err,
			)
		}
	}
	return nil
}

type criticalManagedColumnRequirement struct {
	tableName string
	columns   []string
}

var criticalManagedColumnRequirements = []criticalManagedColumnRequirement{
	{
		tableName: constants.TableObject,
		columns: []string{
			constants.FieldSysObject_IsCustomized,
			constants.FieldSysObject_AccessPolicy,
			constants.FieldSysObject_LockRule,
		},
	},
	{
		tableName: constants.TableRecordType,
		columns: []string{
			constants.FieldSysRecordType_DeveloperName,
			"SobjectType",
		},
	},
	{
		// Recycle-bin scope=mine keys ownership on DeletedById
		// (pingcap-inc/NexusCRM#1750). A start-only boot whose DB has not
		// run `setup apply` would lack this column, and the scope=mine page
		// query would fail with unknown-column at request time. Gating it
		// here fails closed at readiness instead, forcing the operator to
		// apply before serving — the column is added by ReconcileTableColumns
		// under ApplyAll, which a plain restart does not run.
		tableName: constants.TableRecycleBin,
		columns: []string{
			constants.FieldSysRecycleBin_DeletedByID,
		},
	},
	{
		tableName: "_System_ConnectorPushAttempt",
		columns:   []string{"OriginatingUserSessionJSON", "CreateConflictResolutionFingerprint"},
	},
	{
		// _System_ConnectorInboundRemoteIDEvidence backs the inbound register
		// action (registerConnectorInboundRemoteAuthorship), which reads and
		// writes this table inside the atomic inbound flow tx and treats every
		// store error as fatal. A start-only warm-upgrade DB whose completed
		// control_plane_apply_all marker takes the fast path and skips apply
		// would lack the table, and the first connector inbound flow would fail
		// at runtime (rolling the record back / retrying) instead of failing
		// closed at startup. Gate on the per-destination columns so readiness
		// forces an apply before serving — the table is created under ApplyAll,
		// which a plain restart does not run. (Previously deferred while the
		// table had no runtime consumer; epic #2028 issue #2042 is that first
		// consumer, so the gate lands with it.)
		tableName: "_System_ConnectorInboundRemoteIDEvidence",
		columns:   []string{"RemoteIDField", "ActiveKey"},
	},
	{
		// _System_ConnectorEvent.DeliverySequence stores the per-delivery ordinal
		// assigned by the dispatcher's Phase B enqueue loop. ClaimNext and
		// ListDeadEvents order by (CreatedDate ASC, DeliverySequence ASC, Id ASC)
		// to guarantee intra-batch FIFO. A warm-upgrade DB that skipped apply
		// would lack the column and both queries would fail with unknown-column at
		// runtime; gate forces apply before serving.
		// DeliveryId is the app-assigned grouping key minted once per webhook delivery
		// by the dispatcher; ClaimNext's NOT EXISTS guard uses it to serialize
		// execution within one delivery. NULL rows (pre-#2154) are singletons.
		// EnqueueOrdinal is the DB-stable pack-scoped enqueue order fact used by
		// same-source cross-delivery claim serialization.
		// SourceObjectKind/SourceObjectID/SourceLaneKey are materialized from
		// the queue envelope. ClaimNext's same-source serialization uses the
		// indexed SourceLaneKey hot path, while Enqueue writes all three. A
		// warm-upgrade DB that skipped apply would otherwise accept traffic
		// until the first webhook fails with unknown-column.
		// NextAttemptAt is the retry-due gate. ClaimNext, PeekBatch, ClaimByID,
		// stats, and pressure probes all reference it in hot claim predicates, so
		// a warm-upgrade DB must fail closed before workers start.
		// QueueClass is snapshotted from the route and read by queue scans/replay.
		// ExecutionOwner is the durable mixed-version CAS fence between the
		// Event and WorkItem executors. A start-only warm upgrade must not serve
		// until the column exists.
		tableName: "_System_ConnectorEvent",
		columns: []string{
			"DeliverySequence",
			"DeliveryId",
			"EnqueueOrdinal",
			"SourceObjectKind",
			"SourceObjectID",
			"SourceLaneKey",
			"NextAttemptAt",
			"QueueClass",
			"ExecutionOwner",
		},
	},
	{
		// Route execution substrate columns are read by DBRouteResolver,
		// enqueue/replay snapshotting, and worker-side filter/mode execution.
		// A warm-upgrade DB that skips apply would otherwise start serving and
		// fail connector workers at first route resolution with unknown-column
		// errors. Gate these columns so readiness forces schema reconciliation.
		tableName: "_System_ConnectorInboundRoute",
		columns: []string{
			"InvocationMode",
			"BatchInputName",
			"BatchResultOutputName",
			"MaxBatchSize",
			"QueueClass",
			"FlowExecutionFilter",
		},
	},
	{
		// Async create-conflict recovery persists the sparse PATCH intent on the
		// push attempt before the remote create runs. A warm-upgrade DB that
		// skips ApplyAll would otherwise fail when recovery reads that durable
		// intent, so readiness must force reconciliation before serving.
		tableName: "_System_ConnectorPushAttempt",
		columns:   []string{"CreateConflictPatchIntentJSON"},
	},
	{
		// Every field-mapping execution and authorization reader references
		// ProjectionOnly in its fail-closed predicate. A start-only warm
		// upgrade must reconcile the additive column before any request or
		// worker can evaluate those readers.
		tableName: "_System_ObjectFieldMapping",
		columns:   []string{"ProjectionOnly"},
	},
	{
		// Standard-action discovery reads HeaderFallbackEligible in the request
		// path to decide whether unauthored record-header fallback may include
		// an action. A completed-marker warm upgrade can skip ApplyAll, so gate
		// this column here and fail closed before serving an old schema.
		tableName: constants.TableStandardAction,
		columns: []string{
			constants.FieldSysStandardAction_HeaderFallbackEligible,
		},
	},
}

// catalogGatedManagedColumnRequirements are columns that must be present in the
// live schema, but only once the embedded managed catalog actually declares
// them. Unlike criticalManagedColumnRequirements (whose columns the catalog
// already declares, so ReconcileTableColumns under ApplyAll can always provision
// them and gating them is unconditionally remediable), these columns live in the
// schema-catalog skew window: the paired L0 migration adds the column and only a
// later catalog bump declares it here.
//
// Gating such a column unconditionally would be unremediable — controlplane's
// own ReconcileTableColumns provisions strictly from the embedded catalog
// (platform_core.go's ReconcileTableColumns "can only add columns from the
// spec-derived table definition"), so while the catalog lags, `cmd/setup apply`
// cannot add the column and readiness would fail closed forever with no operator
// recovery. Gating on catalog-declared presence makes the requirement
// self-healing: while the catalog lags, the column is not yet required here (the
// seed path's column-existence fallback in packs/platform_policy.go bridges the
// gap — opt-in-false scopes converge, and an explicit opt-in fails loudly with a
// migration-required message); once the catalog that ships alongside the
// dynamic-client runtime declares the column, ApplyAll provisions it AND this
// gate enforces it, so lifecycle convergence never reports ready while the
// runtime read path would still hit unknown-column.
//
// Cross-service boundary: the runtime that reads this column is not in
// controlplane — it ships in NexusCRM#3986 together with the L0 migration that
// adds the column and the catalog regeneration that declares it. This gate is
// controlplane's half: it begins requiring the column the moment the column
// enters controlplane's own expected schema, which is the same release that
// introduces the runtime. controlplane cannot require the column before its
// catalog knows about it (nothing here can create it), so the paired release
// owns the runtime's own unknown-column readiness for its process.
var catalogGatedManagedColumnRequirements = []criticalManagedColumnRequirement{
	{
		// AllowDynamicPublicClients (pingcap-inc/NexusCRM#3986) opts a custom
		// OAuth scope in to dynamic public-client registration. The paired
		// runtime reads this column; gate it here so a warm-upgrade DB that has
		// applied the migration+catalog bump but skipped ApplyAll fails closed
		// before serving instead of erroring at request time.
		tableName: constants.TableOAuthCustomScope,
		columns:   []string{"AllowDynamicPublicClients"},
	},
	{
		// _System_OAuthConnectionRevocation (pingcap-inc/NexusCRM#3986) stores
		// the per-(user, Connected App) disconnect epoch. The paired runtime
		// upserts it on self-service disconnect and reads RevokedAt when
		// resuming a pre-login authorize handle; both treat store errors as
		// fatal, so a start-only warm-upgrade DB that skipped ApplyAll would
		// fail those flows at request time with unknown-table. Gate the
		// columns here (a missing table reports every column missing) so
		// readiness fails closed and forces an apply before serving. Catalog-
		// gated because the table lives in the same skew window as the column
		// above: it enters controlplane's expected schema only with the
		// paired NexusCRM release's catalog bump.
		tableName: "_System_OAuthConnectionRevocation",
		columns:   []string{"UserId", "ConnectedAppId", "RevokedAt"},
	},
}

// catalogDeclaresManagedColumn reports whether the embedded managed catalog
// declares columnName on tableName. It is the self-healing trigger for
// catalogGatedManagedColumnRequirements: a column is enforced at readiness only
// once the catalog (which controlplane provisions from) knows about it.
func catalogDeclaresManagedColumn(defs []schema.TableDefinition, tableName, columnName string) bool {
	for _, def := range defs {
		if !strings.EqualFold(def.TableName, tableName) {
			continue
		}
		for _, col := range def.Columns {
			if strings.EqualFold(col.Name, columnName) || strings.EqualFold(col.PhysicalName, columnName) {
				return true
			}
		}
	}
	return false
}

// collectCatalogGatedMissingColumns returns the catalog-gated required columns
// that the catalog declares but the live schema is missing. Columns the catalog
// does not yet declare are skipped (and not queried) so the skew window neither
// bricks startup nor issues spurious probes.
func collectCatalogGatedMissingColumns(rawDB *sql.DB, defs []schema.TableDefinition) ([]string, error) {
	var missing []string
	for _, entry := range catalogGatedManagedColumnRequirements {
		for _, columnName := range entry.columns {
			if !catalogDeclaresManagedColumn(defs, entry.tableName, columnName) {
				continue
			}
			var count int
			err := rawDB.QueryRow(
				"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = LOWER(?) AND LOWER(COLUMN_NAME) = LOWER(?)",
				entry.tableName,
				columnName,
			).Scan(&count)
			if err != nil {
				return nil, fmt.Errorf("verify catalog-gated required column %s.%s: %w", entry.tableName, columnName, err)
			}
			if count == 0 {
				missing = append(missing, fmt.Sprintf("%s.%s", entry.tableName, columnName))
			}
		}
	}
	return missing, nil
}

func verifyCriticalManagedColumns(rawDB *sql.DB) error {
	var missing []string
	for _, entry := range criticalManagedColumnRequirements {
		for _, columnName := range entry.columns {
			var count int
			err := rawDB.QueryRow(
				"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = LOWER(?) AND LOWER(COLUMN_NAME) = LOWER(?)",
				entry.tableName,
				columnName,
			).Scan(&count)
			if err != nil {
				return fmt.Errorf("verify required column %s.%s: %w", entry.tableName, columnName, err)
			}
			if count == 0 {
				missing = append(missing, fmt.Sprintf("%s.%s", entry.tableName, columnName))
			}
		}
	}

	catalogGatedMissing, err := collectCatalogGatedMissingColumns(rawDB, GetSystemTableDefinitions())
	if err != nil {
		return err
	}
	missing = append(missing, catalogGatedMissing...)

	if len(missing) > 0 {
		return fmt.Errorf(
			"schema readiness missing required columns: %s; run 'cmd/setup apply' to reconcile schema",
			strings.Join(missing, ", "),
		)
	}

	var firstNameNullable string
	err = rawDB.QueryRow(
		"SELECT IS_NULLABLE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = LOWER(?) AND LOWER(COLUMN_NAME) = LOWER(?)",
		constants.TableUser,
		constants.FieldFirstName,
	).Scan(&firstNameNullable)
	if err != nil {
		return fmt.Errorf("verify managed column shape %s.%s: %w", constants.TableUser, constants.FieldFirstName, err)
	}
	if !strings.EqualFold(firstNameNullable, "YES") {
		return fmt.Errorf(
			"schema readiness drift: %s.%s must be nullable for the authoritative create contract; run 'cmd/setup apply' to reconcile schema",
			constants.TableUser,
			constants.FieldFirstName,
		)
	}

	var firstNameRequired int
	err = rawDB.QueryRow(
		"SELECT f.`Required` FROM `"+constants.TableField+"` f JOIN `"+constants.TableObject+
			"` o ON o.Id = f.ObjectId WHERE o.ApiName = ? AND f.ApiName = ?",
		constants.TableUser,
		constants.FieldFirstName,
	).Scan(&firstNameRequired)
	if err != nil {
		return fmt.Errorf("verify managed field metadata %s.%s Required: %w", constants.TableUser, constants.FieldFirstName, err)
	}
	if firstNameRequired != 0 {
		return fmt.Errorf(
			"schema readiness drift: %s.%s _System_Field.Required = %d, must be 0; run 'cmd/setup apply' to reconcile",
			constants.TableUser,
			constants.FieldFirstName,
			firstNameRequired,
		)
	}

	rows, err := rawDB.Query(
		"SELECT rt.`Id`, rt.`Name`, COALESCE(rt.`" + constants.FieldSysRecordType_DeveloperName + "`, ''), COALESCE(rt.`SobjectType`, ''), COALESCE(o.`ApiName`, '') " +
			"FROM `" + constants.TableRecordType + "` rt LEFT JOIN `" + constants.TableObject + "` o ON o.`Id` = rt.`ObjectId` " +
			"WHERE COALESCE(rt.`IsDeleted`, 0) = 0",
	)
	if err != nil {
		return fmt.Errorf("verify canonical record type fields: %w", err)
	}
	defer rows.Close()

	var nonCanonicalRecordTypes int
	for rows.Next() {
		var id, name, developerName, sobjectType, objectAPIName string
		if err := rows.Scan(&id, &name, &developerName, &sobjectType, &objectAPIName); err != nil {
			return fmt.Errorf("verify canonical record type fields: %w", err)
		}
		if strings.TrimSpace(developerName) == "" ||
			strings.TrimSpace(sobjectType) == "" ||
			!strings.EqualFold(strings.TrimSpace(sobjectType), strings.TrimSpace(objectAPIName)) {
			nonCanonicalRecordTypes++
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("verify canonical record type fields: %w", err)
	}
	if nonCanonicalRecordTypes != 0 {
		return fmt.Errorf(
			"schema readiness drift: %s has %d live row(s) with blank %s or non-canonical SobjectType; require nonblank %s and SobjectType = ObjectId.ApiName; run 'cmd/setup apply' to reconcile",
			constants.TableRecordType,
			nonCanonicalRecordTypes,
			constants.FieldSysRecordType_DeveloperName,
			constants.FieldSysRecordType_DeveloperName,
		)
	}

	return nil
}