package bootstrapimpl

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

	"github.com/nexuscrm/db/pkg/schemashape"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
)

// PhysicalColumnShape is the physical INFORMATION_SCHEMA view of a column.
// Table/column map keys returned by LoadPhysicalColumnShapes are lower-cased.
//
// Relocated verbatim to nexuscrm-contract/pkg/schemashape (Wave-0 SLICE B);
// this alias keeps existing importers compiling unchanged. The matcher/loader
// functions below stay here because they depend on the persistence backend.
type PhysicalColumnShape = schemashape.PhysicalColumnShape

// LoadPhysicalColumnShapes loads all physical column shapes for the current DB.
// It is shared by apply-time reconciliation and pack verify so both paths read
// physical schema through the same substrate view.
func LoadPhysicalColumnShapes(rawDB *sql.DB) (map[string]map[string]PhysicalColumnShape, error) {
	rows, err := rawDB.Query(
		"SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE()",
	)
	if err != nil {
		return nil, fmt.Errorf("query INFORMATION_SCHEMA.COLUMNS: %w", err)
	}
	defer rows.Close()

	columns := make(map[string]map[string]PhysicalColumnShape)
	for rows.Next() {
		var tableName, colName, colType, isNullable string
		var defaultVal sql.NullString
		var collationName sql.NullString
		if err := rows.Scan(&tableName, &colName, &colType, &isNullable, &defaultVal, &collationName); err != nil {
			return nil, fmt.Errorf("scan INFORMATION_SCHEMA row: %w", err)
		}
		tableKey := strings.ToLower(tableName)
		if columns[tableKey] == nil {
			columns[tableKey] = make(map[string]PhysicalColumnShape)
		}
		columns[tableKey][strings.ToLower(colName)] = PhysicalColumnShape{
			ColumnType:    strings.ToLower(colType),
			Nullable:      strings.EqualFold(isNullable, "YES"),
			DefaultVal:    defaultVal,
			CollationName: collationName,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate INFORMATION_SCHEMA: %w", err)
	}
	return columns, nil
}

func DesiredColumnType(col schema.ColumnDefinition) string {
	return persistence.ResolveColumnType(col)
}

func ColumnShapeMatches(existing PhysicalColumnShape, desired schema.ColumnDefinition) bool {
	desiredType := DesiredColumnType(desired)
	return ColumnTypesMatch(existing.ColumnType, desiredType) &&
		existing.Nullable == desired.Nullable &&
		defaultShapeMatches(existing.DefaultVal, desired.Default, desiredType) &&
		collationShapeMatches(existing.CollationName, desired.CollationName)
}

// ColumnShapeMatchesForTable applies the narrow managed-table physical
// compatibility contract before comparing the complete column shape.
//
// _System_Session.Token was historically provisioned as VARCHAR(750). The
// canonical desired definition remains VARBINARY(3000), but an exact legacy
// VARCHAR(750) representation is accepted in place without a charset-changing
// MODIFY. All non-type properties still come from the canonical definition and
// are checked normally. No other table, column, source width, or target width
// receives this compatibility.
func ColumnShapeMatchesForTable(tableName string, existing PhysicalColumnShape, desired schema.ColumnDefinition) bool {
	return ColumnShapeMatches(existing, physicalReconciliationColumnDefinition(tableName, existing, desired))
}

func physicalReconciliationColumnDefinition(tableName string, existing PhysicalColumnShape, desired schema.ColumnDefinition) schema.ColumnDefinition {
	if !isLegacySessionTokenPhysicalRepresentation(tableName, existing, desired) {
		return desired
	}
	compatible := desired
	compatible.Type = "VARCHAR(750)"
	compatible.Length = 0
	return compatible
}

func isLegacySessionTokenPhysicalRepresentation(tableName string, existing PhysicalColumnShape, desired schema.ColumnDefinition) bool {
	return strings.EqualFold(strings.TrimSpace(tableName), constants.TableSession) &&
		strings.EqualFold(strings.TrimSpace(desired.Name), constants.FieldSysSession_Token) &&
		normalizedTypeMatch(existing.ColumnType, "VARCHAR(750)") &&
		normalizedTypeMatch(DesiredColumnType(desired), "VARBINARY(3000)")
}

func collationShapeMatches(existing sql.NullString, desired string) bool {
	desired = strings.TrimSpace(desired)
	if desired == "" {
		return true
	}
	if !existing.Valid {
		return false
	}
	return strings.EqualFold(existing.String, desired)
}

func ColumnTypesMatch(existingType, desiredType string) bool {
	return normalizedTypeMatch(existingType, desiredType)
}

func isAutoColumnTypeChangeSupported(existingType, desiredType string) bool {
	if normalizedTypeMatch(existingType, desiredType) {
		return true
	}
	if isJSONToTextChange(existingType, desiredType) {
		return true
	}
	if isVarcharChange(existingType, desiredType) {
		return true
	}
	if isSafeTemporalPrecisionWidening(existingType, desiredType) {
		return true
	}
	return isSafeNumericWidening(existingType, desiredType)
}

// isJSONToTextChange recognizes an intentional removal of the database JSON
// codec. The conversion preserves every existing JSON value as serialized text
// and allows the declared owner to preserve numeric lexemes on future writes.
// ReconcileTableColumns separately proves that retained rows fit the target
// byte capacity before it executes the ALTER.
func isJSONToTextChange(existingType, desiredType string) bool {
	if normalizeColumnType(existingType) != "json" {
		return false
	}
	_, ok := textColumnByteCapacity(desiredType)
	return ok
}

func textColumnByteCapacity(colType string) (int64, bool) {
	switch normalizeColumnType(colType) {
	case "tinytext":
		return 255, true
	case "text":
		return 65_535, true
	case "mediumtext":
		return 16_777_215, true
	case "longtext":
		return 4_294_967_295, true
	default:
		return 0, false
	}
}

// normalizedTypeMatch returns true if two column types are equivalent after normalizing
// MySQL/TiDB type aliases (e.g., BOOLEAN -> tinyint(1), INT -> int(11)).
func normalizedTypeMatch(existingType, desiredType string) bool {
	a := normalizeColumnType(existingType)
	b := normalizeColumnType(desiredType)
	return strings.EqualFold(a, b)
}

func defaultShapeMatches(existing sql.NullString, desired, desiredType string) bool {
	// JSON / TEXT / BLOB cannot carry a physical DEFAULT clause (MySQL/TiDB
	// Error 1101). The metadata default is still meaningful -- persistence
	// reads _System_Field.DefaultValue at insert time -- but it must not
	// surface as physical column drift. Aligns with the apply path in
	// ReconcileTableColumns and persistence.SchemaRepository, which both
	// skip the DEFAULT clause for these types.
	if !persistence.SqlTypeAllowsDefaultClause(desiredType) {
		return true
	}
	if desired == "" {
		return !existing.Valid
	}
	if !existing.Valid {
		return false
	}
	existingDefault := normalizeColumnDefault(existing.String)
	desiredDefault := normalizeColumnDefault(desired)
	if normalizeColumnType(desiredType) == "tinyint(1)" {
		existingBool, existingOK := normalizeBooleanColumnDefault(existingDefault)
		desiredBool, desiredOK := normalizeBooleanColumnDefault(desiredDefault)
		if existingOK && desiredOK {
			return existingBool == desiredBool
		}
	}
	if isTemporalColumnType(desiredType) {
		existingTimestamp, existingOK := normalizeTimestampColumnDefault(existingDefault)
		desiredTimestamp, desiredOK := normalizeTimestampColumnDefault(desiredDefault)
		if existingOK && desiredOK {
			return existingTimestamp == desiredTimestamp
		}
	}
	return strings.EqualFold(existingDefault, desiredDefault)
}

func normalizeColumnDefault(v string) string {
	trimmed := strings.TrimSpace(v)
	if len(trimmed) >= 2 {
		first := trimmed[0]
		last := trimmed[len(trimmed)-1]
		if (first == '\'' && last == '\'') || (first == '"' && last == '"') {
			trimmed = strings.TrimSpace(trimmed[1 : len(trimmed)-1])
		}
	}
	return trimmed
}

func normalizeBooleanColumnDefault(v string) (bool, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "1", "true":
		return true, true
	case "0", "false":
		return false, true
	default:
		return false, false
	}
}

