package bootstrapimpl

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

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
)

var connectorServingEpochReproducibilityTableNames = []string{
	"_System_ConnectorServingEpochReproducibility",
	"_System_ConnectorServingEpochBindingOwner",
	"_System_ConnectorServingEpochCredentialGeneration",
}

type exactManagedAuthorityPhysicalForeignKey struct {
	columnName       string
	referencedSchema string
	currentSchema    string
	referencedTable  string
	referencedColumn string
	deleteRule       string
	updateRule       string
}

// verifyConnectorServingEpochReproducibilityAuthoritySchema is the stopped-
// upgrade gate for the immutable ServingEpoch reproducibility envelope. It
// derives the required physical shape from the pinned managed catalog, then
// checks the live database without creating, repairing, or backfilling rows.
func verifyConnectorServingEpochReproducibilityAuthoritySchema(rawDB *sql.DB) error {
	return verifyExactManagedAuthoritySchema(
		rawDB,
		"Connector ServingEpoch reproducibility",
		connectorServingEpochReproducibilityTableNames,
	)
}

// verifyExactManagedAuthoritySchema is the single physical-readiness path for
// immutable managed authority table sets. It derives the complete catalog-owned
// columns, named indexes, and restrictive foreign keys, and rejects both
// missing and extra physical shape without mutating the database.
func verifyExactManagedAuthoritySchema(rawDB *sql.DB, authorityName string, tableNames []string) error {
	definitions, err := exactManagedAuthorityDefinitions(tableNames)
	if err != nil {
		return err
	}
	physicalColumns, err := LoadPhysicalColumnShapes(rawDB)
	if err != nil {
		return fmt.Errorf("verify %s columns: %w", authorityName, err)
	}
	var columnDrift []string
	for _, definition := range definitions {
		if err := verifyExactManagedAuthorityColumns(physicalColumns, definition); err != nil {
			columnDrift = append(columnDrift, err.Error())
		}
	}
	if len(columnDrift) != 0 {
		return fmt.Errorf(
			"%s schema is incomplete: %s",
			authorityName,
			strings.Join(columnDrift, "; "),
		)
	}
	for _, definition := range definitions {
		if err := verifyExactManagedAuthorityIndexes(rawDB, definition); err != nil {
			return err
		}
		if err := verifyExactManagedAuthorityForeignKeys(rawDB, authorityName, definition); err != nil {
			return err
		}
	}
	return nil
}

func connectorServingEpochReproducibilityDefinitions() ([]schema.TableDefinition, error) {
	return exactManagedAuthorityDefinitions(connectorServingEpochReproducibilityTableNames)
}

func exactManagedAuthorityDefinitions(tableNames []string) ([]schema.TableDefinition, error) {
	managed := GetManagedTableDefinitions()
	definitions := make([]schema.TableDefinition, 0, len(tableNames))
	for _, tableName := range tableNames {
		matches := 0
		var found schema.TableDefinition
		for _, definition := range managed {
			if !strings.EqualFold(definition.TableName, tableName) {
				continue
			}
			matches++
			found = definition
		}
		if matches != 1 {
			return nil, fmt.Errorf(
				"schema readiness catalog must declare exactly one %s table, found %d",
				tableName,
				matches,
			)
		}
		definitions = append(definitions, found)
	}
	return definitions, nil
}

func verifyConnectorServingEpochReproducibilityColumns(
	physicalColumns map[string]map[string]PhysicalColumnShape,
	definition schema.TableDefinition,
) error {
	return verifyExactManagedAuthorityColumns(physicalColumns, definition)
}

