package persistence

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

	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

type managedRowMutationProtection struct {
	storageTableName string
	fields           []schemacatalog.SystemTableRowMutationProtectionField
}

func managedRowMutationProtectionFor(
	ctx context.Context,
	exec Executor,
	tableName string,
	schema *models.ObjectMetadata,
) (managedRowMutationProtection, error) {
	if schema == nil && !constants.IsManagedSystemTable(tableName) {
		return managedRowMutationProtection{}, fmt.Errorf(
			"authoritative schema is required to resolve row mutation protection for logical table %s",
			tableName,
		)
	}
	tableNames, err := canonicalWriteProtectionTableNames(tableName, schema)
	if err != nil {
		return managedRowMutationProtection{}, err
	}

	var declared schemacatalog.SystemTableRowMutationProtection
	found := false
	for _, canonicalTableName := range tableNames {
		fact, ok, err := schemacatalog.SystemTableRowMutationProtectionFor(canonicalTableName)
		if err != nil {
			return managedRowMutationProtection{}, err
		}
		if !ok {
			continue
		}
		if found && !sameManagedRowMutationProtection(declared, fact) {
			return managedRowMutationProtection{}, fmt.Errorf(
				"conflicting row mutation protection declarations for %s",
				tableName,
			)
		}
		declared = fact
		found = true
	}
	if !found {
		return managedRowMutationProtection{}, nil
	}
	return managedRowMutationProtectionFromCatalog(ctx, exec, declared)
}

func managedRowMutationProtectionFromCatalog(
	ctx context.Context,
	exec Executor,
	declared schemacatalog.SystemTableRowMutationProtection,
) (managedRowMutationProtection, error) {
	if !isValidFieldName(declared.TableName) {
		return managedRowMutationProtection{}, fmt.Errorf(
			"invalid row mutation protection storage table %q",
			declared.TableName,
		)
	}
	for _, field := range declared.Fields {
		if !isValidFieldName(field.PhysicalName) {
			return managedRowMutationProtection{}, fmt.Errorf(
				"invalid row mutation protection storage field %q for %s",
				field.PhysicalName,
				declared.TableName,
			)
		}
	}

	if err := inspectManagedRowMutationStorage(ctx, exec, declared.TableName, declared); err != nil {
		return managedRowMutationProtection{}, err
	}
	return managedRowMutationProtection{
		storageTableName: declared.TableName,
		fields:           append([]schemacatalog.SystemTableRowMutationProtectionField(nil), declared.Fields...),
	}, nil
}

func sameManagedRowMutationProtection(
	left, right schemacatalog.SystemTableRowMutationProtection,
) bool {
	if left.TableName != right.TableName ||
		left.Policy != right.Policy ||
		left.RequiredUniqueIndex.Name != right.RequiredUniqueIndex.Name ||
		len(left.RequiredUniqueIndex.Columns) != len(right.RequiredUniqueIndex.Columns) ||
		len(left.ScopeFields) != len(right.ScopeFields) ||
		len(left.Fields) != len(right.Fields) {
		return false
	}
	for i := range left.RequiredUniqueIndex.Columns {
		if left.RequiredUniqueIndex.Columns[i] != right.RequiredUniqueIndex.Columns[i] {
			return false
		}
	}
	for i := range left.ScopeFields {
		if left.ScopeFields[i] != right.ScopeFields[i] {
			return false
		}
	}
	for i := range left.Fields {
		if left.Fields[i] != right.Fields[i] {
			return false
		}
	}
	return true
}

