package persistence

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

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

type managedRowCascadeDeleteTarget struct {
	tableName  string
	parentKeys []string
}

const maxManagedRowCascadeParentKeys = 1000

// preflightManagedRowMutationCascades locks one parent deletion set and every
// directly cascading protected child before the caller mutates either table.
// Parent-first locking prevents a concurrent FK child insert; locking every
// existing child prevents a concurrent authority-marker assignment.
func preflightManagedRowMutationCascades(
	ctx context.Context,
	transactions *TransactionManager,
	target managedRowCascadeDeleteTarget,
) error {
	if transactions == nil || !transactions.ownsPessimisticTx(ctx) {
		return apperrors.NewInternalError(
			"managed row cascade preflight requires the exact TransactionManager-owned pessimistic transaction",
			nil,
		)
	}
	tx, ok := transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return apperrors.NewInternalError(
			"managed row cascade preflight requires the active pessimistic SQL transaction",
			nil,
		)
	}
	if !isManagedRowCascadeIdentifier(target.tableName) {
		return apperrors.NewInternalError(
			fmt.Sprintf("invalid managed row cascade target table %q", target.tableName),
			nil,
		)
	}
	if len(target.parentKeys) == 0 ||
		len(target.parentKeys) > maxManagedRowCascadeParentKeys {
		return apperrors.NewInternalError(
			fmt.Sprintf(
				"managed row cascade preflight requires 1..%d exact parent keys",
				maxManagedRowCascadeParentKeys,
			),
			nil,
		)
	}
	for _, parentKey := range target.parentKeys {
		if strings.TrimSpace(parentKey) == "" {
			return apperrors.NewInternalError(
				"managed row cascade preflight requires non-empty exact parent keys",
				nil,
			)
		}
	}
	relations, err := schemacatalog.SystemTableRowMutationDirectProtectedCascadesFrom(
		target.tableName,
	)
	if err != nil {
		return apperrors.NewInternalError(
			"load managed row cascade declarations",
			err,
		)
	}
	if len(relations) == 0 {
		return nil
	}

	for _, relation := range relations {
		if !strings.EqualFold(relation.ParentTableName, target.tableName) {
			return apperrors.NewInternalError(fmt.Sprintf(
				"managed row cascade parent %s does not match target %s",
				relation.ParentTableName,
				target.tableName,
			), nil)
		}
		for _, identifier := range []string{
			relation.ParentTableName,
			relation.ParentPhysicalName,
			relation.ChildTableName,
			relation.ChildPhysicalName,
		} {
			if !isManagedRowCascadeIdentifier(identifier) {
				return apperrors.NewInternalError(
					fmt.Sprintf("invalid managed row cascade identifier %q", identifier),
					nil,
				)
			}
		}
		parentValues, err := lockManagedRowCascadeParents(
			ctx,
			tx,
			relation,
			target,
		)
		if err != nil {
			return err
		}
		if len(parentValues) == 0 {
			continue
		}
		if err := requireManagedRowCascadeForeignKey(
			ctx,
			tx,
			relation,
		); err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf(
					"validate managed row cascade foreign key for %s",
					relation.ChildTableName,
				),
				err,
			)
		}
		policy, err := managedRowMutationProtectionFromCatalog(
			ctx,
			tx,
			relation.Protection,
		)
		if err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf(
					"validate managed row cascade storage for %s",
					relation.ChildTableName,
				),
				err,
			)
		}
		if policy.storageTableName != relation.ChildTableName {
			return apperrors.NewInternalError(
				fmt.Sprintf(
					"managed row cascade child %s resolves to storage table %s",
					relation.ChildTableName,
					policy.storageTableName,
				),
				nil,
			)
		}
		children, err := lockManagedRowCascadeChildren(
			ctx,
			tx,
			relation,
			parentValues,
			policy,
		)
		if err != nil {
			return err
		}
		for _, child := range children {
			switch policy.classify(child) {
			case managedRowMutationMutable:
				continue
			case managedRowMutationAuthorityOwned:
				return apperrors.NewInvalidFieldForWriteError(
					[]string{policy.fieldNames()[0]},
					fmt.Sprintf(
						"cannot delete %s metadata because a cascading %s row is catalog-owned",
						relation.ParentTableName,
						relation.ChildTableName,
					),
				)
			default:
				return apperrors.NewInternalError(
					fmt.Sprintf(
						"managed row cascade child %s has a partial or unreadable authority assignment",
						relation.ChildTableName,
					),
					nil,
				)
			}
		}
	}
	return nil
}