func isTemporalColumnType(t string) bool {
	normalized := normalizeColumnType(t)
	return strings.HasPrefix(normalized, "datetime") || strings.HasPrefix(normalized, "timestamp")
}

func normalizeTimestampColumnDefault(v string) (string, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "current_timestamp", "current_timestamp()":
		return "current_timestamp", true
	default:
		return "", false
	}
}

func printableDefault(v sql.NullString) string {
	if !v.Valid {
		return "<null>"
	}
	return v.String
}

func printableDesiredDefault(v string) string {
	if v == "" {
		return "<null>"
	}
	return v
}

// normalizeColumnType maps MySQL/TiDB type aliases to their canonical INFORMATION_SCHEMA form.
func normalizeColumnType(t string) string {
	lower := strings.ToLower(strings.TrimSpace(t))
	switch lower {
	case "boolean", "bool":
		return "tinyint(1)"
	case "int":
		return "int(11)"
	case "bigint":
		return "bigint(20)"
	case "smallint":
		return "smallint(6)"
	case "tinyint":
		return "tinyint(4)"
	case "integer":
		return "int(11)"
	default:
		return lower
	}
}

// isVarcharChange returns true if both types are VARCHAR (regardless of width).
func isVarcharChange(existing, desired string) bool {
	return extractVarcharWidth(existing) > 0 && extractVarcharWidth(desired) > 0
}

