package serving

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

const (
	executionTable        = "_System_ConnectorV2Execution"
	executionAttemptTable = "_System_ConnectorV2ExecutionAttempt"
)

var executionColumns = []expectedServingColumn{
	{name: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ExecutionProtocolVersion", columnType: "varchar(64)", collation: "utf8mb4_bin"},
	{name: "ServingScopeDigest", columnType: "char(64)", collation: "utf8mb4_bin"},
	{name: "ConnectorAccountId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "AccountServingGateId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "AccountServingFenceToken", columnType: "bigint(20)"},
	{name: "ServingEpochId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ServingEpochDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "ServingEpochBindingId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "PackGenerationId", columnType: "varchar(255)", collation: "utf8mb4_unicode_ci"},
	{name: "DefinitionRef", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "DefinitionVersion", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "CanonicalInputs", columnType: "mediumblob"},
	{name: "ExternalEffectsMode", columnType: "varchar(32)", collation: "utf8mb4_bin"},
	{name: "InputDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "ExecutionRefDigest", columnType: "char(71)", collation: "utf8mb4_bin"},
	{name: "RuntimeCommandId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "State", columnType: "varchar(32)", collation: "utf8mb4_bin", hasDefault: true, columnDefault: "ready"},
	{name: "ClaimFenceToken", columnType: "bigint(20)", hasDefault: true, columnDefault: "0"},
	{name: "ClaimHolderId", columnType: "varchar(255)", nullable: true, collation: "utf8mb4_bin"},
	{name: "ClaimLeaseUntil", columnType: "datetime(6)", nullable: true},
	{name: "ResultDigest", columnType: "char(71)", nullable: true, collation: "utf8mb4_bin"},
	{name: "RuntimeCommitReceipt", columnType: "char(71)", nullable: true, collation: "utf8mb4_bin"},
	{name: "QuarantineReasonCode", columnType: "varchar(64)", 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 executionIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"Id"}},
	{name: "ux_connector_v2_execution_ref", unique: true, columns: []string{"ExecutionRefDigest"}},
	{name: "ux_connector_v2_execution_runtime_command", unique: true, columns: []string{"RuntimeCommandId"}},
	{name: "idx_connector_v2_execution_claim", columns: []string{"ConnectorAccountId", "State", "ClaimLeaseUntil", "Id"}},
	{name: "idx_connector_v2_execution_epoch", columns: []string{"ServingEpochId", "State", "Id"}},
	{name: "idx_connector_v2_execution_gate", columns: []string{"AccountServingGateId", "State", "Id"}},
}

var executionAttemptColumns = []expectedServingColumn{
	{name: "Id", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "ExecutionId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "AttemptOrdinal", columnType: "bigint(20)"},
	{name: "ClaimFenceToken", columnType: "bigint(20)"},
	{name: "ClaimHolderId", columnType: "varchar(255)", collation: "utf8mb4_bin"},
	{name: "LeaseUntil", columnType: "datetime(6)"},
	{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 executionAttemptIndexes = []expectedServingIndex{
	{name: "PRIMARY", unique: true, columns: []string{"Id"}},
	{name: "ux_connector_v2_execution_attempt_ordinal", unique: true, columns: []string{"ExecutionId", "AttemptOrdinal"}},
	{name: "ux_connector_v2_execution_attempt_fence", unique: true, columns: []string{"ExecutionId", "ClaimFenceToken"}},
	{name: "idx_connector_v2_execution_attempt_holder", columns: []string{"ClaimHolderId", "LeaseUntil", "Id"}},
}

var executionTableSchemas = []servingTableSchema{
	{
		name:    executionTable,
		columns: executionColumns,
		indexes: executionIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "AccountServingGateId", referencedTable: accountServingGateTable, referencedColumn: "Id"},
			{column: "ServingEpochId", referencedTable: servingEpochTable, referencedColumn: "Id"},
			{column: "ServingEpochBindingId", referencedTable: servingEpochBindingOwnerTable, referencedColumn: "ServingEpochBindingId"},
			{column: "PackGenerationId", referencedTable: "_System_PackGenerationConnectorBindingDefinitionSet", referencedColumn: "PackGenerationId"},
		},
	},
	{
		name:    executionAttemptTable,
		columns: executionAttemptColumns,
		indexes: executionAttemptIndexes,
		foreignKeys: []expectedServingForeignKey{
			{column: "ExecutionId", referencedTable: executionTable, referencedColumn: "Id"},
		},
	},
}

// verifyExecutionSchema is a read-only readiness check. Core owns physical
// DDL; Connector neither creates nor repairs a compatibility shape.
func verifyExecutionSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return executionSchemaError("database or context is unavailable", nil)
	}
	if err := verifyServingSchema(ctx, db); err != nil {
		return executionSchemaError("verify exact ServingEpoch schema", err)
	}
	if err := verifyAccountServingGateSchema(ctx, db); err != nil {
		return executionSchemaError("verify exact account gate schema", err)
	}
	for _, table := range executionTableSchemas {
		if err := verifyServingColumns(ctx, db, table.name, table.columns); err != nil {
			return executionSchemaError("verify "+table.name+" columns", err)
		}
		expectedIndexes := table.indexes
		if table.name == executionTable {
			var err error
			expectedIndexes, err = executionIndexesWithPhysicalForeignKeySupport(
				ctx,
				db,
				table.indexes,
				[]string{"ServingEpochBindingId", "PackGenerationId"},
			)
			if err != nil {
				return executionSchemaError("inspect "+table.name+" indexes", err)
			}
		}
		if err := verifyServingIndexes(ctx, db, table.name, expectedIndexes); err != nil {
			return executionSchemaError("verify "+table.name+" indexes", err)
		}
		if err := verifyServingForeignKeys(ctx, db, table.name, table.foreignKeys); err != nil {
			return executionSchemaError("verify "+table.name+" foreign keys", err)
		}
	}

	// These tables are owned outside Connector. Probe only the exact columns the
	// authority lock reads; do not claim or duplicate their full schema contract.
	for _, probe := range []struct {
		name  string
		query string
	}{
		{name: "PackGeneration", query: `SELECT Id, PackName, PackInstallId, AppliedPayloadHash
FROM _System_PackGeneration WHERE 1 = 0`},
		{name: "PackGeneration definition set", query: `SELECT PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest
FROM _System_PackGenerationConnectorBindingDefinitionSet WHERE 1 = 0`},
	} {
		rows, err := db.QueryContext(ctx, probe.query)
		if err != nil {
			return executionSchemaError("verify "+probe.name+" read shape", err)
		}
		if err := rows.Close(); err != nil {
			return executionSchemaError("close "+probe.name+" schema probe", err)
		}
	}
	return nil
}

// TiDB creates one non-unique single-column child index when a physical
// foreign key is not already covered by a catalog-authored index. Core does not
// assign these implementation index names. Admit only those two exact support
// shapes, then pass their observed names into the ordinary strict verifier.
func executionIndexesWithPhysicalForeignKeySupport(
	ctx context.Context,
	db *sql.DB,
	expected []expectedServingIndex,
	supportColumns []string,
) ([]expectedServingIndex, error) {
	expectedNames := make(map[string]struct{}, len(expected))
	coveredSupport := make(map[string]bool, len(supportColumns))
	for _, column := range supportColumns {
		coveredSupport[column] = false
	}
	for _, index := range expected {
		expectedNames[index.name] = struct{}{}
		if len(index.columns) > 0 {
			if _, ok := coveredSupport[index.columns[0]]; ok {
				coveredSupport[index.columns[0]] = true
			}
		}
	}
	type physicalIndex struct {
		nonUnique int
		columns   []string
	}
	physical := make(map[string]physicalIndex)
	rows, err := db.QueryContext(ctx, inspectServingIndexesSQL, executionTable)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	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 nil, err
		}
		if _, canonical := expectedNames[name]; canonical {
			continue
		}
		shape := physical[name]
		if nonUnique != 1 || sequence != len(shape.columns)+1 || !column.Valid || subPart.Valid {
			return nil, fmt.Errorf("unexpected index %s is not an exact non-unique full-column support index", name)
		}
		shape.nonUnique = nonUnique
		shape.columns = append(shape.columns, column.String)
		physical[name] = shape
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	result := append([]expectedServingIndex(nil), expected...)
	names := make([]string, 0, len(physical))
	for name := range physical {
		names = append(names, name)
	}
	sort.Strings(names)
	for _, name := range names {
		shape := physical[name]
		if len(shape.columns) != 1 {
			return nil, fmt.Errorf("unexpected index %s has columns %v", name, shape.columns)
		}
		column := shape.columns[0]
		covered, allowed := coveredSupport[column]
		if !allowed || covered {
			return nil, fmt.Errorf("unexpected index %s has unowned or duplicate support column %s", name, column)
		}
		coveredSupport[column] = true
		result = append(result, expectedServingIndex{name: name, columns: []string{column}})
	}
	for column, covered := range coveredSupport {
		if !covered {
			return nil, fmt.Errorf("foreign-key column %s lacks a physical support index", column)
		}
	}
	return result, nil
}

func executionSchemaError(operation string, cause error) error {
	if cause == nil {
		return fmt.Errorf("%w: %w: %s", ErrSchemaUnavailable, ErrExecutionSchemaUnavailable, operation)
	}
	return fmt.Errorf("%w: %w: %s: %w", ErrSchemaUnavailable, ErrExecutionSchemaUnavailable, operation, cause)
}