package persistence

import (
	"fmt"
	"sort"
	"strings"

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

// RejectAuthorityOnlyFieldWrites fences generic records and repository
// boundaries using the immutable system-table catalog. authority_only is not a
// generic metadata authoring value, so this canonical inventory is the complete
// policy source; cached live metadata is used only to resolve logical and
// physical table names.
func RejectAuthorityOnlyFieldWrites(
	tableName string,
	schema *models.ObjectMetadata,
	records ...models.Record,
) error {
	canonicalTableNames, err := canonicalWriteProtectionTableNames(tableName, schema)
	if err != nil {
		return apperrors.NewInternalError(
			"resolve canonical write-protection table",
			err,
		)
	}
	canonicalAuthorityFields, err := canonicalAuthorityOnlyFieldAliases(canonicalTableNames)
	if err != nil {
		return apperrors.NewInternalError(
			"load canonical system-table write-protection policy",
			err,
		)
	}
	deniedSet := make(map[string]struct{})

	// Inspect only caller-supplied record maps, including explicit nil values.
	// Generated allColumns inventories may include an omitted nullable protected
	// field and must not turn that omission into an authored value.
	for _, record := range records {
		for suppliedField := range record {
			fieldKey := strings.ToLower(strings.TrimSpace(suppliedField))
			if canonicalField, protected := canonicalAuthorityFields[fieldKey]; protected {
				deniedSet[canonicalField] = struct{}{}
			}
		}
	}
	if len(deniedSet) == 0 {
		return nil
	}

	denied := make([]string, 0, len(deniedSet))
	for field := range deniedSet {
		denied = append(denied, field)
	}
	sort.Strings(denied)
	return apperrors.NewInvalidFieldForWriteError(
		denied,
		fmt.Sprintf(
			"fields %s are write-protected by metadata policy %q",
			strings.Join(denied, ", "),
			constants.FieldWriteProtectionAuthorityOnly,
		),
	)
}

func canonicalWriteProtectionTableNames(tableName string, schema *models.ObjectMetadata) ([]string, error) {
	tableNames := []string{tableName}
	if schema == nil || strings.TrimSpace(schema.APIName) == "" {
		return tableNames, nil
	}
	resolvedTableName, err := schema.ResolveReadSourceTable()
	if err != nil {
		return nil, err
	}
	if !strings.EqualFold(strings.TrimSpace(resolvedTableName), strings.TrimSpace(tableName)) {
		tableNames = append(tableNames, resolvedTableName)
	}
	return tableNames, nil
}

func canonicalAuthorityOnlyFieldAliases(tableNames []string) (map[string]string, error) {
	aliases := make(map[string]string)
	for _, tableName := range tableNames {
		fields, err := schemacatalog.SystemTableFieldWriteProtectionsFor(tableName)
		if err != nil {
			return nil, err
		}
		for alias, fact := range fields {
			if strings.TrimSpace(fact.Policy) == constants.FieldWriteProtectionAuthorityOnly {
				aliases[alias] = fact.FieldName
			}
		}
	}
	return aliases, nil
}

// RejectGenericFieldMetadataRemoval keeps physical field lifecycle and its
// _System_Field row under the single schema-persistence owner. Generic record
// deletion cannot safely remove only the metadata row: doing so would orphan
// the physical column, and set-based deletion has no atomic schema counterpart.
// This gate is unbypassable by record security context.
func RejectGenericFieldMetadataRemoval(tableName string, objectSchema *models.ObjectMetadata) error {
	tableNames, err := canonicalWriteProtectionTableNames(tableName, objectSchema)
	if err != nil {
		return apperrors.NewInternalError("resolve canonical field-metadata table", err)
	}
	for _, candidate := range tableNames {
		if strings.EqualFold(strings.TrimSpace(candidate), constants.TableField) {
			return apperrors.NewInvalidFieldForWriteError(
				[]string{"tableName"},
				fmt.Sprintf(
					"%s rows cannot be removed through generic record persistence; field lifecycle is owned by schema persistence",
					constants.TableField,
				),
			)
		}
	}
	return nil
}

// RejectGenericProtectedCascadeParentRemoval keeps parent rows whose physical
// deletion would cascade into catalog-protected children under the explicit
// schema lifecycle owner. Generic record deletion has no authority to inspect
// and lock every protected child before the database applies the cascade.
func RejectGenericProtectedCascadeParentRemoval(
	tableName string,
	objectSchema *models.ObjectMetadata,
) error {
	tableNames, err := canonicalWriteProtectionTableNames(tableName, objectSchema)
	if err != nil {
		return apperrors.NewInternalError("resolve canonical protected-cascade parent", err)
	}
	for _, candidate := range tableNames {
		relations, err := schemacatalog.SystemTableRowMutationDirectProtectedCascadesFrom(candidate)
		if err != nil {
			return apperrors.NewInternalError("load protected-cascade parent policy", err)
		}
		if len(relations) == 0 {
			continue
		}
		children := make([]string, 0, len(relations))
		seen := make(map[string]struct{}, len(relations))
		for _, relation := range relations {
			key := strings.ToLower(relation.ChildTableName)
			if _, duplicate := seen[key]; duplicate {
				continue
			}
			seen[key] = struct{}{}
			children = append(children, relation.ChildTableName)
		}
		sort.Strings(children)
		return apperrors.NewInvalidFieldForWriteError(
			[]string{"tableName"},
			fmt.Sprintf(
				"%s rows cannot be removed through generic record persistence because deletion can cascade into catalog-protected children: %s",
				relations[0].ParentTableName,
				strings.Join(children, ", "),
			),
		)
	}
	return nil
}

func fieldMetadataDeletionAuthored(updates models.Record, objectSchema *models.ObjectMetadata) bool {
	for fieldName := range updates {
		if strings.EqualFold(strings.TrimSpace(fieldName), constants.FieldIsDeleted) {
			return true
		}
		if objectSchema == nil {
			continue
		}
		for i := range objectSchema.Fields {
			field := &objectSchema.Fields[i]
			if !strings.EqualFold(strings.TrimSpace(field.APIName), constants.FieldIsDeleted) {
				continue
			}
			physicalName := strings.TrimSpace(field.PhysicalName)
			if physicalName != "" && strings.EqualFold(strings.TrimSpace(fieldName), physicalName) {
				return true
			}
		}
	}
	return false
}