// isVarcharNarrowing returns true if the type change is VARCHAR(X) -> VARCHAR(Y) where Y < X.
func isVarcharNarrowing(existing, desired string) bool {
	existWidth := extractVarcharWidth(existing)
	desiredWidth := extractVarcharWidth(desired)
	if existWidth <= 0 || desiredWidth <= 0 {
		return false
	}
	return desiredWidth < existWidth
}

// extractVarcharWidth extracts the width from a VARCHAR(N) type string.
// Returns -1 if not a VARCHAR type.
func extractVarcharWidth(colType string) int {
	lower := strings.ToLower(colType)
	if !strings.HasPrefix(lower, "varchar(") {
		return -1
	}
	inner := strings.TrimPrefix(lower, "varchar(")
	inner = strings.TrimSuffix(inner, ")")
	var width int
	if _, err := fmt.Sscanf(inner, "%d", &width); err != nil {
		return -1
	}
	return width
}

func isVarcharToVarbinaryChange(existing, desired string) bool {
	return extractVarcharWidth(existing) > 0 && extractVarbinaryWidth(desired) > 0
}

// isSafeVarcharToVarbinaryChange recognizes when a text-to-byte target
// preserves the full worst-case utf8mb4 byte capacity of the existing VARCHAR.
// Capacity preservation is necessary but not sufficient for an online change:
// TiDB still requires an explicit writer-quiesced migration for the charset and
// any index replacement.
func isSafeVarcharToVarbinaryChange(existing, desired string) bool {
	existingWidth := extractVarcharWidth(existing)
	desiredWidth := extractVarbinaryWidth(desired)
	return existingWidth > 0 && desiredWidth >= existingWidth*4
}

// isSafeTemporalPrecisionWidening admits only a widening of fractional-second
// precision within the same temporal storage family. DATETIME and TIMESTAMP
// have different timezone/default semantics, so neither may be substituted for
// the other by generic reconciliation. A wider fractional precision preserves
// every existing value; a narrower precision could collapse distinct lease or
// ordering instants and must use an explicit migration instead.
func isSafeTemporalPrecisionWidening(existing, desired string) bool {
	existingFamily, existingPrecision, existingOK := temporalColumnPrecision(existing)
	desiredFamily, desiredPrecision, desiredOK := temporalColumnPrecision(desired)
	return existingOK && desiredOK &&
		existingFamily == desiredFamily &&
		desiredPrecision >= existingPrecision
}

