package records

import (
	"context"
	stderrors "errors"
	"fmt"
	"net/http"
	"sort"
	"strings"

	"github.com/nexuscrm/kernel/pkg/apierr"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	infraPersistence "github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/metadatarel"
	"github.com/nexuscrm/shared/pkg/models"
)

type recycleBinPurgeCandidate struct {
	tombstone recycleBinMutationTombstone
	schema    *models.ObjectMetadata
	record    models.SObject
}

// PurgeRecord permanently removes the latest recycle-bin tombstone and its
// already-deleted physical row. Purge is the physical tail of a completed
// logical delete: it does not replay delete lifecycle hooks, audit, rollups, or
// cache invalidation. The only cascade it performs is the leaf-up physical-FK
// subtree required before the parent row can be removed.
func (ps *PersistenceService) PurgeRecord(ctx context.Context, recordID string, currentUser *models.UserSession) error {
	recordID = strings.TrimSpace(recordID)
	if recordID == "" {
		return apierr.Invalid("record id is required")
	}
	if currentUser == nil {
		return apierr.Invalid("user session required")
	}
	if ps == nil || ps.repo == nil || ps.metadata == nil || ps.permissions == nil {
		return apierr.Internal(fmt.Errorf("recycle-bin purge: persistence service is not fully wired"))
	}

	tombstone, err := ps.loadRecycleBinTombstone(ctx, recordID)
	if err != nil {
		return err
	}
	schema := ps.metadata.GetSchema(ctx, tombstone.objectName)
	if !isRecycleBinManagedObject(schema) {
		return recycleBinNotFound()
	}
	if !ps.currentUserCanSeeRecycleBinTombstone(ctx, tombstone, currentUser) {
		return recycleBinNotFound()
	}
	if err := ps.checkOperationPermission(ctx, schema, schema.APIName, constants.PermDelete, currentUser); err != nil {
		return err
	}
	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		if err := rejectToolingOnlyTableWrites(tombstone.objectName); err != nil {
			return err
		}
	}

	purgeTombstones, err := ps.loadRecycleBinCascadePurgeTombstones(ctx, tombstone)
	if err != nil {
		return err
	}
	transactionObjectNames := recycleBinTombstoneObjectNames(purgeTombstones)

	return ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, transactionObjectNames, func(txCtx context.Context) error {
		rootTombstone, err := ps.loadRecycleBinTombstoneForMutation(txCtx, recordID, tombstone.objectName, currentUser, "purge")
		if err != nil {
			return err
		}
		purgeTombstones, err := ps.loadRecycleBinCascadePurgeTombstones(txCtx, rootTombstone)
		if err != nil {
			return err
		}
		lockedTombstones, err := ps.lockRecycleBinTombstonesForMutation(txCtx, purgeTombstones, currentUser, "purge")
		if err != nil {
			return err
		}
		purgeTombstones, err = recycleBinLockedTombstonesInOrder(purgeTombstones, lockedTombstones, "purge")
		if err != nil {
			return err
		}
		return ps.purgeRecycleBinTombstoneTreeInTransaction(txCtx, purgeTombstones, currentUser)
	})
}

func (ps *PersistenceService) loadRecycleBinCascadePurgeTombstones(ctx context.Context, root recycleBinMutationTombstone) ([]recycleBinMutationTombstone, error) {
	if strings.TrimSpace(root.deleteOperationID) == "" {
		return ps.loadLegacyRecycleBinCascadeTombstones(ctx, root, "purge")
	}
	repo, ok := ps.repo.(recycleBinOperationRepository)
	if !ok {
		return nil, appErrors.NewInternalError("recycle-bin cascade purge requires operation lineage repository support", nil)
	}
	rows, err := repo.FindRecycleBinEntriesByDeleteOperation(ctx, root.deleteOperationID)
	if err != nil {
		return nil, appErrors.NewInternalError("failed to load recycle-bin delete operation tombstones for purge", err)
	}
	operationTombstones := make([]recycleBinMutationTombstone, 0, len(rows))
	requestedSeen := false
	var operationRoot recycleBinMutationTombstone
	operationRootSeen := false
	for _, row := range rows {
		tombstone, err := recycleBinMutationTombstoneFromRow(row)
		if err != nil {
			return nil, err
		}
		if !strings.EqualFold(tombstone.deleteOperationID, root.deleteOperationID) ||
			!strings.EqualFold(tombstone.deleteRootObjectName, root.deleteRootObjectName) ||
			!strings.EqualFold(tombstone.deleteRootRecordID, root.deleteRootRecordID) {
			continue
		}
		if strings.EqualFold(tombstone.recordID, root.recordID) && strings.EqualFold(tombstone.objectName, root.objectName) {
			requestedSeen = true
		}
		if tombstone.isDeleteOperationRoot() {
			operationRoot = tombstone
			operationRootSeen = true
		}
		operationTombstones = append(operationTombstones, tombstone)
	}
	if !requestedSeen {
		operationTombstones = append(operationTombstones, root)
	}
	sortRecycleBinRestoreTombstones(operationTombstones)
	if root.isDeleteOperationRoot() {
		if err := validateRecycleBinCascadeLineageTree(root, operationTombstones, "purge"); err != nil {
			return nil, err
		}
		return operationTombstones, nil
	}
	if operationRootSeen {
		if err := validateRecycleBinCascadeLineageTree(operationRoot, operationTombstones, "purge"); err != nil {
			return nil, err
		}
	}
	subtree, err := ps.recycleBinCascadePurgeSubtree(ctx, root, operationTombstones)
	if err != nil {
		return nil, err
	}
	return subtree, nil
}

