package analytics

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
)

var rollupLockSkippedOnOptimistic = promauto.NewCounter(prometheus.CounterOpts{
	Name: "nexuscrm_rollup_lock_skipped_on_optimistic_total",
	Help: "Count of rollup parent-lock attempts rejected because the enclosing transaction was not pessimistic.",
})

var errRollupLockOnOptimisticTx = apperrors.NewInternalError(
	"rollup lockParent: enclosing transaction is not pessimistic; this indicates a rollup-reachable caller that was not migrated",
	nil,
)

// rollupDepthCtxKey is the unexported context key used to track re-entrancy
// depth of ProcessRollups within a single request/transaction.  A distinct
// unexported type prevents collisions with keys from other packages.
type rollupDepthCtxKey struct{}

// rollupChainCtxKey carries the human-readable chain of "(childObj→parentObj)"
// hops accumulated so far, used to build actionable error messages.
type rollupChainCtxKey struct{}

// rollupVisitedCtxKey carries the set of (childObj, parentObj, parentID,
// rollupField) tuples already processed in this request.  It catches short
// cycles (self-loop, 2-cycle) before the depth cap fires, preventing repeated
// UpdateDerived → ProcessRollups churn on the same tuple.
type rollupVisitedCtxKey struct{}

// rollupVisitedKey is the four-tuple that uniquely identifies one rollup hop
// within a request.  Two hops with the same tuple indicate a cycle.
type rollupVisitedKey struct {
	childObj    string
	parentObj   string
	parentID    string
	rollupField string
}

// rollupVisitedFromCtx returns a copy-on-write snapshot of the visited set.
// Returns an empty set if no set is stored yet; callers must not mutate the
// returned map directly — use withRollupVisited to derive a new context.
func rollupVisitedFromCtx(ctx context.Context) map[rollupVisitedKey]struct{} {
	if m, ok := ctx.Value(rollupVisitedCtxKey{}).(map[rollupVisitedKey]struct{}); ok {
		return m
	}
	return map[rollupVisitedKey]struct{}{}
}

// withRollupVisited returns a new context containing a copy of the current
// visited set extended with key k.
func withRollupVisited(ctx context.Context, k rollupVisitedKey) context.Context {
	prev := rollupVisitedFromCtx(ctx)
	next := make(map[rollupVisitedKey]struct{}, len(prev)+1)
	for key := range prev {
		next[key] = struct{}{}
	}
	next[k] = struct{}{}
	return context.WithValue(ctx, rollupVisitedCtxKey{}, next)
}

// maxRollupDepth is the NexusCRM policy cap.  Salesforce itself imposes a
// similar re-entrancy guard; 8 is a defensive choice that allows realistic
// multi-level hierarchies while rejecting runaway cycles.
const maxRollupDepth = 8
const rollupRecalculatePageSize = 100000

type rollupMaintenanceTransactionRunner interface {
	RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error
}

func validateRollupFilter(config *models.RollupConfig) error {
	if config == nil {
		return apperrors.NewInternalError("rollup field missing config", nil)
	}

	if config.Filter != nil && *config.Filter != "" {
		filter := *config.Filter
		if !isValidSQLFilter(filter) {
			return apperrors.NewValidationError("Filter", "invalid filter expression: contains potentially unsafe characters")
		}
	}

	return nil
}

func (rs *RollupService) recalculateParent(ctx context.Context, parentObjName, parentID string, rollupFields []models.FieldMetadata) error {
	if rs.derived == nil {
		return apperrors.NewInternalError("rollup derived writer not wired", nil)
	}

	systemUser, err := rs.derived.LoadSystemUserSession(ctx)
	if err != nil {
		return apperrors.NewInternalError("failed to load rollup system user", err)
	}
	if systemUser == nil {
		return apperrors.NewInternalError("rollup system user not found", nil)
	}

	recalcUpdates := make(models.SObject, len(rollupFields))
	for _, rollupField := range rollupFields {
		rollup := AffectedRollup{
			ParentObjName: parentObjName,
			ParentID:      parentID,
			RollupField:   rollupField,
		}

		newVal, err := rs.CalculateRollup(ctx, rollup)
		if err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to calculate rollup %s for parent %s", rollupField.APIName, parentID), err)
		}
		recalcUpdates[rollupField.APIName] = newVal
	}

	if err := rs.derived.UpdateDerived(ctx, parentObjName, parentID, recalcUpdates, systemUser); err != nil {
		if retryablePrelockError(err) {
			return err
		}
		if clientErr, ok := apperrors.AsClientValidationError(err); ok {
			return clientErr
		}
		return apperrors.NewInternalError(fmt.Sprintf("failed to update parent rollups for parent %s", parentID), err)
	}

	return nil
}

// RecalculateCountRollupValues returns authoritative COUNT rollup values for a
// single parent row. Async rollup-delta drains use this after locking the parent
// so delayed insert markers reconcile against live child rows instead of blindly
// applying stale relative deltas.
func (rs *RollupService) RecalculateCountRollupValues(ctx context.Context, parentObjName, parentID string, rollupFieldNames []string) (models.SObject, error) {
	if rs == nil || rs.metadata == nil {
		return nil, apperrors.NewInternalError("rollup service not wired", nil)
	}
	schema := rs.metadata.GetSchema(ctx, parentObjName)
	if schema == nil {
		return nil, apperrors.NewNotFoundError("Object", parentObjName)
	}

	updates := make(models.SObject, len(rollupFieldNames))
	for _, fieldName := range rollupFieldNames {
		field := models.FindField(schema, fieldName)
		if field == nil || field.RollupConfig == nil || !strings.EqualFold(string(field.Type), string(constants.FieldTypeRollupSummary)) || !strings.EqualFold(strings.TrimSpace(field.RollupConfig.CalcType), "COUNT") {
			return nil, apperrors.NewInternalError(fmt.Sprintf("rollup delta reconcile rejected non-COUNT field %s.%s", parentObjName, fieldName), nil)
		}
		newVal, err := rs.CalculateRollup(ctx, AffectedRollup{
			ParentObjName: parentObjName,
			ParentID:      parentID,
			RollupField:   *field,
		})
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("failed to recalculate rollup %s.%s for parent %s", parentObjName, field.APIName, parentID), err)
		}
		updates[field.APIName] = newVal
	}
	return updates, nil
}

// rollupDepthFromCtx returns the current depth (0 = top-level call).
func rollupDepthFromCtx(ctx context.Context) int {
	d, _ := ctx.Value(rollupDepthCtxKey{}).(int)
	return d
}

// rollupChainFromCtx returns the accumulated chain string for error messages.
func rollupChainFromCtx(ctx context.Context) string {
	s, _ := ctx.Value(rollupChainCtxKey{}).(string)
	return s
}

// withRollupDepth returns a new context with depth incremented by 1 and the
// new hop appended to the chain string.
func withRollupDepth(ctx context.Context, hop string) context.Context {
	newDepth := rollupDepthFromCtx(ctx) + 1
	chain := rollupChainFromCtx(ctx)
	if chain == "" {
		chain = hop
	} else {
		chain = chain + " → " + hop
	}
	ctx = context.WithValue(ctx, rollupDepthCtxKey{}, newDepth)
	ctx = context.WithValue(ctx, rollupChainCtxKey{}, chain)
	return ctx
}

// RollupService handles rollup summary field calculations
type RollupService struct {
	repo      ports.RollupRepository
	metadata  MetadataSource
	txManager ports.TransactionManager
	logger    *logging.Logger
	derived   ports.DerivedWriter
	// now is the clock RecalculateAll measures a deadline against. It is a seam
	// so the deadline guard is testable without sleeping. A nil clock falls back
	// to time.Now so a RollupService built as a struct literal rather than
	// through NewRollupService cannot read the zero instant and treat every
	// deadline as already expired.
	now func() time.Time
}

// NewRollupService creates a new RollupService
func NewRollupService(repo ports.RollupRepository, metadata MetadataSource, txManager ports.TransactionManager, logger *logging.Logger) *RollupService {
	return &RollupService{
		repo:      repo,
		metadata:  metadata,
		txManager: txManager,
		logger:    logger,
		now:       time.Now,
	}
}