func temporalColumnPrecision(colType string) (family string, precision int, ok bool) {
	normalized := normalizeColumnType(colType)
	for _, candidate := range []string{"datetime", "timestamp"} {
		if normalized == candidate {
			return candidate, 0, true
		}
		prefix := candidate + "("
		if !strings.HasPrefix(normalized, prefix) || !strings.HasSuffix(normalized, ")") {
			continue
		}
		precision, err := strconv.Atoi(strings.TrimSuffix(strings.TrimPrefix(normalized, prefix), ")"))
		if err != nil || precision < 0 || precision > 6 {
			return "", 0, false
		}
		return candidate, precision, true
	}
	return "", 0, false
}

// extractVarbinaryWidth extracts the width from a VARBINARY(N) type string.
// Returns -1 if the type is not parameterized VARBINARY.
func extractVarbinaryWidth(colType string) int {
	lower := strings.ToLower(strings.TrimSpace(colType))
	if !strings.HasPrefix(lower, "varbinary(") {
		return -1
	}
	inner := strings.TrimSuffix(strings.TrimPrefix(lower, "varbinary("), ")")
	width, err := strconv.Atoi(strings.TrimSpace(inner))
	if err != nil || width <= 0 {
		return -1
	}
	return width
}

type numericColumnShape struct {
	integerDigits int
	scale         int
	ok            bool
}

func isSafeNumericWidening(existing, desired string) bool {
	existingNumeric := parseNumericColumnShape(existing)
	desiredNumeric := parseNumericColumnShape(desired)
	if !existingNumeric.ok || !desiredNumeric.ok {
		return false
	}
	return desiredNumeric.integerDigits >= existingNumeric.integerDigits &&
		desiredNumeric.scale >= existingNumeric.scale
}

func parseNumericColumnShape(colType string) numericColumnShape {
	normalized := normalizeColumnType(colType)
	normalized = strings.TrimSpace(strings.TrimSuffix(normalized, " zerofill"))
	unsigned := strings.HasSuffix(normalized, " unsigned")
	normalized = strings.TrimSpace(strings.TrimSuffix(normalized, " unsigned"))

	switch {
	case strings.HasPrefix(normalized, "decimal("), strings.HasPrefix(normalized, "numeric("):
		precision, scale, ok := parseDecimalShape(normalized)
		if !ok || precision < scale {
			return numericColumnShape{}
		}
		return numericColumnShape{integerDigits: precision - scale, scale: scale, ok: true}
	case strings.HasPrefix(normalized, "tinyint("):
		if normalized == "tinyint(1)" {
			return numericColumnShape{}
		}
		return integerColumnShape(3, unsigned)
	case strings.HasPrefix(normalized, "smallint("):
		return integerColumnShape(5, unsigned)
	case strings.HasPrefix(normalized, "mediumint("):
		return integerColumnShape(7, unsigned)
	case strings.HasPrefix(normalized, "int("):
		return integerColumnShape(10, unsigned)
	case strings.HasPrefix(normalized, "bigint("):
		return integerColumnShape(19, unsigned)
	default:
		return numericColumnShape{}
	}
}

func integerColumnShape(signedDigits int, unsigned bool) numericColumnShape {
	if unsigned {
		signedDigits++
	}
	return numericColumnShape{integerDigits: signedDigits, scale: 0, ok: true}
}

func parseDecimalShape(normalized string) (int, int, bool) {
	open := strings.Index(normalized, "(")
	close := strings.LastIndex(normalized, ")")
	if open == -1 || close <= open {
		return 0, 0, false
	}
	parts := strings.Split(normalized[open+1:close], ",")
	if len(parts) != 2 {
		return 0, 0, false
	}
	precision, err := strconv.Atoi(strings.TrimSpace(parts[0]))
	if err != nil {
		return 0, 0, false
	}
	scale, err := strconv.Atoi(strings.TrimSpace(parts[1]))
	if err != nil {
		return 0, 0, false
	}
	return precision, scale, true
}