package serving

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

const (
	servingEpochTable                     = "_System_ConnectorServingEpoch"
	servingEpochBindingTable              = "_System_ConnectorServingEpochBinding"
	servingEpochReproducibilityTable      = "_System_ConnectorServingEpochReproducibility"
	servingEpochBindingOwnerTable         = "_System_ConnectorServingEpochBindingOwner"
	servingEpochCredentialGenerationTable = "_System_ConnectorServingEpochCredentialGeneration"

	inspectServingColumnsSQL = `SELECT COLUMN_NAME, LOWER(COLUMN_TYPE), IS_NULLABLE, COLLATION_NAME, COLUMN_DEFAULT, EXTRA
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
ORDER BY ORDINAL_POSITION`

	inspectServingIndexesSQL = `SELECT INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, SUB_PART
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
ORDER BY INDEX_NAME, SEQ_IN_INDEX`

	inspectServingForeignKeysSQL = `SELECT kcu.CONSTRAINT_NAME, kcu.COLUMN_NAME,
	       kcu.REFERENCED_TABLE_SCHEMA, kcu.REFERENCED_TABLE_NAME,
	       kcu.REFERENCED_COLUMN_NAME, DATABASE(),
	       rc.UPDATE_RULE, rc.DELETE_RULE
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS kcu
JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS rc
  ON rc.CONSTRAINT_SCHEMA = kcu.CONSTRAINT_SCHEMA
 AND rc.TABLE_NAME = kcu.TABLE_NAME
 AND rc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
WHERE kcu.CONSTRAINT_SCHEMA = DATABASE()
  AND kcu.TABLE_SCHEMA = DATABASE()
  AND kcu.TABLE_NAME = ?
  AND kcu.REFERENCED_TABLE_NAME IS NOT NULL
ORDER BY kcu.CONSTRAINT_NAME, kcu.ORDINAL_POSITION`
)

type expectedServingColumn struct {
	name                  string
	columnType            string
	nullable              bool
	collation             string
	hasDefault            bool
	columnDefault         string
	allowDefaultGenerated bool
}

type expectedServingIndex struct {
	name    string
	unique  bool
	columns []string
}

type expectedServingForeignKey struct {
	column           string
	referencedTable  string
	referencedColumn string
}

type servingTableSchema struct {
	name        string
	columns     []expectedServingColumn
	indexes     []expectedServingIndex
	foreignKeys []expectedServingForeignKey
}