// nowTime reads the service clock, falling back to the real clock when the
// service was built without one.
func (rs *RollupService) nowTime() time.Time {
	if rs.now == nil {
		return time.Now()
	}
	return rs.now()
}

// SetDerivedWriter wires the records-side derived-field writer post-construction
// to avoid circular DI (PersistenceService already holds a RollupProcessor).
func (rs *RollupService) SetDerivedWriter(dw ports.DerivedWriter) { rs.derived = dw }

func (rs *RollupService) lockParent(ctx context.Context, parentObjName, parentID string) error {
	if rs.txManager == nil || !rs.txManager.IsTransactionActive(ctx) || !rs.txManager.IsPessimisticTx(ctx) {
		rollupLockSkippedOnOptimistic.Inc()
		return errRollupLockOnOptimisticTx
	}
	if rs.derived == nil {
		return apperrors.NewInternalError("rollup derived writer not wired", nil)
	}
	return rs.derived.LockDerivedRecord(ctx, parentObjName, parentID)
}

func retryablePrelockError(err error) bool {
	var retryErr recordlocks.RetryablePrelockError
	return errors.As(err, &retryErr)
}

func shouldPrelockAffectedParents(ctx context.Context) bool {
	return recordlocks.ScopeFromContext(ctx) != nil
}

func (rs *RollupService) prelockAffectedParents(ctx context.Context, affected []AffectedRollup) error {
	if len(affected) == 0 || !shouldPrelockAffectedParents(ctx) {
		return nil
	}
	keys := append([]recordlocks.ParentKey(nil), recordlocks.ForcedParentPrelocksFromContext(ctx)...)
	for _, item := range affected {
		keys = append(keys, recordlocks.ParentKey{
			ObjectName: item.ParentObjName,
			RecordID:   item.ParentID,
		})
	}
	for _, key := range recordlocks.OrderedParentKeys(keys) {
		if err := rs.lockParent(ctx, key.ObjectName, key.RecordID); err != nil {
			return err
		}
	}
	return nil
}

func (rs *RollupService) prelockRecalculateParent(ctx context.Context, parentObjName, parentID string) error {
	keys := append([]recordlocks.ParentKey(nil), recordlocks.ForcedParentPrelocksFromContext(ctx)...)
	keys = append(keys, recordlocks.ParentKey{ObjectName: parentObjName, RecordID: parentID})
	for _, key := range recordlocks.OrderedParentKeys(keys) {
		if err := rs.lockParent(ctx, key.ObjectName, key.RecordID); err != nil {
			return err
		}
	}
	return nil
}

// processAffectedRollups is the shared calculation/update loop used by
// ProcessRollups, ProcessRollupsForUpdate, and ProcessBulkRollups.
// It iterates the affected rollup set, checks the visited-key cycle guard,
// calculates each rollup, and writes the result via DerivedWriter.
func (rs *RollupService) processAffectedRollups(ctx context.Context, childObjName string, affected []AffectedRollup, visited map[rollupVisitedKey]struct{}) error {
	var systemUser *models.UserSession
	for _, group := range groupAffectedRollupsByParent(affected) {
		groupCtx := ctx
		fieldNames := make([]string, 0, len(group.Items))
		for _, item := range group.Items {
			vk := rollupVisitedKey{
				childObj:    childObjName,
				parentObj:   item.ParentObjName,
				parentID:    item.ParentID,
				rollupField: item.RollupField.APIName,
			}
			if _, seen := visited[vk]; seen {
				return apperrors.NewInternalError(fmt.Sprintf(
					"rollup cycle detected at runtime: repeated tuple (childObj=%s, parentObj=%s, parentID=%s, rollupField=%s); "+
						"check rollup field metadata for cycles",
					childObjName, item.ParentObjName, item.ParentID, item.RollupField.APIName,
				), nil)
			}
			fieldNames = append(fieldNames, item.RollupField.APIName)
			groupCtx = withRollupVisited(groupCtx, vk)
		}

		hop := fmt.Sprintf("%s→%s.[%s]", childObjName, group.ParentObjName, strings.Join(fieldNames, ","))
		groupCtx = withRollupDepth(groupCtx, hop)
		lockStarted := time.Now()
		lockErr := rs.lockParent(groupCtx, group.ParentObjName, group.ParentID)
		logRollupStageTiming(groupCtx, "processAffectedRollups.lockParent", group.ParentObjName, group.ParentID, strings.Join(fieldNames, ","), lockStarted, lockErr)
		if lockErr != nil {
			return lockErr
		}

		updates := make(models.SObject, len(group.Items))
		for _, item := range group.Items {
			calcStarted := time.Now()
			newVal, err := rs.CalculateRollup(groupCtx, item)
			logRollupStageTiming(groupCtx, "processAffectedRollups.CalculateRollup", item.ParentObjName, item.ParentID, item.RollupField.APIName, calcStarted, err)
			if err != nil {
				return apperrors.NewInternalError(fmt.Sprintf("failed to calculate rollup %s.%s", item.ParentObjName, item.RollupField.APIName), err)
			}
			updates[item.RollupField.APIName] = newVal
		}
		if rs.derived == nil {
			return apperrors.NewInternalError("rollup derived writer not wired", nil)
		}
		if systemUser == nil {
			loaded, err := rs.derived.LoadSystemUserSession(groupCtx)
			if err != nil {
				return apperrors.NewInternalError("failed to load rollup system user", err)
			}
			if loaded == nil {
				return apperrors.NewInternalError("rollup system user not found", nil)
			}
			systemUser = loaded
		}

		updateStarted := time.Now()
		updErr := rs.derived.UpdateDerived(groupCtx, group.ParentObjName, group.ParentID, updates, systemUser)
		logRollupStageTiming(groupCtx, "processAffectedRollups.UpdateDerived", group.ParentObjName, group.ParentID, strings.Join(fieldNames, ","), updateStarted, updErr)
		if updErr != nil {
			if retryablePrelockError(updErr) {
				return updErr
			}
			if clientErr, ok := apperrors.AsClientValidationError(updErr); ok {
				return clientErr
			}
			return apperrors.NewInternalError(fmt.Sprintf("failed to update parent rollup %s", group.ParentID), updErr)
		}
	}
	return nil
}