func (ps *PersistenceService) recycleBinCascadePurgeSubtree(ctx context.Context, root recycleBinMutationTombstone, operationTombstones []recycleBinMutationTombstone) ([]recycleBinMutationTombstone, error) {
	return recycleBinCascadeMutationSubtree(root, operationTombstones, "purge")
}

func recycleBinCascadeMutationSubtree(root recycleBinMutationTombstone, operationTombstones []recycleBinMutationTombstone, surface string) ([]recycleBinMutationTombstone, error) {
	rootKey := recycleBinTombstoneTreeKey(root.objectName, root.recordID)
	if rootKey == "\x00" {
		return nil, apierr.Conflict(fmt.Sprintf("cannot %s recycle-bin cascade tree because the requested lineage node is missing object or record identity", surface))
	}
	childrenByParent := map[string][]recycleBinMutationTombstone{}
	tombstonesByKey := map[string]recycleBinMutationTombstone{}
	duplicateKeys := map[string]struct{}{}
	for _, tombstone := range operationTombstones {
		key := recycleBinTombstoneTreeKey(tombstone.objectName, tombstone.recordID)
		if key == "\x00" {
			continue
		}
		if _, ok := tombstonesByKey[key]; ok {
			duplicateKeys[key] = struct{}{}
			continue
		}
		tombstonesByKey[key] = tombstone
		if tombstone.isCascadeChild() {
			parentKey := recycleBinTombstoneTreeKey(tombstone.cascadeParentObjectName, tombstone.cascadeParentRecordID)
			if parentKey == "\x00" {
				continue
			}
			childrenByParent[parentKey] = append(childrenByParent[parentKey], tombstone)
		}
	}
	if _, ok := tombstonesByKey[rootKey]; !ok {
		tombstonesByKey[rootKey] = root
	}
	var subtree []recycleBinMutationTombstone
	visiting := map[string]bool{}
	visited := map[string]bool{}
	var visit func(recycleBinMutationTombstone) error
	visit = func(tombstone recycleBinMutationTombstone) error {
		key := recycleBinTombstoneTreeKey(tombstone.objectName, tombstone.recordID)
		if _, ok := duplicateKeys[key]; ok {
			return apierr.Conflict(fmt.Sprintf("cannot %s recycle-bin cascade tree because delete lineage contains duplicate nodes", surface))
		}
		if visiting[key] {
			return apierr.Conflict(fmt.Sprintf("cannot %s recycle-bin cascade tree because delete lineage contains a cycle", surface))
		}
		if visited[key] {
			return nil
		}
		visiting[key] = true
		visited[key] = true
		subtree = append(subtree, tombstone)
		for _, child := range childrenByParent[key] {
			if child.cascadeDepth <= tombstone.cascadeDepth {
				return apierr.Conflict(fmt.Sprintf("cannot %s recycle-bin cascade tree because descendant depth does not match its recorded parent", surface))
			}
			if err := visit(child); err != nil {
				return err
			}
		}
		visiting[key] = false
		return nil
	}
	start := tombstonesByKey[rootKey]
	if err := visit(start); err != nil {
		return nil, err
	}
	sortRecycleBinRestoreTombstones(subtree)
	return subtree, nil
}

