package persistence

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

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

const (
	connectorV2ReceiptTable                 = "_System_ConnectorV2Receipt"
	connectorV2ReceiptProtocolColumn        = "ReceiptProtocolVersion"
	connectorV2ReceiptAccountIndex          = "idx_connector_v2_receipt_account"
	connectorV2ReceiptAdmissionKeyTable     = "_System_ConnectorV2ReceiptAdmissionKey"
	connectorV2ReceiptTransportIndex        = "ux_connector_v2_receipt_key_transport"
	connectorV2ReceiptTransportCutoverIndex = "ux_connector_v2_receipt_key_transport_cutover"
)

const connectorV2ReceiptProtocolColumnInspectionSQL = `SELECT COLUMN_TYPE, COLLATION_NAME, IS_NULLABLE
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`

const connectorV2ReceiptProtocolOversizeProbeSQL = `SELECT 1
FROM _System_ConnectorV2Receipt
WHERE CHAR_LENGTH(ReceiptProtocolVersion) > 32
LIMIT 1`

const connectorV2ReceiptProtocolNarrowSQL = `ALTER TABLE _System_ConnectorV2Receipt
MODIFY COLUMN ReceiptProtocolVersion VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL`

var (
	connectorV2ReceiptAccountIndexColumns = []string{
		"ConnectorAccountId",
		"Id",
	}
	connectorV2ReceiptTransportIdentityColumns = []string{
		"ConnectorAccountId",
		"IngressChannel",
		"TransportIdempotencyKey",
	}
	connectorV2ReceiptLegacyTransportIdentityColumns = []string{
		"ServingScopeDigest",
		"ConnectorAccountId",
		"IngressChannel",
		"TransportIdempotencyKey",
	}
)

const connectorV2ReceiptTransportDuplicateProbeSQL = `SELECT 1
FROM _System_ConnectorV2ReceiptAdmissionKey
GROUP BY ConnectorAccountId, IngressChannel, TransportIdempotencyKey
HAVING COUNT(*) > 1
LIMIT 1`

const connectorV2ReceiptTransportCutoverIndexAddSQL = `ALTER TABLE _System_ConnectorV2ReceiptAdmissionKey
ADD UNIQUE INDEX ux_connector_v2_receipt_key_transport_cutover (ConnectorAccountId, IngressChannel, TransportIdempotencyKey)`

const connectorV2ReceiptLegacyTransportIndexDropSQL = `ALTER TABLE _System_ConnectorV2ReceiptAdmissionKey
DROP INDEX ux_connector_v2_receipt_key_transport`

const connectorV2ReceiptTransportCutoverIndexRenameSQL = `ALTER TABLE _System_ConnectorV2ReceiptAdmissionKey
RENAME INDEX ux_connector_v2_receipt_key_transport_cutover TO ux_connector_v2_receipt_key_transport`

const connectorV2ReceiptAccountIndexAddSQL = `ALTER TABLE _System_ConnectorV2Receipt
ADD INDEX idx_connector_v2_receipt_account (ConnectorAccountId, Id)`

func declaresConnectorV2ReceiptProtocolWidth(def schema.TableDefinition) bool {
	if def.TableName != connectorV2ReceiptTable {
		return false
	}
	for _, column := range def.Columns {
		if column.Name != connectorV2ReceiptProtocolColumn {
			continue
		}
		return strings.EqualFold(column.Type, "VARCHAR(32)") &&
			!column.Nullable &&
			column.CollationName == "utf8mb4_bin"
	}
	return false
}