func (rs *RollupService) processCountInsertRollupsIncrementally(ctx context.Context, childObjName string, childRecord models.SObject, affected []AffectedRollup, visited map[rollupVisitedKey]struct{}) (bool, error) {
	if len(affected) == 0 {
		return false, nil
	}
	recordID, _ := childRecord[constants.FieldID].(string)
	if strings.TrimSpace(recordID) == "" {
		return false, nil
	}
	for _, item := range affected {
		config := item.RollupField.RollupConfig
		if config == nil || !strings.EqualFold(strings.TrimSpace(config.CalcType), "COUNT") {
			return false, nil
		}
		if err := validateRollupFilter(config); err != nil {
			return false, err
		}
	}

	var systemUser *models.UserSession
	for _, group := range groupAffectedRollupsByParent(affected) {
		groupCtx := ctx
		fieldNames := make([]string, 0, len(group.Items))
		for _, item := range group.Items {
			vk := rollupVisitedKey{
				childObj:    childObjName,
				parentObj:   item.ParentObjName,
				parentID:    item.ParentID,
				rollupField: item.RollupField.APIName,
			}
			if _, seen := visited[vk]; seen {
				return false, apperrors.NewInternalError(fmt.Sprintf(
					"rollup cycle detected at runtime: repeated tuple (childObj=%s, parentObj=%s, parentID=%s, rollupField=%s); "+
						"check rollup field metadata for cycles",
					childObjName, item.ParentObjName, item.ParentID, item.RollupField.APIName,
				), nil)
			}
			fieldNames = append(fieldNames, item.RollupField.APIName)
			groupCtx = withRollupVisited(groupCtx, vk)
		}

		hop := fmt.Sprintf("%s→%s.[%s]", childObjName, group.ParentObjName, strings.Join(fieldNames, ","))
		groupCtx = withRollupDepth(groupCtx, hop)
		if rs.derived == nil {
			return false, apperrors.NewInternalError("rollup derived writer not wired", nil)
		}
		if rs.parentObjectMayFeedRollups(groupCtx, group.ParentObjName) {
			return false, nil
		}
		canApplyDelta, err := rs.derived.CanApplyDerivedCountRollupDeltas(groupCtx, group.ParentObjName, fieldNames...)
		if err != nil {
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to check parent rollup delta safety for %s", group.ParentObjName), err)
		}
		if !canApplyDelta {
			return false, nil
		}

		deltas := make(map[string]int64, len(group.Items))
		for _, item := range group.Items {
			matches, err := rs.repo.RecordMatchesRollupFilter(groupCtx, item.RollupField.RollupConfig.SummaryObject, recordID, item.RollupField.RollupConfig.Filter)
			if err != nil {
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to evaluate rollup filter %s.%s", item.ParentObjName, item.RollupField.APIName), err)
			}
			if matches {
				deltas[item.RollupField.APIName]++
			}
		}
		if len(deltas) == 0 {
			continue
		}
		isEventual, err := rs.derived.CountRollupFieldsDeclareEventual(groupCtx, group.ParentObjName, fieldNames...)
		if err != nil {
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to check rollup maintenance mode for %s", group.ParentObjName), err)
		}
		if isEventual {
			if err := rs.derived.EnqueueDerivedCountRollupDeltas(groupCtx, childObjName, recordID, group.ParentObjName, group.ParentID, deltas); err != nil {
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to enqueue parent rollup delta %s", group.ParentID), err)
			}
		} else {
			if systemUser == nil {
				loaded, err := rs.derived.LoadSystemUserSession(groupCtx)
				if err != nil {
					return false, apperrors.NewInternalError("failed to load rollup system user", err)
				}
				if loaded == nil {
					return false, apperrors.NewInternalError("rollup system user not found", nil)
				}
				systemUser = loaded
			}
			if err := rs.derived.ApplyDerivedCountRollupDeltas(groupCtx, group.ParentObjName, group.ParentID, deltas, systemUser); err != nil {
				if retryablePrelockError(err) {
					return false, err
				}
				if clientErr, ok := apperrors.AsClientValidationError(err); ok {
					return false, clientErr
				}
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to update parent rollup %s", group.ParentID), err)
			}
		}
	}
	return true, nil
}

func (rs *RollupService) parentObjectMayFeedRollups(ctx context.Context, parentObjName string) bool {
	return hasRollupChildDependencies(ctx, rs.metadata, parentObjName)
}

// rollupCountUpdatePreimageCtxKey carries the pre-update filter-match results
// (oldMatch) captured BEFORE repo.Update mutates the child row. The update
// COUNT-delta path needs to know whether the OLD physical child row satisfied
// each rollup filter; once repo.Update lands, a DB filter probe yields the NEW
// match instead, so the old answer must be snapshotted earlier in the save.
type rollupCountUpdatePreimageCtxKey struct{}

type rollupFilterMatchIdentity struct {
	summaryObject string
	filter        string
}

// rollupFilterKey normalizes a rollup filter pointer into a stable map key.
// Two rollup definitions that share the same filter string (against the same
// child object + record) have the same match answer, so they share one probe.
func rollupFilterKey(f *string) string {
	if f == nil {
		return ""
	}
	return *f
}

// rollupCountUpdatePreimageFromCtx returns the captured oldMatch map and whether
// it was staged. When absent (ok=false) the update delta path must fall back to
// full recompute rather than guess — fail-closed on correctness.
func rollupCountUpdatePreimageFromCtx(ctx context.Context) (map[rollupFilterMatchIdentity]bool, bool) {
	m, ok := ctx.Value(rollupCountUpdatePreimageCtxKey{}).(map[rollupFilterMatchIdentity]bool)
	return m, ok
}

// recordMatchesRollupFilters deduplicates filters while preserving their first
// observed order, then uses the production repository's optional batch surface.
// Narrow adapters keep the exact legacy one-filter-at-a-time behavior.
func (rs *RollupService) recordMatchesRollupFilters(
	ctx context.Context,
	summaryObject string,
	recordID string,
	filters []*string,
) (map[string]bool, error) {
	unique := make([]*string, 0, len(filters))
	keys := make([]string, 0, len(filters))
	seen := make(map[string]struct{}, len(filters))
	for _, filter := range filters {
		key := rollupFilterKey(filter)
		if _, exists := seen[key]; exists {
			continue
		}
		seen[key] = struct{}{}
		unique = append(unique, filter)
		keys = append(keys, key)
	}
	if len(unique) == 0 {
		return map[string]bool{}, nil
	}

	results := make([]bool, len(unique))
	if batch, ok := rs.repo.(ports.RollupFilterBatchRepository); ok {
		matched, err := batch.RecordMatchesRollupFilters(ctx, summaryObject, recordID, unique)
		if err != nil {
			return nil, err
		}
		if len(matched) != len(unique) {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf(
					"rollup filter batch result count mismatch for %s: got %d, want %d",
					summaryObject, len(matched), len(unique),
				),
				nil,
			)
		}
		copy(results, matched)
	} else {
		for index, filter := range unique {
			matched, err := rs.repo.RecordMatchesRollupFilter(ctx, summaryObject, recordID, filter)
			if err != nil {
				return nil, err
			}
			results[index] = matched
		}
	}

	byFilter := make(map[string]bool, len(keys))
	for index, key := range keys {
		byFilter[key] = results[index]
	}
	return byFilter, nil
}

// rollupParentKey identifies one parent row (object + id) for delta accumulation.
type rollupParentKey struct {
	obj string
	id  string
}

// StageRollupCountUpdatePreimage probes, for every COUNT rollup whose old child
// image links a parent, whether the pre-update child row satisfies the rollup
// filter, and stashes the answers on the returned context. It MUST be called
// after the child row is locked and BEFORE repo.Update writes new values, so the
// probe reads the committed old physical row. Non-COUNT rollups are skipped (the
// update delta path only applies to COUNT). The probe reuses the sole
// authoritative filter-evaluation read path (RecordMatchesRollupFilter), so no
// second filter-evaluation surface is introduced.
func (rs *RollupService) StageRollupCountUpdatePreimage(ctx context.Context, childObjName string, oldRecord models.SObject) (context.Context, error) {
	if oldRecord == nil {
		return ctx, nil
	}
	recordID, _ := oldRecord[constants.FieldID].(string)
	if strings.TrimSpace(recordID) == "" {
		return ctx, nil
	}
	affected, err := rs.FindAffectedRollups(ctx, childObjName, oldRecord)
	if err != nil {
		return ctx, err
	}
	filtersByObject := map[string][]*string{}
	for _, item := range affected {
		config := item.RollupField.RollupConfig
		if config == nil || !strings.EqualFold(strings.TrimSpace(config.CalcType), "COUNT") {
			continue
		}
		if err := validateRollupFilter(config); err != nil {
			return ctx, err
		}
		filtersByObject[config.SummaryObject] = append(filtersByObject[config.SummaryObject], config.Filter)
	}
	matchByFilter := map[rollupFilterMatchIdentity]bool{}
	objects := make([]string, 0, len(filtersByObject))
	for object := range filtersByObject {
		objects = append(objects, object)
	}
	sort.Strings(objects)
	for _, object := range objects {
		matched, err := rs.recordMatchesRollupFilters(
			ctx, object, recordID, filtersByObject[object],
		)
		if err != nil {
			return ctx, apperrors.NewInternalError(
				fmt.Sprintf("failed to evaluate pre-update rollup filters for %s", childObjName), err,
			)
		}
		for filter, value := range matched {
			matchByFilter[rollupFilterMatchIdentity{
				summaryObject: object,
				filter:        filter,
			}] = value
		}
	}
	if len(matchByFilter) == 0 {
		return ctx, nil
	}
	return context.WithValue(ctx, rollupCountUpdatePreimageCtxKey{}, matchByFilter), nil
}