var servingEpochColumns = []expectedServingColumn{
	{name: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "EpochDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "ServingScopeDigest", columnType: "char(64)", collation: "utf8mb4_bin"},
	{name: "ActiveScopeDigest", columnType: "char(64)", nullable: true, collation: "utf8mb4_bin"},
	{name: "State", columnType: "varchar(32)", collation: "utf8mb4_bin", hasDefault: true, columnDefault: "prepared"},
	{name: "OrgId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "AppInstallationId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "AuthorizedWorkloadId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "CatalogRevision", columnType: "decimal(20,0)"},
	{name: "SchemaGeneration", columnType: "decimal(20,0)"},
	{name: "EngineGeneration", columnType: "decimal(20,0)"},
	{name: "SourceSnapshotDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "BindingViewDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "ReceiptProtocolVersion", columnType: "varchar(32)", collation: "utf8mb4_bin"},
	{name: "ReceiptCapabilityVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "SecurityEpoch", columnType: "decimal(20,0)"},
	{name: "L2ConfigurationGenerationRef", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "L2ConfigurationDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "L2VerificationReceipt", columnType: "mediumblob"},
	{name: "L2VerificationReceiptDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
	{name: "LastModifiedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
}

var servingEpochBindingColumns = []expectedServingColumn{
	{name: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "BindingOrdinal", columnType: "int(11)"},
	{name: "BindingIdentityDigest", columnType: "char(64)", collation: "utf8mb4_bin"},
	{name: "Kind", columnType: "varchar(64)", collation: "utf8mb4_bin"},
	{name: "OwnerPack", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "PackInstallationId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "PackPayloadDigest", columnType: "char(85)", collation: "utf8mb4_bin"},
	{name: "BusinessKey", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "DefinitionVersion", columnType: "decimal(20,0)"},
	{name: "CredentialRefs", columnType: "mediumblob"},
	{name: "RuntimeCommandKind", columnType: "varchar(255)", nullable: true, collation: "utf8mb4_bin"},
	{name: "RuntimePayloadFormatVersion", columnType: "bigint(20)", nullable: true},
	{name: "RuntimeDefinitionRef", columnType: "varchar(255)", nullable: true, collation: "utf8mb4_bin"},
	{name: "RuntimeDefinitionVersion", columnType: "varchar(255)", nullable: true, collation: "utf8mb4_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
	{name: "LastModifiedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
}

var servingEpochReproducibilityColumns = []expectedServingColumn{
	{name: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ConnectorDigestProfileId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ServiceConfigurationGenerationProtocolVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ServiceConfigurationReceiptProtocolVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
	{name: "LastModifiedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
}

var servingEpochBindingOwnerColumns = []expectedServingColumn{
	{name: "ServingEpochBindingId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "PackGenerationId", columnType: "varchar(255)", collation: "utf8mb4_unicode_ci"},
	{name: "DefinitionEnvelopeVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "DefinitionReceiptVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "CanonicalDefinitionDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "DefinitionOwnerReceipt", columnType: "mediumblob"},
	{name: "DefinitionOwnerReceiptDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
	{name: "LastModifiedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
}

var servingEpochCredentialGenerationColumns = []expectedServingColumn{
	{name: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "CredentialOrdinal", columnType: "int(11)"},
	{name: "CredentialGenerationId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ReceiptProtocolVersion", columnType: "varchar(32)", collation: "utf8mb4_bin"},
	{name: "TransportConfigurationDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "OwnerReceipt", columnType: "mediumblob"},
	{name: "OwnerReceiptDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
	{name: "LastModifiedDate", columnType: "datetime(6)", hasDefault: true, columnDefault: "CURRENT_TIMESTAMP(6)", allowDefaultGenerated: true},
}

var servingEpochIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"Id"}},
	{name: "ux_connector_serving_epoch_digest", unique: true, columns: []string{"EpochDigest"}},
	{name: "ux_connector_serving_epoch_active_scope", unique: true, columns: []string{"ActiveScopeDigest"}},
	{name: "idx_connector_serving_epoch_scope_state", columns: []string{"ServingScopeDigest", "State", "Id"}},
}

var servingEpochBindingIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"Id"}},
	{name: "ux_connector_epoch_binding_ordinal", unique: true, columns: []string{"ServingEpochId", "BindingOrdinal"}},
	{name: "ux_connector_epoch_binding_identity", unique: true, columns: []string{"ServingEpochId", "BindingIdentityDigest"}},
	{name: "idx_connector_epoch_binding_pack", columns: []string{"ServingEpochId", "PackInstallationId", "BindingOrdinal"}},
}

var servingEpochReproducibilityIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"ServingEpochId"}},
	{name: "idx_connector_epoch_repro_profile", columns: []string{"ConnectorDigestProfileId", "ServingEpochId"}},
}

var servingEpochBindingOwnerIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"ServingEpochBindingId"}},
	{name: "idx_connector_epoch_binding_owner_pack", columns: []string{"PackGenerationId", "ServingEpochBindingId"}},
}

var servingEpochCredentialGenerationIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"ServingEpochId", "CredentialOrdinal"}},
	{name: "ux_connector_epoch_credential_generation", unique: true, columns: []string{"ServingEpochId", "CredentialGenerationId"}},
	{name: "idx_connector_epoch_credential_generation_ref", columns: []string{"CredentialGenerationId", "ServingEpochId"}},
}

var servingTableSchemas = []servingTableSchema{
	{
		name:    servingEpochTable,
		columns: servingEpochColumns,
		indexes: servingEpochIndexes,
	},
	{
		name:    servingEpochBindingTable,
		columns: servingEpochBindingColumns,
		indexes: servingEpochBindingIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "ServingEpochId", referencedTable: servingEpochTable, referencedColumn: "Id"},
		},
	},
	{
		name:    servingEpochReproducibilityTable,
		columns: servingEpochReproducibilityColumns,
		indexes: servingEpochReproducibilityIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "ServingEpochId", referencedTable: servingEpochTable, referencedColumn: "Id"},
		},
	},
	{
		name:    servingEpochBindingOwnerTable,
		columns: servingEpochBindingOwnerColumns,
		indexes: servingEpochBindingOwnerIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "ServingEpochBindingId", referencedTable: servingEpochBindingTable, referencedColumn: "Id"},
			{column: "PackGenerationId", referencedTable: "_System_PackGenerationConnectorBindingDefinitionSet", referencedColumn: "PackGenerationId"},
		},
	},
	{
		name:    servingEpochCredentialGenerationTable,
		columns: servingEpochCredentialGenerationColumns,
		indexes: servingEpochCredentialGenerationIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "ServingEpochId", referencedTable: servingEpochTable, referencedColumn: "Id"},
			{column: "CredentialGenerationId", referencedTable: "_System_ConnectorCredentialGeneration", referencedColumn: "Id"},
		},
	},
}