func lockManagedRowCascadeParents(
	ctx context.Context,
	exec Executor,
	relation schemacatalog.SystemTableRowMutationProtectedCascade,
	target managedRowCascadeDeleteTarget,
) ([]interface{}, error) {
	parentField := quoteManagedRowCascadeIdentifier(relation.ParentPhysicalName)
	placeholders := make([]string, len(target.parentKeys))
	args := make([]interface{}, len(target.parentKeys))
	for i, parentKey := range target.parentKeys {
		placeholders[i] = "?"
		args[i] = parentKey
	}
	sqlText := fmt.Sprintf(
		"SELECT %s FROM %s WHERE %s IN (%s) ORDER BY %s FOR UPDATE",
		parentField,
		quoteManagedRowCascadeIdentifier(relation.ParentTableName),
		parentField,
		strings.Join(placeholders, ", "),
		parentField,
	)
	rows, err := exec.QueryContext(ctx, sqlText, args...)
	if err != nil {
		return nil, apperrors.NewInternalError(fmt.Sprintf(
			"lock managed row cascade parent %s",
			relation.ParentTableName,
		), err)
	}
	defer rows.Close()

	result := make([]interface{}, 0)
	for rows.Next() {
		var value interface{}
		if err := rows.Scan(&value); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf(
				"scan managed row cascade parent %s",
				relation.ParentTableName,
			), err)
		}
		if bytes, ok := value.([]byte); ok {
			value = append([]byte(nil), bytes...)
		}
		if value != nil {
			result = append(result, value)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError(fmt.Sprintf(
			"iterate managed row cascade parent %s",
			relation.ParentTableName,
		), err)
	}
	return result, nil
}

func requireManagedRowCascadeForeignKey(
	ctx context.Context,
	exec Executor,
	relation schemacatalog.SystemTableRowMutationProtectedCascade,
) error {
	var count int
	err := exec.QueryRowContext(ctx, `SELECT COUNT(*)
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE k
JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc
  ON rc.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
 AND rc.CONSTRAINT_NAME = k.CONSTRAINT_NAME
 AND rc.TABLE_NAME = k.TABLE_NAME
WHERE k.TABLE_SCHEMA = DATABASE()
  AND k.TABLE_NAME = ?
  AND k.COLUMN_NAME = ?
  AND k.REFERENCED_TABLE_SCHEMA = DATABASE()
  AND k.REFERENCED_TABLE_NAME = ?
  AND k.REFERENCED_COLUMN_NAME = ?
  AND UPPER(COALESCE(rc.DELETE_RULE, '')) = 'CASCADE'
  AND NOT EXISTS (
      SELECT 1
      FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE extra
      WHERE extra.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
        AND extra.CONSTRAINT_NAME = k.CONSTRAINT_NAME
        AND extra.TABLE_SCHEMA = k.TABLE_SCHEMA
        AND extra.TABLE_NAME = k.TABLE_NAME
        AND extra.ORDINAL_POSITION <> k.ORDINAL_POSITION
  )`,
		relation.ChildTableName,
		relation.ChildPhysicalName,
		relation.ParentTableName,
		relation.ParentPhysicalName,
	).Scan(&count)
	if err != nil {
		return fmt.Errorf(
			"inspect exact %s.%s -> %s.%s cascade: %w",
			relation.ChildTableName,
			relation.ChildPhysicalName,
			relation.ParentTableName,
			relation.ParentPhysicalName,
			err,
		)
	}
	if count != 1 {
		return fmt.Errorf(
			"expected one live single-column %s.%s -> %s.%s ON DELETE CASCADE constraint, found %d",
			relation.ChildTableName,
			relation.ChildPhysicalName,
			relation.ParentTableName,
			relation.ParentPhysicalName,
			count,
		)
	}
	return nil
}

func lockManagedRowCascadeChildren(
	ctx context.Context,
	exec Executor,
	relation schemacatalog.SystemTableRowMutationProtectedCascade,
	parentValues []interface{},
	policy managedRowMutationProtection,
) ([]map[string]any, error) {
	fields := append([]string{constants.FieldID}, policy.physicalFieldNames()...)
	quotedFields := make([]string, len(fields))
	for i, field := range fields {
		if !isManagedRowCascadeIdentifier(field) {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("invalid managed row cascade child field %q", field),
				nil,
			)
		}
		quotedFields[i] = quoteManagedRowCascadeIdentifier(field)
	}
	placeholders := make([]string, len(parentValues))
	for i := range placeholders {
		placeholders[i] = "?"
	}
	sqlText := fmt.Sprintf(
		"SELECT %s FROM %s WHERE %s IN (%s) ORDER BY %s FOR UPDATE",
		strings.Join(quotedFields, ", "),
		quoteManagedRowCascadeIdentifier(relation.ChildTableName),
		quoteManagedRowCascadeIdentifier(relation.ChildPhysicalName),
		strings.Join(placeholders, ", "),
		quoteManagedRowCascadeIdentifier(constants.FieldID),
	)
	rows, err := exec.QueryContext(ctx, sqlText, parentValues...)
	if err != nil {
		return nil, apperrors.NewInternalError(
			fmt.Sprintf("lock managed row cascade child %s", relation.ChildTableName),
			err,
		)
	}
	defer rows.Close()
	children, err := query.ScanRowsToMaps(rows)
	if err != nil {
		return nil, apperrors.NewInternalError(
			fmt.Sprintf("scan managed row cascade child %s", relation.ChildTableName),
			err,
		)
	}
	return children, nil
}

func isManagedRowCascadeIdentifier(identifier string) bool {
	return isValidFieldName(identifier) && !strings.Contains(identifier, ".")
}

func quoteManagedRowCascadeIdentifier(identifier string) string {
	return "`" + identifier + "`"
}