// rollupUpdateDepKey identifies a single rollup definition (parent object +
// parent rollup field). The parent RECORD id is intentionally excluded: a
// reparent changes the id but the definition is the same, and the delta path
// pairs the old-image parent against the new-image parent under one definition.
type rollupUpdateDepKey struct {
	parentObj string
	fieldAPI  string
}

// rollupParentDelta accumulates per-field COUNT deltas for one parent row.
type rollupParentDelta struct {
	obj    string
	id     string
	fields map[string]models.FieldMetadata
	deltas map[string]float64
}

func boolToInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

// processCountUpdateRollupsIncrementally applies ±1 COUNT deltas for a child
// update instead of re-COUNTing every sibling under the parent. It is the
// update-path analogue of processCountInsertRollupsIncrementally and the
// general solution of which insert (oldContrib=0) and delete (newContrib=0) are
// special cases.
//
// For each affected COUNT rollup definition it derives:
//
//	oldContrib = (oldParent linked && oldRow matched filter) ? 1 : 0
//	newContrib = (newParent linked && newRow matched filter) ? 1 : 0
//
// Same parent  → delta = newContrib - oldContrib on that parent.
// Reparent     → -oldContrib on old parent, +newContrib on new parent.
//
// oldMatch comes from the pre-update snapshot staged on ctx (the old physical
// row is gone after repo.Update); newMatch is probed live via the sole
// authoritative filter read path. The whole batch falls back to full recompute
// (returns applied=false) if any affected rollup is non-COUNT or the pre-update
// snapshot is missing — fail-closed on correctness, never guess a delta.
func (rs *RollupService) processCountUpdateRollupsIncrementally(ctx context.Context, childObjName string, finalRecord, oldRecord models.SObject, finalAffected, oldAffected []AffectedRollup, visited map[rollupVisitedKey]struct{}) (bool, error) {
	if len(finalAffected) == 0 && len(oldAffected) == 0 {
		return false, nil
	}
	preimage, ok := rollupCountUpdatePreimageFromCtx(ctx)
	if !ok {
		return false, nil
	}

	newByDep := make(map[rollupUpdateDepKey]AffectedRollup, len(finalAffected))
	oldByDep := make(map[rollupUpdateDepKey]AffectedRollup, len(oldAffected))
	depOrder := make([]rollupUpdateDepKey, 0, len(finalAffected)+len(oldAffected))
	seenDep := map[rollupUpdateDepKey]struct{}{}
	addDep := func(item AffectedRollup, into map[rollupUpdateDepKey]AffectedRollup) (bool, error) {
		config := item.RollupField.RollupConfig
		if config == nil || !strings.EqualFold(strings.TrimSpace(config.CalcType), "COUNT") {
			return false, nil
		}
		if err := validateRollupFilter(config); err != nil {
			return false, err
		}
		key := rollupUpdateDepKey{parentObj: item.ParentObjName, fieldAPI: item.RollupField.APIName}
		into[key] = item
		if _, exists := seenDep[key]; !exists {
			seenDep[key] = struct{}{}
			depOrder = append(depOrder, key)
		}
		return true, nil
	}
	for _, item := range finalAffected {
		okCount, err := addDep(item, newByDep)
		if err != nil {
			return false, err
		}
		if !okCount {
			return false, nil
		}
	}
	for _, item := range oldAffected {
		okCount, err := addDep(item, oldByDep)
		if err != nil {
			return false, err
		}
		if !okCount {
			return false, nil
		}
	}

	recordID, _ := finalRecord[constants.FieldID].(string)
	if strings.TrimSpace(recordID) == "" {
		recordID, _ = oldRecord[constants.FieldID].(string)
	}
	if strings.TrimSpace(recordID) == "" {
		return false, nil
	}

	newFiltersByObject := map[string][]*string{}
	for _, dep := range depOrder {
		item, hasNew := newByDep[dep]
		if !hasNew || strings.TrimSpace(item.ParentID) == "" || item.RollupField.RollupConfig == nil {
			continue
		}
		config := item.RollupField.RollupConfig
		newFiltersByObject[config.SummaryObject] = append(
			newFiltersByObject[config.SummaryObject], config.Filter,
		)
	}
	newMatchCache := map[rollupFilterMatchIdentity]bool{}
	newMatchObjects := make([]string, 0, len(newFiltersByObject))
	for object := range newFiltersByObject {
		newMatchObjects = append(newMatchObjects, object)
	}
	sort.Strings(newMatchObjects)
	for _, object := range newMatchObjects {
		matched, err := rs.recordMatchesRollupFilters(
			ctx, object, recordID, newFiltersByObject[object],
		)
		if err != nil {
			return false, apperrors.NewInternalError(
				fmt.Sprintf("failed to evaluate post-update rollup filters for %s", childObjName), err,
			)
		}
		for filter, value := range matched {
			newMatchCache[rollupFilterMatchIdentity{summaryObject: object, filter: filter}] = value
		}
	}
	parents := map[rollupParentKey]*rollupParentDelta{}
	parentOrder := make([]rollupParentKey, 0, len(depOrder))
	accumulate := func(obj, id string, field models.FieldMetadata, d float64) {
		if id == "" || d == 0 {
			return
		}
		pk := rollupParentKey{obj: obj, id: id}
		pd, exists := parents[pk]
		if !exists {
			pd = &rollupParentDelta{obj: obj, id: id, fields: map[string]models.FieldMetadata{}, deltas: map[string]float64{}}
			parents[pk] = pd
			parentOrder = append(parentOrder, pk)
		}
		pd.fields[field.APIName] = field
		pd.deltas[field.APIName] += d
	}

	for _, dep := range depOrder {
		newItem, hasNew := newByDep[dep]
		oldItem, hasOld := oldByDep[dep]
		var field models.FieldMetadata
		var config *models.RollupConfig
		if hasNew {
			field = newItem.RollupField
			config = newItem.RollupField.RollupConfig
		} else {
			field = oldItem.RollupField
			config = oldItem.RollupField.RollupConfig
		}

		newParentID := ""
		if hasNew {
			newParentID = newItem.ParentID
		}
		oldParentID := ""
		if hasOld {
			oldParentID = oldItem.ParentID
		}

		oldMatch := false
		if oldParentID != "" {
			m, found := preimage[rollupFilterMatchIdentity{
				summaryObject: config.SummaryObject,
				filter:        rollupFilterKey(config.Filter),
			}]
			if !found {
				return false, nil // pre-update snapshot incomplete → fall back to recompute
			}
			oldMatch = m
		}
		newMatch := false
		if newParentID != "" {
			fk := rollupFilterMatchIdentity{
				summaryObject: config.SummaryObject,
				filter:        rollupFilterKey(config.Filter),
			}
			if cached, done := newMatchCache[fk]; done {
				newMatch = cached
			} else {
				return false, apperrors.NewInternalError(
					fmt.Sprintf("missing post-update rollup filter result %s.%s", dep.parentObj, dep.fieldAPI), nil,
				)
			}
		}

		oldContrib := boolToInt(oldParentID != "" && oldMatch)
		newContrib := boolToInt(newParentID != "" && newMatch)
		if oldParentID == newParentID {
			accumulate(dep.parentObj, newParentID, field, float64(newContrib-oldContrib))
		} else {
			accumulate(dep.parentObj, oldParentID, field, float64(-oldContrib))
			accumulate(dep.parentObj, newParentID, field, float64(newContrib))
		}
	}

	if len(parentOrder) == 0 {
		// All deltas cancelled to zero — values unchanged, nothing to write.
		return true, nil
	}
	sort.Slice(parentOrder, func(i, j int) bool {
		if parentOrder[i].obj != parentOrder[j].obj {
			return parentOrder[i].obj < parentOrder[j].obj
		}
		return parentOrder[i].id < parentOrder[j].id
	})

	var systemUser *models.UserSession
	for _, pk := range parentOrder {
		pd := parents[pk]
		groupCtx := ctx
		fieldNames := make([]string, 0, len(pd.deltas))
		for fieldAPI := range pd.deltas {
			vk := rollupVisitedKey{
				childObj:    childObjName,
				parentObj:   pd.obj,
				parentID:    pd.id,
				rollupField: fieldAPI,
			}
			if _, seen := visited[vk]; seen {
				return false, apperrors.NewInternalError(fmt.Sprintf(
					"rollup cycle detected at runtime: repeated tuple (childObj=%s, parentObj=%s, parentID=%s, rollupField=%s); "+
						"check rollup field metadata for cycles",
					childObjName, pd.obj, pd.id, fieldAPI,
				), nil)
			}
			fieldNames = append(fieldNames, fieldAPI)
			groupCtx = withRollupVisited(groupCtx, vk)
		}
		sort.Strings(fieldNames)

		hop := fmt.Sprintf("%s→%s.[%s]", childObjName, pd.obj, strings.Join(fieldNames, ","))
		groupCtx = withRollupDepth(groupCtx, hop)
		if err := rs.lockParent(groupCtx, pd.obj, pd.id); err != nil {
			return false, err
		}
		if rs.derived == nil {
			return false, apperrors.NewInternalError("rollup derived writer not wired", nil)
		}
		parentRow, err := rs.derived.GetDerivedRecordForUpdate(groupCtx, pd.obj, pd.id)
		if err != nil {
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to read locked parent %s", pd.id), err)
		}

		updates := make(models.SObject, len(pd.deltas))
		for _, fieldAPI := range fieldNames {
			current, err := rollupNumericValueAsFloat64(parentRow[fieldAPI])
			if err != nil {
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to parse parent rollup value %s.%s", pd.obj, fieldAPI), err)
			}
			next := current + pd.deltas[fieldAPI]
			if next < 0 {
				// COUNT is a non-negative invariant; a negative result would
				// only arise from pre-existing drift, and recompute would also
				// floor at 0. Clamp rather than persist an impossible count.
				next = 0
			}
			updates[fieldAPI] = next
		}
		if systemUser == nil {
			loaded, err := rs.derived.LoadSystemUserSession(groupCtx)
			if err != nil {
				return false, apperrors.NewInternalError("failed to load rollup system user", err)
			}
			if loaded == nil {
				return false, apperrors.NewInternalError("rollup system user not found", nil)
			}
			systemUser = loaded
		}
		if err := rs.derived.UpdateDerived(groupCtx, pd.obj, pd.id, updates, systemUser); err != nil {
			if retryablePrelockError(err) {
				return false, err
			}
			if clientErr, ok := apperrors.AsClientValidationError(err); ok {
				return false, clientErr
			}
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to update parent rollup %s", pd.id), err)
		}
	}
	return true, nil
}

