package schemacatalog

import (
	"fmt"
	"sort"
	"strings"
	"sync"

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

// SystemTableRowMutationProtectedCascade is one direct physical FK cascade
// whose child rows carry catalog-owned row-mutation authority.
type SystemTableRowMutationProtectedCascade struct {
	ParentTableName    string
	ParentFieldName    string
	ParentPhysicalName string
	ChildTableName     string
	ChildFieldName     string
	ChildPhysicalName  string
	Protection         SystemTableRowMutationProtection
}

type systemTableRowMutationProtectedCascadeInventory map[string][]SystemTableRowMutationProtectedCascade

var loadSystemTableRowMutationProtectedCascadeInventory = sync.OnceValues(func() (systemTableRowMutationProtectedCascadeInventory, error) {
	definitions, err := ManagedTableDefinitions()
	if err != nil {
		return systemTableRowMutationProtectedCascadeInventory{}, err
	}
	return systemTableRowMutationProtectedCascadeInventoryFromDefinitions(definitions)
})

// SystemTableRowMutationDirectProtectedCascadesFrom returns direct protected
// children of one canonical parent table. The direct qualifier is deliberate:
// multi-hop cascade locking needs a separately reviewed lock-order contract.
func SystemTableRowMutationDirectProtectedCascadesFrom(
	parentTableName string,
) ([]SystemTableRowMutationProtectedCascade, error) {
	inventory, err := loadSystemTableRowMutationProtectedCascadeInventory()
	if err != nil {
		return nil, err
	}
	relations := inventory[normalizeWriteProtectionLookupKey(parentTableName)]
	return cloneSystemTableRowMutationProtectedCascades(relations), nil
}

func systemTableRowMutationProtectedCascadeInventoryFromDefinitions(
	definitions []schema.TableDefinition,
) (systemTableRowMutationProtectedCascadeInventory, error) {
	protections, err := systemTableRowMutationProtectionInventoryFromDefinitions(definitions)
	if err != nil {
		return nil, err
	}

	definitionsByName := make(map[string]schema.TableDefinition, len(definitions))
	hasCascadeParent := make(map[string]bool, len(definitions))
	for _, definition := range definitions {
		tableName := strings.TrimSpace(definition.TableName)
		tableKey := normalizeWriteProtectionLookupKey(tableName)
		if tableName == "" {
			return nil, fmt.Errorf("row mutation protected cascade has an empty table name")
		}
		if _, duplicate := definitionsByName[tableKey]; duplicate {
			return nil, fmt.Errorf("row mutation protected cascade has duplicate table %s", tableName)
		}
		definitionsByName[tableKey] = definition
		for _, foreignKey := range definition.ForeignKeys {
			if strings.EqualFold(strings.TrimSpace(foreignKey.OnDelete), "CASCADE") {
				hasCascadeParent[tableKey] = true
			}
		}
	}

	inventory := make(systemTableRowMutationProtectedCascadeInventory)
	seen := make(map[string]struct{})
	for _, child := range definitions {
		protection, protected := protections[normalizeWriteProtectionLookupKey(child.TableName)]
		if !protected {
			continue
		}
		for _, foreignKey := range child.ForeignKeys {
			if !strings.EqualFold(strings.TrimSpace(foreignKey.OnDelete), "CASCADE") {
				continue
			}
			parentTableName, parentFieldName, err := parseSystemTableForeignKeyReference(foreignKey.References)
			if err != nil {
				return nil, fmt.Errorf(
					"system table %s protected cascade field %s: %w",
					child.TableName,
					foreignKey.Column,
					err,
				)
			}
			parent, ok := definitionsByName[normalizeWriteProtectionLookupKey(parentTableName)]
			if !ok || parent.TableName != parentTableName {
				return nil, fmt.Errorf(
					"system table %s protected cascade references unknown exact parent table %s",
					child.TableName,
					parentTableName,
				)
			}
			if hasCascadeParent[normalizeWriteProtectionLookupKey(parentTableName)] {
				return nil, fmt.Errorf(
					"system table %s protected cascade through %s is multi-hop and requires an explicit lock contract",
					child.TableName,
					parentTableName,
				)
			}
			childFieldName, childPhysicalName, ok := exactSystemTableColumn(child, foreignKey.Column)
			if !ok {
				return nil, fmt.Errorf(
					"system table %s protected cascade references unknown exact child field %s",
					child.TableName,
					foreignKey.Column,
				)
			}
			parentFieldName, parentPhysicalName, ok := exactSystemTableColumn(parent, parentFieldName)
			if !ok {
				return nil, fmt.Errorf(
					"system table %s protected cascade references unknown exact parent field %s.%s",
					child.TableName,
					parentTableName,
					parentFieldName,
				)
			}

			relation := SystemTableRowMutationProtectedCascade{
				ParentTableName:    parent.TableName,
				ParentFieldName:    parentFieldName,
				ParentPhysicalName: parentPhysicalName,
				ChildTableName:     child.TableName,
				ChildFieldName:     childFieldName,
				ChildPhysicalName:  childPhysicalName,
				Protection:         protection,
			}
			relationKey := strings.ToLower(strings.Join([]string{
				relation.ParentTableName,
				relation.ParentFieldName,
				relation.ChildTableName,
				relation.ChildFieldName,
			}, "\x00"))
			if _, duplicate := seen[relationKey]; duplicate {
				return nil, fmt.Errorf(
					"duplicate row mutation protected cascade %s.%s -> %s.%s",
					relation.ParentTableName,
					relation.ParentFieldName,
					relation.ChildTableName,
					relation.ChildFieldName,
				)
			}
			seen[relationKey] = struct{}{}
			parentKey := normalizeWriteProtectionLookupKey(relation.ParentTableName)
			inventory[parentKey] = append(inventory[parentKey], relation)
		}
	}
	for parentKey := range inventory {
		sort.Slice(inventory[parentKey], func(i, j int) bool {
			left := inventory[parentKey][i]
			right := inventory[parentKey][j]
			if left.ChildTableName != right.ChildTableName {
				return left.ChildTableName < right.ChildTableName
			}
			if left.ChildFieldName != right.ChildFieldName {
				return left.ChildFieldName < right.ChildFieldName
			}
			return left.ParentFieldName < right.ParentFieldName
		})
	}
	return inventory, nil
}

func parseSystemTableForeignKeyReference(reference string) (string, string, error) {
	reference = strings.TrimSpace(reference)
	open := strings.IndexByte(reference, '(')
	if open <= 0 || !strings.HasSuffix(reference, ")") ||
		strings.Contains(reference[open+1:len(reference)-1], "(") {
		return "", "", fmt.Errorf("foreign key reference %q must be Parent(Field)", reference)
	}
	tableName := strings.TrimSpace(reference[:open])
	fieldName := strings.TrimSpace(reference[open+1 : len(reference)-1])
	if tableName == "" || fieldName == "" {
		return "", "", fmt.Errorf("foreign key reference %q must be Parent(Field)", reference)
	}
	return tableName, fieldName, nil
}

func exactSystemTableColumn(
	definition schema.TableDefinition,
	fieldName string,
) (string, string, bool) {
	fieldName = strings.TrimSpace(fieldName)
	for _, column := range definition.Columns {
		if column.Name != fieldName {
			continue
		}
		physicalName := strings.TrimSpace(column.PhysicalName)
		if physicalName == "" {
			physicalName = column.Name
		}
		return column.Name, physicalName, true
	}
	return "", "", false
}

func cloneSystemTableRowMutationProtectedCascades(
	relations []SystemTableRowMutationProtectedCascade,
) []SystemTableRowMutationProtectedCascade {
	cloned := make([]SystemTableRowMutationProtectedCascade, len(relations))
	for i, relation := range relations {
		cloned[i] = relation
		cloned[i].Protection = cloneSystemTableRowMutationProtection(relation.Protection)
	}
	return cloned
}