func (ps *PersistenceService) purgeRecycleBinTombstoneTreeInTransaction(txCtx context.Context, tombstones []recycleBinMutationTombstone, currentUser *models.UserSession) error {
	if _, err := ps.prelockRecycleBinMutationParentRows(txCtx, tombstones); err != nil {
		return err
	}
	candidates, err := ps.lockAndValidateRecycleBinPurgeCandidates(txCtx, tombstones, currentUser)
	if err != nil {
		return err
	}
	if err := ps.validateRecycleBinPurgeCompleteness(txCtx, tombstones); err != nil {
		return err
	}
	sort.SliceStable(candidates, func(i, j int) bool {
		if candidates[i].tombstone.cascadeDepth != candidates[j].tombstone.cascadeDepth {
			return candidates[i].tombstone.cascadeDepth > candidates[j].tombstone.cascadeDepth
		}
		if !strings.EqualFold(candidates[i].tombstone.objectName, candidates[j].tombstone.objectName) {
			return strings.ToLower(candidates[i].tombstone.objectName) > strings.ToLower(candidates[j].tombstone.objectName)
		}
		return strings.ToLower(candidates[i].tombstone.recordID) > strings.ToLower(candidates[j].tombstone.recordID)
	})
	for _, candidate := range candidates {
		if err := ps.repo.PhysicalDelete(txCtx, candidate.schema.APIName, candidate.tombstone.recordID); err != nil {
			return appErrors.NewInternalError(fmt.Sprintf("failed to purge record %s/%s", candidate.schema.APIName, candidate.tombstone.recordID), err)
		}
		if err := ps.requestPermissionStateAdvanceForMutation(txCtx, candidate.schema.APIName, candidate.record, nil); err != nil {
			return appErrors.NewInternalError("permission state advance request failed", err)
		}
		InvalidateRecordReadCache(txCtx, candidate.schema.APIName, candidate.tombstone.recordID)
		if err := ps.repo.PhysicalDelete(txCtx, constants.TableRecycleBin, candidate.tombstone.id); err != nil {
			return apierr.Internal(fmt.Errorf("failed to remove recycle-bin entry after purge: %w", err))
		}
	}
	return nil
}

func (ps *PersistenceService) validateRecycleBinPurgeCompleteness(txCtx context.Context, tombstones []recycleBinMutationTombstone) error {
	candidateKeys := make(map[string]struct{}, len(tombstones))
	for _, tombstone := range tombstones {
		candidateKeys[recycleBinTombstoneTreeKey(tombstone.objectName, tombstone.recordID)] = struct{}{}
	}
	edgesByParentObject := ps.recycleBinPurgeCompletenessEdges(txCtx, tombstones)
	for _, tombstone := range tombstones {
		for _, edge := range edgesByParentObject[strings.ToLower(strings.TrimSpace(tombstone.objectName))] {
			deletedChildren, err := ps.repo.GetDeletedChildren(txCtx, edge.childObjectName, edge.fieldAPIName, tombstone.recordID)
			if err != nil {
				if edge.historicalOnly && infraPersistence.IsSchemaDrift(err) {
					continue
				}
				return appErrors.NewInternalError(
					fmt.Sprintf("failed to recheck deleted children of %s via %s before recycle-bin purge", edge.childObjectName, edge.fieldAPIName),
					err,
				)
			}
			for _, child := range deletedChildren {
				childID := strings.TrimSpace(stringValue(child[constants.FieldID]))
				if childID == "" {
					continue
				}
				if _, ok := candidateKeys[recycleBinTombstoneTreeKey(edge.childObjectName, childID)]; ok {
					continue
				}
				return apierr.Conflict("cannot purge recycle-bin record because deleted cascade children remain outside the selected delete lineage")
			}
			activeChildren, err := ps.repo.GetChildren(txCtx, edge.childObjectName, edge.fieldAPIName, tombstone.recordID)
			if err != nil {
				if edge.historicalOnly && infraPersistence.IsSchemaDrift(err) {
					continue
				}
				return appErrors.NewInternalError(
					fmt.Sprintf("failed to recheck active children of %s via %s before recycle-bin purge", edge.childObjectName, edge.fieldAPIName),
					err,
				)
			}
			if len(activeChildren) > 0 {
				return apierr.Conflict("cannot purge recycle-bin record because active cascade children remain")
			}
		}
	}
	return nil
}