func rollupNumericValueAsFloat64(v interface{}) (float64, error) {
	switch typed := v.(type) {
	case nil:
		return 0, nil
	case float64:
		return typed, nil
	case float32:
		return float64(typed), nil
	case int:
		return float64(typed), nil
	case int64:
		return float64(typed), nil
	case int32:
		return float64(typed), nil
	case uint64:
		return float64(typed), nil
	case uint32:
		return float64(typed), nil
	case string:
		if strings.TrimSpace(typed) == "" {
			return 0, nil
		}
		return strconv.ParseFloat(typed, 64)
	case []byte:
		if len(typed) == 0 {
			return 0, nil
		}
		return strconv.ParseFloat(string(typed), 64)
	default:
		return 0, fmt.Errorf("unsupported numeric rollup type %T", v)
	}
}

type affectedParentGroup struct {
	ParentObjName string
	ParentID      string
	Items         []AffectedRollup
}

func groupAffectedRollupsByParent(affected []AffectedRollup) []affectedParentGroup {
	if len(affected) == 0 {
		return nil
	}
	groups := make([]affectedParentGroup, 0, len(affected))
	for _, item := range affected {
		last := len(groups) - 1
		if last < 0 || groups[last].ParentObjName != item.ParentObjName || groups[last].ParentID != item.ParentID {
			groups = append(groups, affectedParentGroup{
				ParentObjName: item.ParentObjName,
				ParentID:      item.ParentID,
				Items:         []AffectedRollup{item},
			})
			continue
		}
		groups[last].Items = append(groups[last].Items, item)
	}
	return groups
}

// ProcessInsertRollups handles rollups for a newly inserted child row and may
// use COUNT-specific delta updates that are only safe for inserts.
func (rs *RollupService) ProcessInsertRollups(ctx context.Context, childObjName string, childRecord models.SObject) error {
	return rs.processRollups(ctx, childObjName, childRecord, true)
}

// ProcessRollups handles generic child mutations. Callers that are not known
// inserts must use this path so rollups are recalculated from live child rows.
func (rs *RollupService) ProcessRollups(ctx context.Context, childObjName string, childRecord models.SObject) error {
	return rs.processRollups(ctx, childObjName, childRecord, false)
}

// processRollups orchestration method.
// The ctx parameter ensures rollup updates participate in the caller's transaction for ACID compliance (if injected).
//
// Runtime depth guard: ProcessRollups tracks re-entrancy depth via the context
// to prevent runaway loops caused by cyclic rollup graphs or metadata-out-of-sync
// states that bypass the pack-apply cycle validator.  Depth is capped at
// maxRollupDepth (8).  On exceed the call fails closed with an explicit error
// naming the cap and the accumulated rollup chain.
func (rs *RollupService) processRollups(ctx context.Context, childObjName string, childRecord models.SObject, allowCountInsertDelta bool) error {
	depth := rollupDepthFromCtx(ctx)
	if depth >= maxRollupDepth {
		chain := rollupChainFromCtx(ctx)
		return apperrors.NewInternalError(fmt.Sprintf(
			"rollup depth cap exceeded (depth=%d, cap=%d): rollup chain %s; "+
				"check rollup field metadata for cycles or excessively deep hierarchies",
			depth, maxRollupDepth, chain,
		), nil)
	}

	affected, err := rs.FindAffectedRollups(ctx, childObjName, childRecord)
	if err != nil {
		return err
	}
	sort.Sort(affectedRollupByParentKey(affected))
	visited := rollupVisitedFromCtx(ctx)
	if allowCountInsertDelta {
		canSkipPrelock, err := rs.canProcessInsertRollupsWithoutParentPrelock(ctx, childObjName, childRecord, affected)
		if err != nil {
			return err
		}
		if canSkipPrelock {
			if applied, err := rs.processCountInsertRollupsIncrementally(ctx, childObjName, childRecord, affected, visited); applied || err != nil {
				return err
			}
		}
	}

	if err := rs.prelockAffectedParents(ctx, affected); err != nil {
		return err
	}

	if allowCountInsertDelta {
		if applied, err := rs.processCountInsertRollupsIncrementally(ctx, childObjName, childRecord, affected, visited); applied || err != nil {
			return err
		}
	}
	return rs.processAffectedRollups(ctx, childObjName, affected, visited)
}

// AffectedParentRows exposes the same rollup parent discovery used by
// ProcessRollups, but as lock keys. Records callers use it before acquiring a
// child-row lock so direct child writes and parent cascade deletes take parent
// locks in the same order.
func (rs *RollupService) AffectedParentRows(ctx context.Context, childObjName string, childRecord models.SObject) ([]recordlocks.ParentKey, error) {
	return rs.affectedParentRows(ctx, childObjName, childRecord, false)
}

// AffectedParentRowsForInsert exposes the insert-specific lockset. Explicit
// eventual COUNT rollups whose durable delta path can run before parent prelock
// are intentionally omitted from this set.
func (rs *RollupService) AffectedParentRowsForInsert(ctx context.Context, childObjName string, childRecord models.SObject) ([]recordlocks.ParentKey, error) {
	return rs.affectedParentRows(ctx, childObjName, childRecord, true)
}