func (r *SchemaRepository) reconcileConnectorV2ReceiptProtocolWidth(ctx context.Context) error {
	columnType, collation, nullable, err := inspectConnectorV2ReceiptProtocolColumn(ctx, r.db)
	if err != nil {
		return fmt.Errorf("reconcile connector V2 receipt protocol width: %w", err)
	}
	if columnType == "varchar(32)" && collation == "utf8mb4_bin" && nullable == "NO" {
		return nil
	}
	if columnType != "varchar(255)" || collation != "utf8mb4_bin" || nullable != "NO" {
		return fmt.Errorf(
			"reconcile connector V2 receipt protocol width: physical %s.%s has incompatible shape (%s/%s/nullable=%s)",
			connectorV2ReceiptTable,
			connectorV2ReceiptProtocolColumn,
			columnType,
			collation,
			nullable,
		)
	}

	var oversize int
	err = r.db.QueryRowContext(ctx, connectorV2ReceiptProtocolOversizeProbeSQL).Scan(&oversize)
	switch {
	case err == nil:
		return fmt.Errorf("reconcile connector V2 receipt protocol width: retained ReceiptProtocolVersion exceeds the canonical VARCHAR(32) bound")
	case err != sql.ErrNoRows:
		return fmt.Errorf("reconcile connector V2 receipt protocol width: inspect retained protocol values: %w", err)
	}
	if _, err := r.db.ExecContext(ctx, connectorV2ReceiptProtocolNarrowSQL); err != nil {
		return fmt.Errorf("reconcile connector V2 receipt protocol width: narrow exact legacy column: %w", err)
	}

	columnType, collation, nullable, err = inspectConnectorV2ReceiptProtocolColumn(ctx, r.db)
	if err != nil {
		return fmt.Errorf("reconcile connector V2 receipt protocol width: verify migrated column: %w", err)
	}
	if columnType != "varchar(32)" || collation != "utf8mb4_bin" || nullable != "NO" {
		return fmt.Errorf(
			"reconcile connector V2 receipt protocol width: migrated %s.%s has incompatible shape (%s/%s/nullable=%s)",
			connectorV2ReceiptTable,
			connectorV2ReceiptProtocolColumn,
			columnType,
			collation,
			nullable,
		)
	}
	return nil
}

func inspectConnectorV2ReceiptProtocolColumn(
	ctx context.Context,
	db *sql.DB,
) (string, string, string, error) {
	var columnType, collation, nullable string
	if err := db.QueryRowContext(
		ctx,
		connectorV2ReceiptProtocolColumnInspectionSQL,
		connectorV2ReceiptTable,
		connectorV2ReceiptProtocolColumn,
	).Scan(&columnType, &collation, &nullable); err != nil {
		return "", "", "", err
	}
	return strings.ToLower(columnType), collation, nullable, nil
}

func declaresConnectorV2ReceiptAccountIndex(def schema.TableDefinition) bool {
	if def.TableName != connectorV2ReceiptTable {
		return false
	}
	for _, index := range def.Indices {
		if index.Name == connectorV2ReceiptAccountIndex {
			return !index.Unique && equalExactColumnNames(index.Columns, connectorV2ReceiptAccountIndexColumns)
		}
	}
	return false
}

func declaresConnectorV2ReceiptTransportIdentity(def schema.TableDefinition) bool {
	if def.TableName != connectorV2ReceiptAdmissionKeyTable {
		return false
	}
	for _, index := range def.Indices {
		if index.Name == connectorV2ReceiptTransportIndex {
			return index.Unique && equalExactColumnNames(index.Columns, connectorV2ReceiptTransportIdentityColumns)
		}
	}
	return false
}