func (ps *PersistenceService) recycleBinPurgeCompletenessEdges(ctx context.Context, tombstones []recycleBinMutationTombstone) map[string][]recycleBinPurgeCascadeEdge {
	edgesByParentObject := map[string][]recycleBinPurgeCascadeEdge{}
	edgesByKey := map[string]recycleBinPurgeCascadeEdge{}
	addEdge := func(edge recycleBinPurgeCascadeEdge) {
		edge.parentObjectName = strings.TrimSpace(edge.parentObjectName)
		edge.childObjectName = strings.TrimSpace(edge.childObjectName)
		edge.fieldAPIName = strings.TrimSpace(edge.fieldAPIName)
		if edge.parentObjectName == "" || edge.childObjectName == "" || edge.fieldAPIName == "" {
			return
		}
		parentKey := strings.ToLower(edge.parentObjectName)
		key := parentKey + "|" + strings.ToLower(edge.childObjectName) + "|" + strings.ToLower(edge.fieldAPIName)
		if existing, ok := edgesByKey[key]; ok {
			if !edge.historicalOnly {
				existing.historicalOnly = false
				edgesByKey[key] = existing
			}
			return
		}
		edgesByKey[key] = edge
	}
	for _, tombstone := range tombstones {
		for _, edge := range ps.physicalCascadeChildEdges(ctx, tombstone.objectName) {
			addEdge(edge)
		}
	}
	// Purge can trust recorded cascade lineage after metadata drift, so the
	// same historical edge must also be used to recheck non-lineage children.
	for _, tombstone := range tombstones {
		if !tombstone.isCascadeChild() {
			continue
		}
		addEdge(recycleBinPurgeCascadeEdge{
			parentObjectName: tombstone.cascadeParentObjectName,
			childObjectName:  tombstone.objectName,
			fieldAPIName:     tombstone.cascadeFieldAPIName,
			historicalOnly:   true,
		})
	}
	for _, edge := range edgesByKey {
		parentKey := strings.ToLower(edge.parentObjectName)
		edgesByParentObject[parentKey] = append(edgesByParentObject[parentKey], edge)
	}
	for parentKey := range edgesByParentObject {
		sort.SliceStable(edgesByParentObject[parentKey], func(i, j int) bool {
			if !strings.EqualFold(edgesByParentObject[parentKey][i].childObjectName, edgesByParentObject[parentKey][j].childObjectName) {
				return strings.ToLower(edgesByParentObject[parentKey][i].childObjectName) < strings.ToLower(edgesByParentObject[parentKey][j].childObjectName)
			}
			return strings.ToLower(edgesByParentObject[parentKey][i].fieldAPIName) < strings.ToLower(edgesByParentObject[parentKey][j].fieldAPIName)
		})
	}
	return edgesByParentObject
}

func (ps *PersistenceService) lockAndValidateRecycleBinPurgeCandidates(txCtx context.Context, tombstones []recycleBinMutationTombstone, currentUser *models.UserSession) ([]recycleBinPurgeCandidate, error) {
	candidates := make([]recycleBinPurgeCandidate, 0, len(tombstones))
	for _, tombstone := range tombstones {
		schema := ps.metadata.GetSchema(txCtx, tombstone.objectName)
		if !isRecycleBinManagedObject(schema) {
			return nil, recycleBinNotFound()
		}
		if err := ps.checkOperationPermission(txCtx, schema, schema.APIName, constants.PermDelete, currentUser); err != nil {
			return nil, err
		}
		if !constants.GetSecurityContext(txCtx).BypassesCRUDandFLS() {
			if err := rejectToolingOnlyTableWrites(schema.APIName); err != nil {
				return nil, err
			}
		}

		lockedRecord, err := ps.loadAndValidateAccessForOperation(txCtx, schema.APIName, tombstone.recordID, schema, constants.PermDelete, currentUser, func() error {
			return apierr.Conflict("recycle bin record changed during purge")
		}, func() error {
			return recycleBinPurgePermissionDenied(schema.APIName, tombstone.recordID)
		})
		if err != nil {
			return nil, err
		}
		if !isGuardFieldTruthy(lockedRecord[constants.FieldIsDeleted]) {
			return nil, apierr.Conflict("record is no longer deleted")
		}
		if err := validateRecycleBinPurgeCascadeRecordStillMatchesLineage(tombstone, lockedRecord, schema); err != nil {
			return nil, err
		}
		if err := ps.validateRecycleBinPurgeMutation(txCtx, schema.APIName, lockedRecord, currentUser); err != nil {
			return nil, err
		}
		candidates = append(candidates, recycleBinPurgeCandidate{
			tombstone: tombstone,
			schema:    schema,
			record:    lockedRecord,
		})
	}
	return candidates, nil
}