// verifyServingSchema is a readiness check only. The physical schema is owned
// by the NexusCRM catalog and converged by Controlplane; Connector must never
// create, alter, or tolerate a rolling compatibility shape here.
func verifyServingSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return fmt.Errorf("%w: database or context is unavailable", ErrSchemaUnavailable)
	}
	for _, table := range servingTableSchemas {
		if err := verifyServingColumns(ctx, db, table.name, table.columns); err != nil {
			return fmt.Errorf("%w: verify %s columns: %w", ErrSchemaUnavailable, table.name, err)
		}
	}
	for _, table := range servingTableSchemas {
		if err := verifyServingIndexes(ctx, db, table.name, table.indexes); err != nil {
			return fmt.Errorf("%w: verify %s indexes: %w", ErrSchemaUnavailable, table.name, err)
		}
	}
	for _, table := range servingTableSchemas {
		if err := verifyServingForeignKeys(ctx, db, table.name, table.foreignKeys); err != nil {
			return fmt.Errorf("%w: verify %s foreign keys: %w", ErrSchemaUnavailable, table.name, err)
		}
	}
	return nil
}

func verifyServingColumns(
	ctx context.Context,
	db *sql.DB,
	table string,
	expected []expectedServingColumn,
) error {
	rows, err := db.QueryContext(ctx, inspectServingColumnsSQL, table)
	if err != nil {
		return err
	}
	defer rows.Close()

	expectedByName := make(map[string]expectedServingColumn, len(expected))
	for _, column := range expected {
		expectedByName[column.name] = column
	}
	found := make(map[string]struct{}, len(expected))
	for rows.Next() {
		var (
			name          string
			columnType    string
			nullable      string
			collation     sql.NullString
			columnDefault sql.NullString
			extra         string
		)
		if err := rows.Scan(
			&name,
			&columnType,
			&nullable,
			&collation,
			&columnDefault,
			&extra,
		); err != nil {
			return err
		}
		want, ok := expectedByName[name]
		if !ok {
			return fmt.Errorf("%s has unexpected column %s", table, name)
		}
		if _, duplicate := found[name]; duplicate {
			return fmt.Errorf("%s reports column %s more than once", table, name)
		}
		found[name] = struct{}{}

		wantNullable := "NO"
		if want.nullable {
			wantNullable = "YES"
		}
		if !servingColumnTypeMatches(columnType, want.columnType) || nullable != wantNullable ||
			collation.Valid != (want.collation != "") ||
			(collation.Valid && collation.String != want.collation) ||
			!servingColumnDefaultMatches(columnDefault, want.hasDefault, want.columnDefault) ||
			!servingColumnExtraMatches(extra, want.allowDefaultGenerated) {
			actualCollation := "NULL"
			if collation.Valid {
				actualCollation = collation.String
			}
			wantCollation := "NULL"
			if want.collation != "" {
				wantCollation = want.collation
			}
			actualDefault := "NULL"
			if columnDefault.Valid {
				actualDefault = columnDefault.String
			}
			wantDefault := "NULL"
			if want.hasDefault {
				wantDefault = want.columnDefault
			}
			wantExtra := "empty"
			if want.allowDefaultGenerated {
				wantExtra = "empty or DEFAULT_GENERATED"
			}
			return fmt.Errorf(
				"%s.%s is type=%s nullable=%s collation=%s default=%s, want type=%s nullable=%s collation=%s default=%s; extra=%q, want extra=%s",
				table, name, columnType, nullable, actualCollation, actualDefault,
				want.columnType, wantNullable, wantCollation, wantDefault, extra, wantExtra,
			)
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for _, column := range expected {
		if _, ok := found[column.name]; !ok {
			return fmt.Errorf("%s is missing column %s", table, column.name)
		}
	}
	return nil
}

func servingColumnTypeMatches(actual, expected string) bool {
	if actual == expected {
		return true
	}
	// TiDB versions differ only in whether INFORMATION_SCHEMA retains the
	// deprecated integer display width. INT and INT(11), and BIGINT and
	// BIGINT(20), have the same physical signed integer semantics here.
	return (actual == "int" && expected == "int(11)") ||
		(actual == "int(11)" && expected == "int") ||
		(actual == "bigint" && expected == "bigint(20)") ||
		(actual == "bigint(20)" && expected == "bigint")
}

func servingColumnDefaultMatches(actual sql.NullString, hasExpected bool, expected string) bool {
	if !hasExpected {
		return !actual.Valid
	}
	if !actual.Valid {
		return false
	}
	if expected == "CURRENT_TIMESTAMP(6)" {
		return strings.EqualFold(actual.String, expected)
	}
	return actual.String == expected
}

func servingColumnExtraMatches(actual string, allowDefaultGenerated bool) bool {
	if actual == "" {
		return true
	}
	return allowDefaultGenerated && strings.EqualFold(actual, "DEFAULT_GENERATED")
}

func verifyServingIndexes(
	ctx context.Context,
	db *sql.DB,
	table string,
	expected []expectedServingIndex,
) error {
	rows, err := db.QueryContext(ctx, inspectServingIndexesSQL, table)
	if err != nil {
		return err
	}
	defer rows.Close()

	expectedByName := make(map[string]expectedServingIndex, len(expected))
	for _, index := range expected {
		expectedByName[index.name] = index
	}
	type observedIndex struct {
		unique  bool
		set     bool
		columns []string
	}
	observed := make(map[string]observedIndex, len(expected))
	for rows.Next() {
		var (
			name      string
			nonUnique int
			sequence  int
			column    sql.NullString
			subPart   sql.NullInt64
		)
		if err := rows.Scan(&name, &nonUnique, &sequence, &column, &subPart); err != nil {
			return err
		}
		if nonUnique != 0 && nonUnique != 1 {
			return fmt.Errorf("%s index %s has invalid NON_UNIQUE=%d", table, name, nonUnique)
		}
		if _, required := expectedByName[name]; !required {
			if nonUnique == 0 {
				return fmt.Errorf("%s has unexpected unique index %s", table, name)
			}
			return fmt.Errorf("%s has unexpected index %s", table, name)
		}
		shape := observed[name]
		unique := nonUnique == 0
		if (shape.set && shape.unique != unique) || sequence != len(shape.columns)+1 ||
			!column.Valid || subPart.Valid {
			return fmt.Errorf("%s index %s has unsafe sequence, column, uniqueness, or prefix", table, name)
		}
		shape.set = true
		shape.unique = unique
		shape.columns = append(shape.columns, column.String)
		observed[name] = shape
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for _, want := range expected {
		got, ok := observed[want.name]
		if !ok || !got.set || got.unique != want.unique || !equalServingColumns(got.columns, want.columns) {
			return fmt.Errorf(
				"%s index %s is columns=%v unique=%t, want columns=%v unique=%t",
				table, want.name, got.columns, got.unique, want.columns, want.unique,
			)
		}
	}
	return nil
}

func verifyServingForeignKeys(
	ctx context.Context,
	db *sql.DB,
	table string,
	expected []expectedServingForeignKey,
) error {
	rows, err := db.QueryContext(ctx, inspectServingForeignKeysSQL, table)
	if err != nil {
		return err
	}
	defer rows.Close()

	expectedByColumn := make(map[string]expectedServingForeignKey, len(expected))
	for _, foreignKey := range expected {
		expectedByColumn[foreignKey.column] = foreignKey
	}
	found := make(map[string]struct{}, len(expected))
	for rows.Next() {
		var (
			constraint, column, referencedSchema, referencedTable   string
			referencedColumn, currentSchema, updateRule, deleteRule string
		)
		if err := rows.Scan(
			&constraint,
			&column,
			&referencedSchema,
			&referencedTable,
			&referencedColumn,
			&currentSchema,
			&updateRule,
			&deleteRule,
		); err != nil {
			return err
		}
		want, ok := expectedByColumn[column]
		if !ok {
			return fmt.Errorf("%s has unexpected foreign key %s on column %s", table, constraint, column)
		}
		if _, duplicate := found[column]; duplicate {
			return fmt.Errorf("%s reports foreign-key column %s more than once", table, column)
		}
		found[column] = struct{}{}
		if referencedSchema != currentSchema ||
			referencedTable != want.referencedTable || referencedColumn != want.referencedColumn ||
			!restrictiveForeignKeyRule(updateRule) || !restrictiveForeignKeyRule(deleteRule) {
			return fmt.Errorf(
				"%s foreign key %s is %s -> %s.%s.%s (current database %s) update=%s delete=%s",
				table, constraint, column, referencedSchema,
				referencedTable, referencedColumn, currentSchema, updateRule, deleteRule,
			)
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for _, foreignKey := range expected {
		if _, ok := found[foreignKey.column]; !ok {
			return fmt.Errorf("%s is missing foreign key on column %s", table, foreignKey.column)
		}
	}
	return nil
}

func restrictiveForeignKeyRule(rule string) bool {
	return strings.EqualFold(rule, "RESTRICT") || strings.EqualFold(rule, "NO ACTION")
}

func equalServingColumns(first, second []string) bool {
	if len(first) != len(second) {
		return false
	}
	for i := range first {
		if first[i] != second[i] {
			return false
		}
	}
	return true
}