func verifyExactManagedAuthorityColumns(
	physicalColumns map[string]map[string]PhysicalColumnShape,
	definition schema.TableDefinition,
) error {
	actual := physicalColumns[strings.ToLower(definition.TableName)]
	if len(actual) != len(definition.Columns) {
		return fmt.Errorf(
			"schema readiness drift: %s has %d columns, require exact catalog-owned shape with %d; run 'cmd/setup apply' to reconcile schema",
			definition.TableName,
			len(actual),
			len(definition.Columns),
		)
	}
	for _, catalogColumn := range definition.Columns {
		desired := catalogColumn
		// This FK coordinate intentionally inherits the existing DefinitionSet
		// parent's database-default collation. The Connector reader supplies the
		// byte-exact residual check above that physical FK-compatible lookup.
		if strings.EqualFold(definition.TableName, "_System_ConnectorServingEpochBindingOwner") &&
			strings.EqualFold(desired.Name, "PackGenerationId") {
			parentTable := physicalColumns[strings.ToLower("_System_PackGenerationConnectorBindingDefinitionSet")]
			parent, ok := parentTable[strings.ToLower("PackGenerationId")]
			if !ok || !parent.CollationName.Valid {
				return fmt.Errorf(
					"schema readiness drift: %s.%s cannot inherit the exact DefinitionSet parent collation; run 'cmd/setup apply' to reconcile schema",
					definition.TableName,
					desired.Name,
				)
			}
			desired.CollationName = parent.CollationName.String
		}
		column, ok := actual[strings.ToLower(desired.Name)]
		if !ok || !ColumnShapeMatches(column, desired) {
			return fmt.Errorf(
				"schema readiness drift: %s.%s does not match catalog type/nullability/default/collation; run 'cmd/setup apply' to reconcile schema",
				definition.TableName,
				desired.Name,
			)
		}
	}
	return nil
}

func verifyConnectorServingEpochReproducibilityIndexes(
	rawDB *sql.DB,
	definition schema.TableDefinition,
) error {
	return verifyExactManagedAuthorityIndexes(rawDB, definition)
}

func verifyExactManagedAuthorityIndexes(rawDB *sql.DB, definition schema.TableDefinition) error {
	for _, required := range exactManagedAuthorityRequiredIndexes(definition) {
		if err := persistence.VerifyExactTableIndex(
			context.Background(),
			rawDB,
			required.tableName,
			required.indexName,
			required.unique,
			required.columns,
		); err != nil {
			indexKind := ""
			if required.unique {
				indexKind = "unique "
			}
			return fmt.Errorf(
				"schema readiness drift: %s.%s(%s) must be the exact full-column %sindex: %w; run 'cmd/setup apply' to reconcile schema",
				required.tableName,
				required.indexName,
				strings.Join(required.columns, ","),
				indexKind,
				err,
			)
		}
	}
	return nil
}

func connectorServingEpochRequiredIndexes(definition schema.TableDefinition) []requiredManagedIndex {
	return exactManagedAuthorityRequiredIndexes(definition)
}

func exactManagedAuthorityRequiredIndexes(definition schema.TableDefinition) []requiredManagedIndex {
	required := make([]requiredManagedIndex, 0, len(definition.Indices)+1)
	hasExplicitPrimary := false
	for _, index := range definition.Indices {
		if index.Primary {
			hasExplicitPrimary = true
			required = append(required, requiredManagedIndex{
				tableName: definition.TableName,
				indexName: "PRIMARY",
				columns:   append([]string(nil), index.Columns...),
				unique:    true,
			})
			continue
		}
		required = append(required, requiredManagedIndex{
			tableName: definition.TableName,
			indexName: index.Name,
			columns:   append([]string(nil), index.Columns...),
			unique:    index.Unique,
		})
	}
	if hasExplicitPrimary {
		return required
	}
	primaryColumns := make([]string, 0, 1)
	for _, column := range definition.Columns {
		if column.PrimaryKey {
			primaryColumns = append(primaryColumns, column.Name)
		}
	}
	if len(primaryColumns) == 0 {
		return required
	}
	return append([]requiredManagedIndex{{
		tableName: definition.TableName,
		indexName: "PRIMARY",
		columns:   primaryColumns,
		unique:    true,
	}}, required...)
}

func verifyConnectorServingEpochReproducibilityForeignKeys(
	rawDB *sql.DB,
	definition schema.TableDefinition,
) error {
	return verifyExactManagedAuthorityForeignKeys(rawDB, "Connector ServingEpoch reproducibility", definition)
}