func (rs *RollupService) affectedParentRows(ctx context.Context, childObjName string, childRecord models.SObject, allowInsertDeltaSkip bool) ([]recordlocks.ParentKey, error) {
	affected, err := rs.FindAffectedRollups(ctx, childObjName, childRecord)
	if err != nil {
		return nil, err
	}
	sort.Sort(affectedRollupByParentKey(affected))
	canSkip := false
	if allowInsertDeltaSkip {
		canSkip, err = rs.canProcessInsertRollupsWithoutParentPrelock(ctx, childObjName, childRecord, affected)
		if err != nil {
			return nil, err
		}
	}
	keys := make([]recordlocks.ParentKey, 0, len(affected))
	for _, group := range groupAffectedRollupsByParent(affected) {
		if canSkip {
			skipGroup, err := rs.affectedParentGroupCanUseEventualInsertDeltas(ctx, group)
			if err != nil {
				return nil, err
			}
			if skipGroup {
				continue
			}
		}
		if err := rs.augmentRollupHistorySequencePlan(ctx, group.ParentObjName, group.Items); err != nil {
			return nil, err
		}
		keys = append(keys, recordlocks.ParentKey{ObjectName: group.ParentObjName, RecordID: group.ParentID})
	}
	return recordlocks.OrderedParentKeys(keys), nil
}

// augmentRollupHistorySequencePlan reuses the exact affected-rollup closure
// that owns the parent-row plan. Audit V2 serializes every persisted field
// change, not only fields exposed through the user-facing TrackHistory flag,
// so every stable derived-field identity participates. Attempt-local state is
// optional so legacy/non-retrying callers preserve their existing behavior.
func (rs *RollupService) augmentRollupHistorySequencePlan(ctx context.Context, parentObject string, affected []AffectedRollup) error {
	fieldIDs := make([]string, 0, len(affected))
	fieldNames := make([]string, 0, len(affected))
	for _, item := range affected {
		if strings.TrimSpace(item.RollupField.ID) != "" {
			fieldIDs = append(fieldIDs, item.RollupField.ID)
		}
		fieldNames = append(fieldNames, item.RollupField.APIName)
	}
	if planner, ok := rs.derived.(ports.DerivedUpdateHistoryPlanner); ok {
		planned, err := planner.DerivedUpdateHistoryFieldIDs(ctx, parentObject, fieldNames...)
		if err != nil {
			return apperrors.NewInternalError("plan derived rollup history fields", err)
		}
		fieldIDs = planned
	}
	ports.AugmentPlannedHistorySequenceLocks(ctx, fieldIDs)
	return nil
}

func (rs *RollupService) canProcessInsertRollupsWithoutParentPrelock(ctx context.Context, childObjName string, childRecord models.SObject, affected []AffectedRollup) (bool, error) {
	if len(affected) == 0 {
		return false, nil
	}
	recordID, _ := childRecord[constants.FieldID].(string)
	if strings.TrimSpace(recordID) == "" {
		return false, nil
	}
	for _, group := range groupAffectedRollupsByParent(affected) {
		ok, err := rs.affectedParentGroupCanUseEventualInsertDeltas(ctx, group)
		if err != nil || !ok {
			return ok, err
		}
	}
	return true, nil
}

func (rs *RollupService) affectedParentGroupCanUseEventualInsertDeltas(ctx context.Context, group affectedParentGroup) (bool, error) {
	if rs == nil || rs.derived == nil || len(group.Items) == 0 {
		return false, nil
	}
	fieldNames := make([]string, 0, len(group.Items))
	for _, item := range group.Items {
		config := item.RollupField.RollupConfig
		if config == nil || !strings.EqualFold(strings.TrimSpace(config.CalcType), "COUNT") {
			return false, nil
		}
		if err := validateRollupFilter(config); err != nil {
			return false, err
		}
		fieldNames = append(fieldNames, item.RollupField.APIName)
	}
	if rs.parentObjectMayFeedRollups(ctx, group.ParentObjName) {
		return false, nil
	}
	canApplyDelta, err := rs.derived.CanApplyDerivedCountRollupDeltas(ctx, group.ParentObjName, fieldNames...)
	if err != nil || !canApplyDelta {
		return canApplyDelta, err
	}
	isEventual, err := rs.derived.CountRollupFieldsDeclareEventual(ctx, group.ParentObjName, fieldNames...)
	if err != nil || !isEventual {
		return isEventual, err
	}
	return true, nil
}

// affectedRollupByParentKey sorts AffectedRollup items by the deterministic
// lockset key (parentObj, parentID, rollupField) to prevent deadlocks when
// concurrent transactions reparent children in opposite directions.
type affectedRollupByParentKey []AffectedRollup

func (a affectedRollupByParentKey) Len() int      { return len(a) }
func (a affectedRollupByParentKey) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a affectedRollupByParentKey) Less(i, j int) bool {
	if a[i].ParentObjName != a[j].ParentObjName {
		return a[i].ParentObjName < a[j].ParentObjName
	}
	if a[i].ParentID != a[j].ParentID {
		return a[i].ParentID < a[j].ParentID
	}
	return a[i].RollupField.APIName < a[j].RollupField.APIName
}

// affectedRollupDedupKey is the dedup key for merging affected rollups from
// both child images.  Two AffectedRollup items with the same
// (parentObj, parentID, rollupField) represent the same parent rollup to
// recalculate and must be deduplicated.
type affectedRollupDedupKey struct {
	parentObj   string
	parentID    string
	rollupField string
}

// ProcessRollupsForUpdate handles rollup processing for a child record update
// that may involve a reparent (parent field change).  It merges the affected
// rollup sets from both the final (new) and old (previous) child images,
// deduplicates by (parentObj, parentID, rollupField), sorts the merged set
// by stable key for deterministic lock acquisition order, and processes each
// unique parent exactly once.
//
// This prevents two concurrent reparent transactions (P1→P2 and P2→P1) from
// acquiring parent locks in conflicting orders, which would cause a deadlock
// under the old two-call ProcessRollups approach.
func (rs *RollupService) ProcessRollupsForUpdate(ctx context.Context, childObjName string, finalRecord, oldRecord models.SObject) error {
	depth := rollupDepthFromCtx(ctx)
	if depth >= maxRollupDepth {
		chain := rollupChainFromCtx(ctx)
		return apperrors.NewInternalError(fmt.Sprintf(
			"rollup depth cap exceeded (depth=%d, cap=%d): rollup chain %s; "+
				"check rollup field metadata for cycles or excessively deep hierarchies",
			depth, maxRollupDepth, chain,
		), nil)
	}

	affected, err := rs.affectedRollupsForUpdate(ctx, childObjName, finalRecord, oldRecord)
	if err != nil {
		return err
	}
	if len(affected.merged) == 0 {
		return nil
	}
	if err := rs.prelockAffectedParents(ctx, affected.merged); err != nil {
		return err
	}

	visited := rollupVisitedFromCtx(ctx)
	// COUNT-only fast path: apply ±1 deltas instead of re-COUNTing all siblings.
	// Falls back (applied=false) to full recompute when any affected rollup is
	// non-COUNT or the pre-update filter-match snapshot is unavailable.
	if applied, err := rs.processCountUpdateRollupsIncrementally(ctx, childObjName, finalRecord, oldRecord, affected.finalAffected, affected.oldAffected, visited); applied || err != nil {
		return err
	}
	return rs.processAffectedRollups(ctx, childObjName, affected.merged, visited)
}