func inspectManagedRowMutationStorage(
	ctx context.Context,
	exec Executor,
	tableName string,
	fact schemacatalog.SystemTableRowMutationProtection,
) error {
	if exec == nil {
		return fmt.Errorf("row mutation protection storage inspection requires an executor")
	}
	rows, err := exec.QueryContext(ctx, `SELECT COLUMN_NAME, LOWER(COLUMN_TYPE), IS_NULLABLE, COLUMN_DEFAULT, EXTRA
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
ORDER BY ORDINAL_POSITION`, tableName)
	if err != nil {
		return fmt.Errorf("inspect row mutation protection storage for %s: %w", tableName, err)
	}
	defer rows.Close()

	type physicalColumnShape struct {
		columnType string
		nullable   bool
		defaultSet bool
		extra      string
	}
	expectedColumns := append(
		append(
			[]schemacatalog.SystemTableRowMutationProtectionField(nil),
			fact.ScopeFields...,
		),
		fact.Fields...,
	)
	found := make(map[string]physicalColumnShape, len(expectedColumns))
	totalColumns := 0
	for rows.Next() {
		totalColumns++
		var name, columnType, nullable, extra string
		var defaultValue sql.NullString
		if err := rows.Scan(&name, &columnType, &nullable, &defaultValue, &extra); err != nil {
			return fmt.Errorf("scan row mutation protection storage for %s: %w", tableName, err)
		}
		for _, field := range expectedColumns {
			if strings.EqualFold(name, field.PhysicalName) {
				fieldKey := strings.ToLower(field.PhysicalName)
				if _, duplicate := found[fieldKey]; duplicate {
					return fmt.Errorf(
						"row mutation protection storage for %s has duplicate physical field %s",
						tableName,
						field.PhysicalName,
					)
				}
				found[fieldKey] = physicalColumnShape{
					columnType: columnType,
					nullable:   strings.EqualFold(nullable, "YES"),
					defaultSet: defaultValue.Valid,
					extra:      strings.TrimSpace(extra),
				}
			}
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate row mutation protection storage for %s: %w", tableName, err)
	}

	if totalColumns == 0 {
		return fmt.Errorf("row mutation protection storage table %s is missing", tableName)
	}
	if len(found) == 0 {
		return fmt.Errorf(
			"row mutation protection storage for %s is unavailable: declared scope and authority columns are absent",
			tableName,
		)
	}
	if len(found) != len(expectedColumns) {
		return fmt.Errorf(
			"row mutation protection storage for %s is partially deployed",
			tableName,
		)
	}
	for _, field := range fact.ScopeFields {
		shape := found[strings.ToLower(field.PhysicalName)]
		if shape.nullable ||
			shape.defaultSet ||
			shape.extra != "" ||
			!compatiblePhysicalColumnType(shape.columnType, field.SQLType) {
			return fmt.Errorf(
				"row mutation protection scope for %s.%s has incompatible physical shape",
				tableName,
				field.PhysicalName,
			)
		}
	}
	for _, field := range fact.Fields {
		shape := found[strings.ToLower(field.PhysicalName)]
		if !shape.nullable ||
			shape.defaultSet ||
			shape.extra != "" ||
			!compatiblePhysicalColumnType(shape.columnType, field.SQLType) {
			return fmt.Errorf(
				"row mutation protection storage for %s.%s has incompatible physical shape",
				tableName,
				field.PhysicalName,
			)
		}
	}
	if err := rejectManagedRowMutationAutomaticTableScope(ctx, exec, tableName, fact.ScopeFields); err != nil {
		return err
	}
	if err := verifyExactTableIndexWithQueryer(
		ctx,
		exec,
		tableName,
		fact.RequiredUniqueIndex.Name,
		true,
		fact.RequiredUniqueIndex.Columns,
	); err != nil {
		return fmt.Errorf("row mutation protection storage for %s is incomplete: %w", tableName, err)
	}
	return nil
}

func rejectManagedRowMutationAutomaticTableScope(
	ctx context.Context,
	exec Executor,
	tableName string,
	scopeFields []schemacatalog.SystemTableRowMutationProtectionField,
) error {
	var shardingInfo string
	var primaryField sql.NullString
	err := exec.QueryRowContext(ctx, `SELECT COALESCE(t.TIDB_ROW_ID_SHARDING_INFO, ''),
       (
         SELECT s.COLUMN_NAME
         FROM INFORMATION_SCHEMA.STATISTICS s
         WHERE s.TABLE_SCHEMA = t.TABLE_SCHEMA
           AND s.TABLE_NAME = t.TABLE_NAME
           AND s.INDEX_NAME = 'PRIMARY'
           AND s.SEQ_IN_INDEX = 1
         LIMIT 1
       )
FROM INFORMATION_SCHEMA.TABLES t
WHERE t.TABLE_SCHEMA = DATABASE() AND t.TABLE_NAME = ?
LIMIT 1`, tableName).Scan(&shardingInfo, &primaryField)
	if err != nil {
		return fmt.Errorf("inspect row mutation protection table generation for %s: %w", tableName, err)
	}
	if !strings.HasPrefix(
		strings.ToUpper(strings.TrimSpace(shardingInfo)),
		"PK_AUTO_RANDOM_BITS=",
	) {
		return nil
	}
	if !primaryField.Valid || strings.TrimSpace(primaryField.String) == "" {
		return fmt.Errorf(
			"row mutation protection storage for %s declares AUTO_RANDOM without an inspectable primary field",
			tableName,
		)
	}
	for _, field := range scopeFields {
		if strings.EqualFold(field.PhysicalName, primaryField.String) {
			return fmt.Errorf(
				"row mutation protection scope for %s.%s uses TiDB AUTO_RANDOM",
				tableName,
				field.PhysicalName,
			)
		}
	}
	return nil
}

func (p managedRowMutationProtection) mutablePredicate() string {
	parts := make([]string, 0, len(p.fields))
	for _, field := range p.fields {
		parts = append(parts, fmt.Sprintf("`%s` IS NULL", field.PhysicalName))
	}
	return strings.Join(parts, " AND ")
}

func (p managedRowMutationProtection) protectedPredicate() string {
	parts := make([]string, 0, len(p.fields))
	for _, field := range p.fields {
		parts = append(parts, fmt.Sprintf("`%s` IS NOT NULL", field.PhysicalName))
	}
	return strings.Join(parts, " OR ")
}

func (p managedRowMutationProtection) fieldNames() []string {
	fields := make([]string, 0, len(p.fields))
	for _, field := range p.fields {
		fields = append(fields, field.FieldName)
	}
	sort.Strings(fields)
	return fields
}

func (p managedRowMutationProtection) physicalFieldNames() []string {
	fields := make([]string, 0, len(p.fields))
	for _, field := range p.fields {
		fields = append(fields, field.PhysicalName)
	}
	sort.Strings(fields)
	return fields
}

type managedRowMutationState uint8

const (
	managedRowMutationMutable managedRowMutationState = iota
	managedRowMutationAuthorityOwned
	managedRowMutationCorrupt
)

func (p managedRowMutationProtection) classify(record map[string]any) managedRowMutationState {
	nonNull := 0
	for _, field := range p.fields {
		found := false
		for fieldName, value := range record {
			if strings.EqualFold(fieldName, field.FieldName) ||
				strings.EqualFold(fieldName, field.PhysicalName) {
				found = true
				if value != nil {
					nonNull++
				}
				break
			}
		}
		if !found {
			return managedRowMutationCorrupt
		}
	}
	switch nonNull {
	case 0:
		return managedRowMutationMutable
	case len(p.fields):
		return managedRowMutationAuthorityOwned
	default:
		return managedRowMutationCorrupt
	}
}