package persistence

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

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

// VerifyRequiredTableColumns is the shared physical-schema readiness check for
// Runtime-owned stores. Each store remains the owner of its required columns;
// this helper owns the single database inspection and type-compatibility path.
func VerifyRequiredTableColumns(ctx context.Context, db *sql.DB, table string, required map[string]string) error {
	if db == nil {
		return errors.New("table shape: database is required")
	}
	if strings.TrimSpace(table) == "" || len(required) == 0 {
		return errors.New("table shape: table and required columns are required")
	}

	rows, err := db.QueryContext(ctx, `SELECT COLUMN_NAME, LOWER(COLUMN_TYPE)
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, table)
	if err != nil {
		return fmt.Errorf("table shape: inspect %s: %w", table, err)
	}
	defer rows.Close()

	found := make(map[string]string, len(required))
	for rows.Next() {
		var name, columnType string
		if err := rows.Scan(&name, &columnType); err != nil {
			return fmt.Errorf("table shape: scan %s: %w", table, err)
		}
		found[strings.ToLower(name)] = strings.ToLower(columnType)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("table shape: iterate %s: %w", table, err)
	}

	names := make([]string, 0, len(required))
	for name := range required {
		names = append(names, name)
	}
	sort.Strings(names)
	for _, name := range names {
		expected := required[name]
		actual, ok := found[strings.ToLower(name)]
		if !ok || !compatiblePhysicalColumnType(actual, expected) {
			return fmt.Errorf("table shape: %s.%s is missing or incompatible", table, name)
		}
	}
	return nil
}

// VerifyExactTableIndex proves one named physical index has exactly the
// declared uniqueness and ordered columns. Runtime readiness uses this for
// constraints whose absence would change ownership or replay semantics; a
// column-only check is not sufficient for those invariants.
func VerifyExactTableIndex(ctx context.Context, db *sql.DB, table, index string, unique bool, requiredColumns []string) error {
	if db == nil {
		return errors.New("table shape: database is required")
	}
	return verifyExactTableIndexWithQueryer(ctx, db, table, index, unique, requiredColumns)
}

func verifyExactTableIndexWithQueryer(
	ctx context.Context,
	queryer contextQueryer,
	table, index string,
	unique bool,
	requiredColumns []string,
) error {
	if queryer == nil {
		return errors.New("table shape: queryer is required")
	}
	if strings.TrimSpace(table) == "" || strings.TrimSpace(index) == "" || len(requiredColumns) == 0 {
		return errors.New("table shape: table, index, and columns are required")
	}

	rows, err := queryer.QueryContext(ctx, exactTableIndexInspectionSQL, table, index)
	if err != nil {
		return fmt.Errorf("table shape: inspect %s index %s: %w", table, index, err)
	}
	defer rows.Close()

	found := make([]string, 0, len(requiredColumns))
	for rows.Next() {
		var nonUnique, position int
		var column string
		var prefixLength sql.NullInt64
		if err := rows.Scan(&nonUnique, &position, &column, &prefixLength); err != nil {
			return fmt.Errorf("table shape: scan %s index %s: %w", table, index, err)
		}
		if prefixLength.Valid {
			return fmt.Errorf("table shape: %s index %s uses a column prefix", table, index)
		}
		if (nonUnique == 0) != unique || position != len(found)+1 {
			return fmt.Errorf("table shape: %s index %s has incompatible uniqueness or column order", table, index)
		}
		found = append(found, column)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("table shape: iterate %s index %s: %w", table, index, err)
	}
	if len(found) != len(requiredColumns) {
		return fmt.Errorf("table shape: %s index %s columns are %v, want %v", table, index, found, requiredColumns)
	}
	for i := range requiredColumns {
		if !strings.EqualFold(found[i], requiredColumns[i]) {
			return fmt.Errorf("table shape: %s index %s columns are %v, want %v", table, index, found, requiredColumns)
		}
	}
	return nil
}

func compatiblePhysicalColumnType(actual, expected string) bool {
	actual = strings.ToLower(strings.TrimSpace(actual))
	expected = strings.ToLower(strings.TrimSpace(expected))
	if actual == expected {
		return true
	}
	if strings.Contains(expected, "(") || !strings.HasPrefix(actual, expected+"(") || !strings.HasSuffix(actual, ")") {
		return false
	}
	width := strings.TrimSuffix(strings.TrimPrefix(actual, expected+"("), ")")
	if width == "" {
		return false
	}
	for _, char := range width {
		if char < '0' || char > '9' {
			return false
		}
	}
	return true
}