// ProcessBulkRollups handles rollup processing for a bulk insert of child
// records.  It calls FindAffectedRollups for each landed row, collects all
// affected tuples, deduplicates by (parentObj, parentID, rollupField), sorts
// for deterministic lock acquisition order, and processes each unique tuple
// via the shared processAffectedRollups path.
//
// Only landed rows contribute; INSERT IGNORE rows (landed=false) are excluded
// because they never committed and must not produce phantom rollup
// contributions.
func (rs *RollupService) ProcessBulkRollups(ctx context.Context, childObjName string, childRecords []models.SObject, landedRows []bool) error {
	if len(childRecords) == 0 {
		return nil
	}

	depth := rollupDepthFromCtx(ctx)
	if depth >= maxRollupDepth {
		chain := rollupChainFromCtx(ctx)
		return apperrors.NewInternalError(fmt.Sprintf(
			"rollup depth cap exceeded (depth=%d, cap=%d): rollup chain %s; "+
				"check rollup field metadata for cycles or excessively deep hierarchies",
			depth, maxRollupDepth, chain,
		), nil)
	}

	// Collect affected rollup tuples from all landed child rows.
	seen := make(map[affectedRollupDedupKey]struct{})
	var allAffected []AffectedRollup
	var affectedRows []bulkAffectedRollup
	for i, record := range childRecords {
		if i >= len(landedRows) || !landedRows[i] {
			continue
		}
		recordID, _ := record[constants.FieldID].(string)
		affected, err := rs.FindAffectedRollups(ctx, childObjName, record)
		if err != nil {
			return err
		}
		for _, item := range affected {
			affectedRows = append(affectedRows, bulkAffectedRollup{recordID: recordID, item: item})
			key := affectedRollupDedupKey{
				parentObj:   item.ParentObjName,
				parentID:    item.ParentID,
				rollupField: item.RollupField.APIName,
			}
			if _, exists := seen[key]; !exists {
				seen[key] = struct{}{}
				allAffected = append(allAffected, item)
			}
		}
	}

	if len(allAffected) == 0 {
		return nil
	}

	// Sort for deterministic lock acquisition order.
	sort.Sort(affectedRollupByParentKey(allAffected))
	visited := rollupVisitedFromCtx(ctx)
	if applied, err := rs.processBulkCountInsertRollupsIncrementally(ctx, childObjName, allAffected, affectedRows, visited); applied || err != nil {
		return err
	}
	if err := rs.prelockAffectedParents(ctx, allAffected); err != nil {
		return err
	}
	return rs.processAffectedRollups(ctx, childObjName, allAffected, visited)
}

// AffectedRollup represents a rollup calculation that needs to be performed
type AffectedRollup struct {
	ParentObjName string
	ParentID      string
	RollupField   models.FieldMetadata
}

type bulkAffectedRollup struct {
	recordID string
	item     AffectedRollup
}

func (rs *RollupService) processBulkCountInsertRollupsIncrementally(ctx context.Context, childObjName string, affected []AffectedRollup, affectedRows []bulkAffectedRollup, visited map[rollupVisitedKey]struct{}) (bool, error) {
	if len(affected) == 0 {
		return false, nil
	}
	for _, row := range affectedRows {
		if strings.TrimSpace(row.recordID) == "" {
			return false, nil
		}
	}
	for _, item := range affected {
		config := item.RollupField.RollupConfig
		if config == nil || !strings.EqualFold(strings.TrimSpace(config.CalcType), "COUNT") {
			return false, nil
		}
		if err := validateRollupFilter(config); err != nil {
			return false, err
		}
	}

	var systemUser *models.UserSession
	for _, group := range groupAffectedRollupsByParent(affected) {
		groupCtx := ctx
		fieldNames := make([]string, 0, len(group.Items))
		for _, item := range group.Items {
			vk := rollupVisitedKey{
				childObj:    childObjName,
				parentObj:   item.ParentObjName,
				parentID:    item.ParentID,
				rollupField: item.RollupField.APIName,
			}
			if _, seen := visited[vk]; seen {
				return false, apperrors.NewInternalError(fmt.Sprintf(
					"rollup cycle detected at runtime: repeated tuple (childObj=%s, parentObj=%s, parentID=%s, rollupField=%s); "+
						"check rollup field metadata for cycles",
					childObjName, item.ParentObjName, item.ParentID, item.RollupField.APIName,
				), nil)
			}
			fieldNames = append(fieldNames, item.RollupField.APIName)
			groupCtx = withRollupVisited(groupCtx, vk)
		}

		hop := fmt.Sprintf("%s→%s.[%s]", childObjName, group.ParentObjName, strings.Join(fieldNames, ","))
		groupCtx = withRollupDepth(groupCtx, hop)
		if rs.derived == nil {
			return false, apperrors.NewInternalError("rollup derived writer not wired", nil)
		}
		if rs.parentObjectMayFeedRollups(groupCtx, group.ParentObjName) {
			return false, nil
		}
		canApplyDelta, err := rs.derived.CanApplyDerivedCountRollupDeltas(groupCtx, group.ParentObjName, fieldNames...)
		if err != nil {
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to check parent rollup delta safety for %s", group.ParentObjName), err)
		}
		if !canApplyDelta {
			return false, nil
		}

		isEventual, err := rs.derived.CountRollupFieldsDeclareEventual(groupCtx, group.ParentObjName, fieldNames...)
		if err != nil {
			return false, apperrors.NewInternalError(fmt.Sprintf("failed to check rollup maintenance mode for %s", group.ParentObjName), err)
		}

		deltas := make(map[string]int64, len(group.Items))
		for _, row := range affectedRows {
			if !strings.EqualFold(row.item.ParentObjName, group.ParentObjName) || row.item.ParentID != group.ParentID {
				continue
			}
			config := row.item.RollupField.RollupConfig
			matches, err := rs.repo.RecordMatchesRollupFilter(groupCtx, config.SummaryObject, row.recordID, config.Filter)
			if err != nil {
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to evaluate rollup filter %s.%s", row.item.ParentObjName, row.item.RollupField.APIName), err)
			}
			if !matches {
				continue
			}
			if isEventual {
				rowDeltas := map[string]int64{row.item.RollupField.APIName: 1}
				if err := rs.derived.EnqueueDerivedCountRollupDeltas(groupCtx, childObjName, row.recordID, group.ParentObjName, group.ParentID, rowDeltas); err != nil {
					return false, apperrors.NewInternalError(fmt.Sprintf("failed to enqueue parent rollup delta %s", group.ParentID), err)
				}
			} else {
				deltas[row.item.RollupField.APIName]++
			}
		}
		if !isEventual && len(deltas) > 0 {
			if systemUser == nil {
				loaded, err := rs.derived.LoadSystemUserSession(groupCtx)
				if err != nil {
					return false, apperrors.NewInternalError("failed to load rollup system user", err)
				}
				if loaded == nil {
					return false, apperrors.NewInternalError("rollup system user not found", nil)
				}
				systemUser = loaded
			}
			if err := rs.derived.ApplyDerivedCountRollupDeltas(groupCtx, group.ParentObjName, group.ParentID, deltas, systemUser); err != nil {
				if retryablePrelockError(err) {
					return false, err
				}
				if clientErr, ok := apperrors.AsClientValidationError(err); ok {
					return false, clientErr
				}
				return false, apperrors.NewInternalError(fmt.Sprintf("failed to update parent rollup %s", group.ParentID), err)
			}
		}
	}
	return true, nil
}

// FindAffectedRollups identifies parent records that need rollup recalculation
// based on a change to a child record.
func (rs *RollupService) FindAffectedRollups(ctx context.Context, childObjName string, childRecord models.SObject) ([]AffectedRollup, error) {
	var affected []AffectedRollup

	deps := rollupChildDependencies(ctx, rs.metadata, childObjName)
	for _, dep := range deps {
		relField := dep.RollupField.RollupConfig.RelationshipField
		if relField == "" {
			// Strict: RelationshipField is required on rollup config.
			continue
		}

		parentIDVal, ok := childRecord[relField]
		if !ok || parentIDVal == nil {
			continue // No parent linked, nothing to roll up
		}

		parentID, ok := parentIDVal.(string)
		if !ok || parentID == "" {
			continue
		}

		affected = append(affected, AffectedRollup{
			ParentObjName: dep.ParentObjectAPIName,
			ParentID:      parentID,
			RollupField:   dep.RollupField,
		})
	}

	return affected, nil
}