func (r *SchemaRepository) reconcileConnectorV2ReceiptTransportIdentityIndex(ctx context.Context) error {
	columns, unique, hasPrefix, err := inspectManagedPhysicalIndex(
		ctx,
		r.db,
		connectorV2ReceiptAdmissionKeyTable,
		connectorV2ReceiptTransportIndex,
	)
	if err != nil {
		return fmt.Errorf("reconcile connector V2 receipt transport identity index: %w", err)
	}
	cutoverColumns, cutoverUnique, cutoverHasPrefix, err := inspectManagedPhysicalIndex(
		ctx,
		r.db,
		connectorV2ReceiptAdmissionKeyTable,
		connectorV2ReceiptTransportCutoverIndex,
	)
	if err != nil {
		return fmt.Errorf("reconcile connector V2 receipt transport identity index: inspect resumable cutover index: %w", err)
	}
	canonicalIsNew := unique && !hasPrefix && equalExactColumnNames(columns, connectorV2ReceiptTransportIdentityColumns)
	canonicalIsLegacy := unique && !hasPrefix && equalExactColumnNames(columns, connectorV2ReceiptLegacyTransportIdentityColumns)
	canonicalIsMissing := len(columns) == 0
	cutoverIsNew := cutoverUnique && !cutoverHasPrefix && equalExactColumnNames(cutoverColumns, connectorV2ReceiptTransportIdentityColumns)
	cutoverIsMissing := len(cutoverColumns) == 0

	if canonicalIsNew {
		if !cutoverIsMissing {
			return fmt.Errorf("reconcile connector V2 receipt transport identity index: canonical and cutover indexes are both present")
		}
		return nil
	}
	if !canonicalIsLegacy && !canonicalIsMissing {
		return fmt.Errorf(
			"reconcile connector V2 receipt transport identity index: physical %s.%s has incompatible shape (unique=%t, prefix=%t, columns=%v)",
			connectorV2ReceiptAdmissionKeyTable,
			connectorV2ReceiptTransportIndex,
			unique,
			hasPrefix,
			columns,
		)
	}
	if !cutoverIsMissing && !cutoverIsNew {
		return fmt.Errorf(
			"reconcile connector V2 receipt transport identity index: physical cutover index has incompatible shape (unique=%t, prefix=%t, columns=%v)",
			cutoverUnique,
			cutoverHasPrefix,
			cutoverColumns,
		)
	}
	if canonicalIsMissing && cutoverIsMissing {
		return fmt.Errorf("reconcile connector V2 receipt transport identity index: canonical and cutover indexes are both absent")
	}

	if cutoverIsMissing {
		var duplicate int
		err = r.db.QueryRowContext(ctx, connectorV2ReceiptTransportDuplicateProbeSQL).Scan(&duplicate)
		switch {
		case err == nil:
			return fmt.Errorf(
				"reconcile connector V2 receipt transport identity index: retained rows conflict on (ConnectorAccountId, IngressChannel, TransportIdempotencyKey); remove the duplicate authority before retrying schema apply",
			)
		case err != sql.ErrNoRows:
			return fmt.Errorf("reconcile connector V2 receipt transport identity index: inspect retained transport identities: %w", err)
		}

		// The new constraint is deliberately created first while the legacy
		// scope-inclusive constraint still has its canonical name. If this ADD
		// loses a race to a writer or the process stops, the old schema remains
		// intact and a later halted apply can safely retry.
		if _, err := r.db.ExecContext(ctx, connectorV2ReceiptTransportCutoverIndexAddSQL); err != nil {
			return fmt.Errorf("reconcile connector V2 receipt transport identity index: add scope-independent cutover index while retaining the legacy constraint: %w", err)
		}
		if err := VerifyExactTableIndex(
			ctx,
			r.db,
			connectorV2ReceiptAdmissionKeyTable,
			connectorV2ReceiptTransportCutoverIndex,
			true,
			connectorV2ReceiptTransportIdentityColumns,
		); err != nil {
			return fmt.Errorf("reconcile connector V2 receipt transport identity index: verify cutover index: %w", err)
		}
	}

	if canonicalIsLegacy {
		if _, err := r.db.ExecContext(ctx, connectorV2ReceiptLegacyTransportIndexDropSQL); err != nil {
			return fmt.Errorf("reconcile connector V2 receipt transport identity index: remove superseded scope-inclusive index after cutover constraint is active: %w", err)
		}
	}
	if _, err := r.db.ExecContext(ctx, connectorV2ReceiptTransportCutoverIndexRenameSQL); err != nil {
		return fmt.Errorf("reconcile connector V2 receipt transport identity index: promote cutover index to canonical name: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		r.db,
		connectorV2ReceiptAdmissionKeyTable,
		connectorV2ReceiptTransportIndex,
		true,
		connectorV2ReceiptTransportIdentityColumns,
	); err != nil {
		return fmt.Errorf("reconcile connector V2 receipt transport identity index: verify migrated index: %w", err)
	}
	return nil
}

func (r *SchemaRepository) reconcileConnectorV2ReceiptAccountIndex(ctx context.Context) error {
	columns, unique, hasPrefix, err := inspectManagedPhysicalIndex(
		ctx,
		r.db,
		connectorV2ReceiptTable,
		connectorV2ReceiptAccountIndex,
	)
	if err != nil {
		return fmt.Errorf("reconcile connector V2 receipt account index: %w", err)
	}
	if len(columns) != 0 {
		if !unique && !hasPrefix && equalExactColumnNames(columns, connectorV2ReceiptAccountIndexColumns) {
			return nil
		}
		return fmt.Errorf(
			"reconcile connector V2 receipt account index: physical %s.%s has incompatible shape (unique=%t, prefix=%t, columns=%v)",
			connectorV2ReceiptTable,
			connectorV2ReceiptAccountIndex,
			unique,
			hasPrefix,
			columns,
		)
	}

	if _, err := r.db.ExecContext(ctx, connectorV2ReceiptAccountIndexAddSQL); err != nil {
		return fmt.Errorf("reconcile connector V2 receipt account index: add account-leading retained-authority index: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		r.db,
		connectorV2ReceiptTable,
		connectorV2ReceiptAccountIndex,
		false,
		connectorV2ReceiptAccountIndexColumns,
	); err != nil {
		return fmt.Errorf("reconcile connector V2 receipt account index: verify migrated index: %w", err)
	}
	return nil
}