func verifyExactManagedAuthorityForeignKeys(
	rawDB *sql.DB,
	authorityName string,
	definition schema.TableDefinition,
) error {
	rows, err := rawDB.Query(`SELECT
k.COLUMN_NAME, k.REFERENCED_TABLE_SCHEMA, DATABASE(), k.REFERENCED_TABLE_NAME,
k.REFERENCED_COLUMN_NAME, r.DELETE_RULE, r.UPDATE_RULE
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS r
  ON r.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
 AND r.CONSTRAINT_NAME = k.CONSTRAINT_NAME
 AND r.TABLE_NAME = k.TABLE_NAME
WHERE k.TABLE_SCHEMA = DATABASE()
  AND LOWER(k.TABLE_NAME) = LOWER(?)
  AND k.REFERENCED_TABLE_NAME IS NOT NULL`, definition.TableName)
	if err != nil {
		return fmt.Errorf("verify %s foreign keys on %s: %w", authorityName, definition.TableName, err)
	}
	defer rows.Close()

	actual := make([]exactManagedAuthorityPhysicalForeignKey, 0, len(definition.ForeignKeys))
	for rows.Next() {
		var foreignKey exactManagedAuthorityPhysicalForeignKey
		if err := rows.Scan(
			&foreignKey.columnName,
			&foreignKey.referencedSchema,
			&foreignKey.currentSchema,
			&foreignKey.referencedTable,
			&foreignKey.referencedColumn,
			&foreignKey.deleteRule,
			&foreignKey.updateRule,
		); err != nil {
			return fmt.Errorf("verify %s foreign keys on %s: %w", authorityName, definition.TableName, err)
		}
		actual = append(actual, foreignKey)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("verify %s foreign keys on %s: %w", authorityName, definition.TableName, err)
	}
	if len(actual) != len(definition.ForeignKeys) {
		return fmt.Errorf(
			"schema readiness drift: %s has %d foreign keys, require exact catalog-owned set with %d; run 'cmd/setup apply' to reconcile schema",
			definition.TableName,
			len(actual),
			len(definition.ForeignKeys),
		)
	}

	for _, required := range definition.ForeignKeys {
		referencedTable, referencedColumn, ok := parseExactManagedAuthorityForeignKeyReference(required.References)
		if !ok {
			return fmt.Errorf(
				"schema readiness catalog has invalid foreign-key reference %s.%s -> %q",
				definition.TableName,
				required.Column,
				required.References,
			)
		}
		matches := 0
		for _, foreignKey := range actual {
			if strings.EqualFold(foreignKey.columnName, required.Column) &&
				foreignKey.referencedSchema == foreignKey.currentSchema &&
				strings.EqualFold(foreignKey.referencedTable, referencedTable) &&
				strings.EqualFold(foreignKey.referencedColumn, referencedColumn) &&
				exactManagedAuthorityForeignKeyRuleMatches(foreignKey.deleteRule, required.OnDelete) &&
				exactManagedAuthorityForeignKeyRuleMatches(foreignKey.updateRule, required.OnUpdate) {
				matches++
			}
		}
		if matches != 1 {
			return fmt.Errorf(
				"schema readiness drift: %s must have exactly one foreign key %s -> %s(%s) with restrictive actions; run 'cmd/setup apply' to reconcile schema",
				definition.TableName,
				required.Column,
				referencedTable,
				referencedColumn,
			)
		}
	}
	return nil
}

func parseConnectorServingEpochForeignKeyReference(reference string) (string, string, bool) {
	return parseExactManagedAuthorityForeignKeyReference(reference)
}

func parseExactManagedAuthorityForeignKeyReference(reference string) (string, string, bool) {
	open := strings.LastIndex(reference, "(")
	if open <= 0 || !strings.HasSuffix(reference, ")") {
		return "", "", false
	}
	table := strings.TrimSpace(reference[:open])
	column := strings.TrimSpace(reference[open+1 : len(reference)-1])
	return table, column, table != "" && column != ""
}

func connectorServingEpochForeignKeyRuleMatches(actual, catalog string) bool {
	return exactManagedAuthorityForeignKeyRuleMatches(actual, catalog)
}

func exactManagedAuthorityForeignKeyRuleMatches(actual, catalog string) bool {
	required := strings.TrimSpace(catalog)
	if required == "" || strings.EqualFold(required, "RESTRICT") {
		return strings.EqualFold(actual, "RESTRICT") || strings.EqualFold(actual, "NO ACTION")
	}
	return strings.EqualFold(actual, required)
}

// Backward-compatible test alias while readiness consumers converge on the
// generic exact managed-authority vocabulary.
type connectorServingEpochPhysicalForeignKey = exactManagedAuthorityPhysicalForeignKey