// CalculateRollup performs the aggregation query and returns the value
func (rs *RollupService) CalculateRollup(ctx context.Context, rollup AffectedRollup) (interface{}, error) {
	config := rollup.RollupField.RollupConfig

	// Repo handles defaults: SUM/AVG → 0, MIN/MAX → null.
	if err := validateRollupFilter(config); err != nil {
		return nil, err
	}

	if rs.txManager != nil && rs.txManager.IsPessimisticTx(ctx) {
		return rs.repo.CalculateRollupCurrentRead(
			ctx,
			config.CalcType,
			config.SummaryObject,
			config.SummaryField,
			config.RelationshipField,
			rollup.ParentID,
			config.Filter,
		)
	}

	return rs.repo.CalculateRollup(
		ctx,
		config.CalcType,
		config.SummaryObject,
		config.SummaryField,
		config.RelationshipField,
		rollup.ParentID,
		config.Filter,
	)
}

// isValidSQLFilter validates that a SQL filter expression contains only safe characters.
// It rejects expressions that could be used for SQL injection attacks.
// Allowed: field names, operators, literals, and common SQL keywords.
func isValidSQLFilter(filter string) bool {
	// Reject empty filters
	if filter == "" {
		return false
	}

	// Reject dangerous SQL keywords/patterns (case-insensitive check)
	lowerFilter := strings.ToLower(filter)
	dangerousPatterns := []string{
		";",                  // Statement terminator
		"--",                 // Comment
		"/*",                 // Block comment
		"drop ",              // DDL
		"delete ",            // DML (not in WHERE context)
		"insert ",            // DML
		"update ",            // DML
		"truncate ",          // DDL
		"alter ",             // DDL
		"create ",            // DDL
		"exec ",              // Execute
		"execute ",           // Execute
		"union ",             // UNION attacks
		"information_schema", // Schema probing
		"sleep(",             // Time-based attacks
		"benchmark(",         // Time-based attacks
		"load_file(",         // File access
		"into outfile",       // File write
		"into dumpfile",      // File write
	}

	for _, pattern := range dangerousPatterns {
		if strings.Contains(lowerFilter, pattern) {
			return false
		}
	}

	return true
}

// RecalculateAll batch-recalculates rollup fields for a given parent object.
// This heals rollup drift after a bulk data load or a historical migration.
//
// The pass is bounded by bounds and resumable: it stops once the count or time
// bound is reached and reports the cursor to resume from. A zero-value bounds
// walks the whole object, which is the original behavior. Bounds are evaluated
// only between parents, after the previous parent's transaction has committed,
// so the reported cursor always names a parent whose work is durable.
func (rs *RollupService) RecalculateAll(ctx context.Context, parentObjName string, bounds ports.RollupRecalculateBounds) (ports.RollupRecalculateProgress, error) {
	rs.logger.Info(fmt.Sprintf("Starting rollup recalculation for %s", parentObjName))

	var progress ports.RollupRecalculateProgress

	// Get schema for this parent object
	schema := rs.metadata.GetSchema(ctx, parentObjName)
	if schema == nil {
		return progress, apperrors.NewNotFoundError("Object", parentObjName)
	}

	// Find all rollup summary fields on this parent object
	var rollupFields []models.FieldMetadata
	for _, field := range schema.Fields {
		if field.RollupConfig != nil {
			rollupFields = append(rollupFields, field)
		}
	}

	if len(rollupFields) == 0 {
		rs.logger.Info(fmt.Sprintf("No rollup summary fields found on %s", parentObjName))
		progress.Complete = true
		return progress, nil
	}

	rs.logger.Info(fmt.Sprintf("Found %d rollup summary fields on %s", len(rollupFields), parentObjName))
	txRunner, ok := rs.derived.(rollupMaintenanceTransactionRunner)
	if !ok {
		return progress, apperrors.NewInternalError("rollup derived writer does not own the maintenance transaction boundary", nil)
	}

	afterID := bounds.AfterID
	for {
		fetchSize := rollupRecalculatePageSize
		if bounds.MaxParents > 0 {
			// Never read more parent Ids than this pass can still recalculate.
			if remaining := bounds.MaxParents - progress.Recalculated; remaining < fetchSize {
				fetchSize = remaining
			}
		}
		parentIDs, err := rs.repo.GetParentIDsPage(ctx, parentObjName, afterID, fetchSize)
		if err != nil {
			return progress, apperrors.NewInternalError("failed to get parent IDs", err)
		}
		if len(parentIDs) == 0 {
			if progress.Recalculated == 0 {
				rs.logger.Info(fmt.Sprintf("No records found in %s", parentObjName))
			}
			progress.Complete = true
			break
		}

		rs.logger.Info(fmt.Sprintf("Processing %d records in %s", len(parentIDs), parentObjName))
		boundReached := false
		drainedPage := true
		for i, parentID := range parentIDs {
			parentCtx := rs.txManager.DetachTx(ctx)
			// The records writer owns READ COMMITTED retry and carries any newly
			// discovered upstream parents into the next attempt. Acquire those
			// intents with the target before reading children, then keep the
			// aggregate and derived save in that same per-parent transaction.
			err := txRunner.RunInPessimisticTransaction(parentCtx, func(txCtx context.Context) error {
				if err := rs.prelockRecalculateParent(txCtx, parentObjName, parentID); err != nil {
					return err
				}
				return rs.recalculateParent(txCtx, parentObjName, parentID, rollupFields)
			})
			if err != nil {
				rs.logger.Error(fmt.Sprintf("RecalculateAll failed for parent %s", parentID), err)
				if clientErr, ok := apperrors.AsClientValidationError(err); ok {
					return progress, clientErr
				}
				return progress, apperrors.NewInternalError(
					fmt.Sprintf("failed to recalculate rollups for parent %s", parentID),
					err,
				)
			}

			progress.Recalculated++
			progress.NextAfterID = parentID
			afterID = parentID
			if (i+1)%100 == 0 || i+1 == len(parentIDs) {
				rs.logger.Info(fmt.Sprintf("Processed %d parents in current page", i+1))
			}

			if rs.recalculateBoundReached(bounds, progress.Recalculated) {
				boundReached = true
				drainedPage = i == len(parentIDs)-1
				break
			}
		}

		// A short page means the keyset scan reached the end of the object, but
		// only if this pass actually consumed it — a bound that fired mid-page
		// leaves parents behind even though the page was short.
		//
		// This reports Complete when a DEADLINE fires on the last parent of a
		// short page. It cannot do so for the count bound, because fetchSize is
		// clamped to the remaining count, so a page that is exactly the tail is
		// a full page rather than a short one. Exhausting MaxParents on the
		// final parent therefore costs one extra no-op request to learn the
		// scan is done.
		if drainedPage && len(parentIDs) < fetchSize {
			progress.Complete = true
			break
		}
		if boundReached {
			rs.logger.Info(fmt.Sprintf(
				"RecalculateAll: bound reached for %s after %d records; resume after %s",
				parentObjName, progress.Recalculated, progress.NextAfterID,
			))
			break
		}
	}

	if progress.Complete {
		// Never hand back a cursor for a scan that has nothing left. Complete is
		// the signal a driver must branch on, but a driver that instead loops
		// while the cursor is non-empty has to terminate too, or the endpoint
		// gets hammered forever by a client that believes it is making progress.
		progress.NextAfterID = ""
	}

	rs.logger.Info(fmt.Sprintf(
		"RecalculateAll: Completed %d records for %s (complete=%t)",
		progress.Recalculated, parentObjName, progress.Complete,
	))
	return progress, nil
}

// recalculateBoundReached reports whether this pass must stop before starting
// another parent. It is consulted only after a parent's transaction commits, so
// a pass overruns its budget by at most one parent's cost — the budget governs
// whether to start more work, not when the call returns.
func (rs *RollupService) recalculateBoundReached(bounds ports.RollupRecalculateBounds, recalculated int) bool {
	if bounds.MaxParents > 0 && recalculated >= bounds.MaxParents {
		return true
	}
	if !bounds.Deadline.IsZero() && !rs.nowTime().Before(bounds.Deadline) {
		return true
	}
	return false
}