func validateRecycleBinPurgeCascadeRecordStillMatchesLineage(tombstone recycleBinMutationTombstone, record models.SObject, schema *models.ObjectMetadata) error {
	if !tombstone.isCascadeChild() {
		return nil
	}
	cascadeFieldAPIName := strings.TrimSpace(tombstone.cascadeFieldAPIName)
	if schema != nil && cascadeFieldAPIName != "" && models.FindField(schema, cascadeFieldAPIName) == nil {
		return nil
	}
	return validateRecycleBinCascadeRecordStillMatchesLineage(tombstone, record, "purge")
}

type recycleBinPurgeCascadeEdge struct {
	parentObjectName string
	childObjectName  string
	fieldAPIName     string
	historicalOnly   bool
}

func (ps *PersistenceService) physicalCascadeChildEdges(ctx context.Context, parentObjName string) []recycleBinPurgeCascadeEdge {
	if ps == nil || ps.metadata == nil {
		return nil
	}
	edges := make([]recycleBinPurgeCascadeEdge, 0)
	parentKey := strings.ToLower(strings.TrimSpace(parentObjName))
	for _, schema := range ps.metadata.GetChildRelationships(ctx, parentObjName) {
		if schema == nil {
			continue
		}
		relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, schema.APIName))
		for _, field := range schema.Fields {
			if !metadatarel.FieldReferencesParent(field, parentObjName) {
				continue
			}
			deleteSemantics := metadatarel.ResolveDeleteSemantics(field, relationshipsByField[parentKey+"|"+strings.ToLower(field.APIName)])
			if !deleteSemantics.Cascade {
				continue
			}
			edges = append(edges, recycleBinPurgeCascadeEdge{
				parentObjectName: parentObjName,
				childObjectName:  schema.APIName,
				fieldAPIName:     field.APIName,
			})
		}
	}
	sort.SliceStable(edges, func(i, j int) bool {
		if !strings.EqualFold(edges[i].childObjectName, edges[j].childObjectName) {
			return strings.ToLower(edges[i].childObjectName) < strings.ToLower(edges[j].childObjectName)
		}
		return strings.ToLower(edges[i].fieldAPIName) < strings.ToLower(edges[j].fieldAPIName)
	})
	return edges
}

func (ps *PersistenceService) validateRecycleBinPurgeMutation(ctx context.Context, objectName string, record models.SObject, currentUser *models.UserSession) error {
	if err := rejectManagedEmailTemplateGenericMutation(objectName, record, "purged"); err != nil {
		return err
	}
	if constants.GetSecurityContext(ctx) < constants.SystemContext &&
		strings.EqualFold(objectName, constants.TableRecordShare) {
		if err := ps.checkRecordShareParentNotLocked(ctx, record, currentUser); err != nil {
			return err
		}
	}
	if constants.GetSecurityContext(ctx) < constants.SystemContext {
		if spec, ok := lookupProtectedSpec(objectName); ok && spec.GuardField != "" && isGuardFieldTruthy(record[spec.GuardField]) {
			return apierr.Forbidden(fmt.Sprintf("%s: %s", spec.ErrorCode, spec.Message))
		}
	}
	recordID := strings.TrimSpace(stringValue(record[constants.FieldID]))
	if err := ps.runRecordWriteValidators(ctx, recordWriteValidationBeforePurge, recordWriteValidationEvent{
		ObjectName:  objectName,
		RecordID:    recordID,
		Record:      record,
		OldRecord:   record,
		CurrentUser: currentUser,
	}); err != nil {
		return err
	}
	return nil
}

func recycleBinPurgePermissionDenied(objectName, recordID string) *apierr.Error {
	return apierr.Forbidden(fmt.Sprintf("permission denied: cannot delete %s/%s", objectName, recordID))
}

func isRecycleBinNotFoundError(err error) bool {
	var apiErr *apierr.Error
	return stderrors.As(err, &apiErr) && apiErr.Status == http.StatusNotFound
}