package records

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"sort"
	"strconv"
	"strings"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	rollupDeltaStatusPending     = "pending"
	rollupDeltaStatusProcessed   = "processed"
	rollupDeltaStatusFailed      = "failed"
	defaultRollupDeltaBatch      = 500
	defaultRollupDeltaRetries    = 3
	rollupDeltaWorkerTickTimeout = 5 * time.Minute
)

func rollupDeltaBindingPeekSQL(excludeClause string) string {
	// NULL is the historical representation of immediately available work;
	// retries carry an explicit NextAttemptAt. Take one index-ordered head from
	// each range, then compare only those two candidates. This preserves fair
	// availability order without scanning every due row in the Runtime scope.
	return fmt.Sprintf(`
SELECT Id
FROM (
  (SELECT Id, CreatedDate AS WakeAt, CreatedDate
    FROM _System_RollupDelta USE INDEX (idx_rollup_delta_scope_claim)
    WHERE ScopeDigest = ? AND Status = ? AND NextAttemptAt IS NULL
%s
    ORDER BY CreatedDate ASC, Id ASC
    LIMIT 1)
  UNION ALL
  (SELECT Id, NextAttemptAt AS WakeAt, CreatedDate
     FROM _System_RollupDelta USE INDEX (idx_rollup_delta_scope_claim)
    WHERE ScopeDigest = ? AND Status = ? AND NextAttemptAt IS NOT NULL
      AND NextAttemptAt <= NOW()
%s
    ORDER BY NextAttemptAt ASC, CreatedDate ASC, Id ASC
    LIMIT 1)
) AS due
ORDER BY WakeAt ASC, CreatedDate ASC, Id ASC
LIMIT 1`, excludeClause, excludeClause)
}

const rollupDeltaBindingLoadSQL = `
SELECT COALESCE(OrgId, ''), COALESCE(AppInstallationId, ''), COALESCE(RuntimeCapabilityId, ''),
       COALESCE(CAST(CatalogRevision AS CHAR), ''), COALESCE(CAST(SchemaGeneration AS CHAR), ''),
       COALESCE(CAST(EngineGeneration AS CHAR), ''), COALESCE(CatalogSnapshotDigest, ''),
       COALESCE(RuntimeMetadataViewDigest, '')
  FROM _System_RollupDelta
 WHERE Id = ? AND ScopeDigest = ?
 LIMIT 1`

type countRollupReconciler func(ctx context.Context, parentObjectName, parentRecordID string, rollupFieldNames []string) (models.SObject, error)

type rollupDeltaParentLane struct {
	parentObject   string
	parentRecordID string
}

// EnqueueDerivedCountRollupDeltas records insert-only COUNT dirty markers in
// the caller's transaction. The drain worker later serializes by parent and
// reconciles affected fields from live child rows outside the child insert hot
// path.
func (ps *PersistenceService) EnqueueDerivedCountRollupDeltas(ctx context.Context, sourceObjectName, sourceRecordID, parentObjectName, parentRecordID string, deltas map[string]int64) error {
	if strings.TrimSpace(sourceObjectName) == "" || strings.TrimSpace(sourceRecordID) == "" || strings.TrimSpace(parentObjectName) == "" || strings.TrimSpace(parentRecordID) == "" {
		return appErrors.NewInternalError("derived rollup delta enqueue missing source or parent identity", nil)
	}
	if err := ps.validateDerivedCountRollupDeltas(ctx, parentObjectName, deltas); err != nil {
		return err
	}
	canApply, err := ps.CanApplyDerivedCountRollupDeltas(ctx, parentObjectName, derivedCountRollupDeltaFieldNames(deltas)...)
	if err != nil {
		return err
	}
	if !canApply {
		return appErrors.NewInternalError(fmt.Sprintf("derived rollup delta enqueue rejected unsafe parent lifecycle for %s", parentObjectName), nil)
	}

	exec, err := ps.rollupDeltaExecutor(ctx)
	if err != nil {
		return err
	}
	binding, bound := ps.runtimeRollupDeltaBinding(ctx)
	if ps.metadata != nil && ps.metadata.UsesRequestBoundRuntimeView() && !bound {
		return appErrors.NewInternalError("rollup delta enqueue requires the request-bound Runtime metadata view", nil)
	}
	var scopeDigest string
	if bound {
		scopeDigest, err = rollupDeltaScopeDigest(binding)
		if err != nil {
			return appErrors.NewInternalError("failed to derive rollup delta Runtime scope", err)
		}
	}
	for fieldName, delta := range deltas {
		if delta == 0 {
			continue
		}
		var definitionBinding *RollupDeltaDefinitionBinding
		if bound {
			definitionBinding = &binding
		}
		dedupeKey, err := rollupDeltaDedupeKey(sourceObjectName, sourceRecordID, parentObjectName, parentRecordID, fieldName, definitionBinding)
		if err != nil {
			return appErrors.NewInternalError("failed to derive revision-bound rollup delta identity", err)
		}
		id := "rld-" + dedupeKey
		if bound {
			if _, err := exec.ExecContext(ctx, `
INSERT IGNORE INTO _System_RollupDelta
  (Id, SourceObject, SourceRecordId, ParentObject, ParentRecordId, RollupField, Delta, DedupeKey,
   ScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CatalogRevision, SchemaGeneration, EngineGeneration,
   CatalogSnapshotDigest, RuntimeMetadataViewDigest, Status, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', NOW(), NOW())`,
				id, sourceObjectName, sourceRecordID, parentObjectName, parentRecordID, fieldName, delta, dedupeKey,
				scopeDigest, binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID, binding.CatalogRevision,
				binding.SchemaGeneration, binding.EngineGeneration, binding.CatalogSnapshotDigest, binding.RuntimeMetadataViewDigest,
			); err != nil {
				return appErrors.NewInternalError("failed to enqueue revision-bound derived rollup delta", err)
			}
			continue
		}
		if _, err := exec.ExecContext(ctx, `
INSERT IGNORE INTO _System_RollupDelta
  (Id, SourceObject, SourceRecordId, ParentObject, ParentRecordId, RollupField, Delta, DedupeKey, Status, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', NOW(), NOW())`,
			id, sourceObjectName, sourceRecordID, parentObjectName, parentRecordID, fieldName, delta, dedupeKey,
		); err != nil {
			return appErrors.NewInternalError("failed to enqueue derived rollup delta", err)
		}
	}
	return nil
}

func rollupDeltaDedupeKey(sourceObjectName, sourceRecordID, parentObjectName, parentRecordID, rollupField string, binding *RollupDeltaDefinitionBinding) (string, error) {
	parts := []string{sourceObjectName, sourceRecordID, parentObjectName, parentRecordID, rollupField, "insert"}
	if binding != nil {
		// A durable work item is not just a business-row identity. The immutable
		// metadata view is part of the obligation: a marker produced under a new
		// Catalog revision must coexist with stale or poison work from an older
		// revision instead of being discarded by INSERT IGNORE. Reuse Catalog's
		// canonical binding encoding so queue identity cannot omit one lineage
		// component as the shared binding evolves.
		encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(*binding)
		if err != nil {
			return "", err
		}
		parts = append(parts, catalogsnapshot.RuntimeWorkBindingSchemaV1, encoded)
	}
	h := sha256.New()
	for _, part := range parts {
		fmt.Fprintf(h, "%d:%s|", len(part), part)
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func rollupDeltaScopeDigest(binding RollupDeltaDefinitionBinding) (string, error) {
	return (persistence.RuntimeWorkScope{
		OrgID: binding.OrgID, AppInstallationID: binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	}).Digest()
}

func (ps *PersistenceService) rollupDeltaExecutor(ctx context.Context) (persistence.Executor, error) {
	if ps == nil {
		return nil, appErrors.NewInternalError("persistence service not wired", nil)
	}
	if ps.txManager != nil {
		if tx, ok := ps.txManager.ExtractTx(ctx).(*sql.Tx); ok && tx != nil {
			return tx, nil
		}
	}
	if ps.rawDB != nil {
		return ps.rawDB, nil
	}
	return nil, appErrors.NewInternalError("rollup delta queue requires raw DB or active transaction", nil)
}

func (ps *PersistenceService) peekRollupDeltaBinding(ctx context.Context, exclude []rollupDeltaParentLane) (string, RollupDeltaDefinitionBinding, bool, error) {
	worker := ps.rollupDeltaWorker
	if worker == nil || worker.db == nil {
		return "", RollupDeltaDefinitionBinding{}, false, appErrors.NewInternalError("rollup delta worker database is unavailable", nil)
	}
	scopeDigest, err := worker.scope.Digest()
	if err != nil {
		return "", RollupDeltaDefinitionBinding{}, false, err
	}
	baseArgs := []any{scopeDigest, rollupDeltaStatusPending}
	excludeArgs := make([]any, 0, len(exclude)*2)
	excludeClause := ""
	if len(exclude) != 0 {
		parts := make([]string, 0, len(exclude))
		for _, lane := range exclude {
			if lane.parentObject == "" || lane.parentRecordID == "" {
				continue
			}
			parts = append(parts, "(ParentObject = ? AND ParentRecordId = ?)")
			excludeArgs = append(excludeArgs, lane.parentObject, lane.parentRecordID)
		}
		if len(parts) != 0 {
			excludeClause = "   AND NOT (" + strings.Join(parts, " OR ") + ")"
		}
	}
	var binding RollupDeltaDefinitionBinding
	var rowID string
	var catalogRevision, schemaGeneration, engineGeneration, snapshotDigest, viewDigest string
	args := append(append(append([]any{}, baseArgs...), excludeArgs...), baseArgs...)
	args = append(args, excludeArgs...)
	err = worker.db.QueryRowContext(ctx, rollupDeltaBindingPeekSQL(excludeClause), args...).Scan(&rowID)
	if err == sql.ErrNoRows {
		return "", RollupDeltaDefinitionBinding{}, false, nil
	}
	if err != nil {
		return "", RollupDeltaDefinitionBinding{}, false, appErrors.NewInternalError("failed to select rollup delta definition binding", err)
	}
	err = worker.db.QueryRowContext(ctx, rollupDeltaBindingLoadSQL, rowID, scopeDigest).Scan(
		&binding.OrgID, &binding.AppInstallationID, &binding.RuntimeCapabilityID,
		&catalogRevision, &schemaGeneration, &engineGeneration, &snapshotDigest, &viewDigest,
	)
	if err == sql.ErrNoRows {
		return rowID, RollupDeltaDefinitionBinding{}, false, nil
	}
	if err != nil {
		return "", RollupDeltaDefinitionBinding{}, false, appErrors.NewInternalError("failed to read rollup delta definition binding", err)
	}
	for name, raw := range map[string]string{
		"catalog revision": catalogRevision, "schema generation": schemaGeneration, "engine generation": engineGeneration,
	} {
		value, parseErr := strconv.ParseUint(raw, 10, 64)
		if parseErr != nil || value == 0 {
			return rowID, RollupDeltaDefinitionBinding{}, false, appErrors.NewInternalError("rollup delta has no valid "+name, parseErr)
		}
		switch name {
		case "catalog revision":
			binding.CatalogRevision = value
		case "schema generation":
			binding.SchemaGeneration = value
		case "engine generation":
			binding.EngineGeneration = value
		}
	}
	binding.CatalogSnapshotDigest = catalogsnapshot.Digest(snapshotDigest)
	binding.RuntimeMetadataViewDigest = catalogsnapshot.Digest(viewDigest)
	if err := binding.Validate(); err != nil {
		return rowID, binding, false, err
	}
	if binding.OrgID != worker.scope.OrgID ||
		binding.AppInstallationID != worker.scope.AppInstallationID ||
		binding.RuntimeCapabilityID != worker.scope.RuntimeCapabilityID {
		return rowID, binding, false, appErrors.NewInternalError("rollup delta scope digest disagrees with explicit Runtime scope", nil)
	}
	return rowID, binding, true, nil
}

func (ps *PersistenceService) markMalformedRollupDeltaDefinition(ctx context.Context, rowID string, cause error) error {
	if ps == nil || ps.rollupDeltaWorker == nil || ps.txManager == nil || rowID == "" {
		return appErrors.NewInternalError("malformed rollup definition requires exact fenced work", cause)
	}
	worker := ps.rollupDeltaWorker
	return ps.txManager.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		tx, ok := ps.txManager.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return appErrors.NewInternalError("malformed rollup definition transaction is unavailable", nil)
		}
		if err := worker.ownership.LockActive(txCtx, tx); err != nil {
			return err
		}
		var lockedID string
		err := tx.QueryRowContext(txCtx, `SELECT Id FROM _System_RollupDelta
WHERE Id = ? AND Status = 'pending'
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`, rowID, worker.scope.OrgID, worker.scope.AppInstallationID, worker.scope.RuntimeCapabilityID).Scan(&lockedID)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return appErrors.NewInternalError("failed to lock malformed rollup definition", err)
		}
		_, err = ps.markRollupDeltaFailure(txCtx, tx, []string{lockedID}, cause)
		return err
	})
}

func (ps *PersistenceService) markRollupDeltaDefinitionFailure(ctx context.Context, rowID string, binding RollupDeltaDefinitionBinding, cause error) error {
	if ps == nil || ps.rollupDeltaWorker == nil || ps.txManager == nil || rowID == "" {
		return appErrors.NewInternalError("rollup delta definition failure requires exact fenced work", cause)
	}
	worker := ps.rollupDeltaWorker
	return ps.txManager.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		tx, ok := ps.txManager.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return appErrors.NewInternalError("rollup delta definition failure transaction is unavailable", nil)
		}
		if err := worker.ownership.LockActive(txCtx, tx); err != nil {
			return err
		}
		var lockedID string
		err := tx.QueryRowContext(txCtx, `SELECT Id FROM _System_RollupDelta
WHERE Id = ? AND Status = 'pending'
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND CatalogRevision = ? AND SchemaGeneration = ? AND EngineGeneration = ?
  AND CatalogSnapshotDigest = ? AND RuntimeMetadataViewDigest = ?
FOR UPDATE`, rowID, binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID,
			binding.CatalogRevision, binding.SchemaGeneration, binding.EngineGeneration,
			binding.CatalogSnapshotDigest, binding.RuntimeMetadataViewDigest).Scan(&lockedID)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return appErrors.NewInternalError("failed to lock exact rollup definition failure", err)
		}
		_, err = ps.markRollupDeltaFailure(txCtx, tx, []string{lockedID}, cause)
		return err
	})
}

func rollupDeltaBindingPredicate(binding *RollupDeltaDefinitionBinding) string {
	if binding == nil {
		return ""
	}
	return `
   AND ScopeDigest = ?
   AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
   AND CatalogRevision = ? AND SchemaGeneration = ? AND EngineGeneration = ?
   AND CatalogSnapshotDigest = ? AND RuntimeMetadataViewDigest = ?`
}

func rollupDeltaLaneArgs(binding *RollupDeltaDefinitionBinding, parentObject, parentRecordID string, batchSize int) ([]any, error) {
	args := []any{rollupDeltaStatusPending, parentObject, parentRecordID}
	if binding != nil {
		scopeDigest, err := rollupDeltaScopeDigest(*binding)
		if err != nil {
			return nil, err
		}
		args = append(args,
			scopeDigest,
			binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID,
			binding.CatalogRevision, binding.SchemaGeneration, binding.EngineGeneration,
			binding.CatalogSnapshotDigest, binding.RuntimeMetadataViewDigest,
		)
	}
	return append(args, batchSize), nil
}

func rollupDeltaLaneLoadSQL(binding *RollupDeltaDefinitionBinding) string {
	indexHint := "USE INDEX (idx_rollup_delta_parent_lane)"
	if binding != nil {
		// A Runtime worker must stay inside its installation scope while it
		// loads the selected parent lane. The global parent-lane index can
		// otherwise scan and lock rows from other installations that happen to
		// use the same parent identity.
		indexHint = "USE INDEX (idx_rollup_delta_scope_parent_lane)"
	}
	return fmt.Sprintf(`
SELECT Id, RollupField
  FROM _System_RollupDelta %s
 WHERE Status = ?
   AND (NextAttemptAt IS NULL OR NextAttemptAt <= NOW())
   AND ParentObject = ?
   AND ParentRecordId = ?
%s
 ORDER BY CreatedDate ASC, Id ASC
 LIMIT ?
 FOR UPDATE`, indexHint, rollupDeltaBindingPredicate(binding))
}

// ProcessRollupDeltaBatch drains one parent lane. The method returns the
// number of queue rows handled, not the number of parent updates.
func (ps *PersistenceService) ProcessRollupDeltaBatch(ctx context.Context, batchSize int, reconcile countRollupReconciler) (int, error) {
	processed, _, err := ps.processRollupDeltaBatch(ctx, batchSize, reconcile, nil)
	return processed, err
}

func (ps *PersistenceService) processRollupDeltaBatch(ctx context.Context, batchSize int, reconcile countRollupReconciler, exclude []rollupDeltaParentLane) (int, rollupDeltaParentLane, error) {
	if ps == nil || ps.txManager == nil {
		return 0, rollupDeltaParentLane{}, appErrors.NewInternalError("rollup delta worker requires transaction manager", nil)
	}
	if reconcile == nil {
		return 0, rollupDeltaParentLane{}, appErrors.NewInternalError("rollup delta worker requires count reconciler", nil)
	}
	if batchSize <= 0 {
		batchSize = defaultRollupDeltaBatch
	}
	executionCtx := ctx
	var requiredBinding *RollupDeltaDefinitionBinding
	var candidateRowID string
	if worker := ps.rollupDeltaWorker; worker != nil {
		if err := worker.ownership.VerifyActive(ctx); err != nil {
			return 0, rollupDeltaParentLane{}, err
		}
		rowID, binding, found, err := ps.peekRollupDeltaBinding(ctx, exclude)
		if err != nil {
			if rowID != "" {
				if markErr := ps.markMalformedRollupDeltaDefinition(ctx, rowID, err); markErr != nil {
					return 0, rollupDeltaParentLane{}, markErr
				}
			}
			return 0, rollupDeltaParentLane{}, err
		}
		if !found {
			return 0, rollupDeltaParentLane{}, nil
		}
		executionCtx, err = worker.binder(ctx, binding)
		if err != nil {
			cause := appErrors.NewInternalError("rollup delta definition binding rejected", err)
			if markErr := ps.markRollupDeltaDefinitionFailure(ctx, rowID, binding, cause); markErr != nil {
				return 0, rollupDeltaParentLane{}, markErr
			}
			return 0, rollupDeltaParentLane{}, cause
		}
		requiredBinding = &binding
		candidateRowID = rowID
	}
	var processed int
	var lane rollupDeltaParentLane
	err := ps.txManager.WithPessimisticRetry(executionCtx, func(txCtx context.Context) error {
		n, claimed, err := ps.processRollupDeltaBatchTx(txCtx, batchSize, reconcile, exclude, candidateRowID, requiredBinding)
		processed = n
		lane = claimed
		return err
	}, defaultRollupDeltaRetries)
	observeRollupDeltaParentLaneRows(lane, processed, err)
	return processed, lane, err
}

func (ps *PersistenceService) processRollupDeltaBatchTx(ctx context.Context, batchSize int, reconcile countRollupReconciler, exclude []rollupDeltaParentLane, candidateRowID string, requiredBinding *RollupDeltaDefinitionBinding) (int, rollupDeltaParentLane, error) {
	exec, err := ps.rollupDeltaExecutor(ctx)
	if err != nil {
		return 0, rollupDeltaParentLane{}, err
	}

	if worker := ps.rollupDeltaWorker; worker != nil {
		tx, ok := ps.txManager.ExtractTx(ctx).(*sql.Tx)
		if !ok || tx == nil {
			return 0, rollupDeltaParentLane{}, appErrors.NewInternalError("rollup delta ownership fence requires the active transaction", nil)
		}
		if err := worker.ownership.LockActive(ctx, tx); err != nil {
			return 0, rollupDeltaParentLane{}, err
		}
	}

	var parentObject, parentRecordID string
	claimQuery := `
SELECT ParentObject, ParentRecordId
  FROM _System_RollupDelta
 WHERE Status = ?
   AND (NextAttemptAt IS NULL OR NextAttemptAt <= NOW())
%s
 ORDER BY CreatedDate ASC, Id ASC
 LIMIT 1
 FOR UPDATE`
	claimArgs := []any{rollupDeltaStatusPending}
	excludeClause := ""
	if requiredBinding != nil {
		scopeDigest, digestErr := rollupDeltaScopeDigest(*requiredBinding)
		if digestErr != nil {
			return 0, rollupDeltaParentLane{}, digestErr
		}
		claimQuery = `
SELECT ParentObject, ParentRecordId
  FROM _System_RollupDelta
 WHERE Id = ? AND ScopeDigest = ? AND Status = ?
	   AND (NextAttemptAt IS NULL OR NextAttemptAt <= NOW())
	   AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
	   AND CatalogRevision = ? AND SchemaGeneration = ? AND EngineGeneration = ?
	   AND CatalogSnapshotDigest = ? AND RuntimeMetadataViewDigest = ?
 LIMIT 1
 FOR UPDATE`
		claimArgs = []any{
			candidateRowID, scopeDigest, rollupDeltaStatusPending,
			requiredBinding.OrgID, requiredBinding.AppInstallationID, requiredBinding.RuntimeCapabilityID,
			requiredBinding.CatalogRevision, requiredBinding.SchemaGeneration, requiredBinding.EngineGeneration,
			requiredBinding.CatalogSnapshotDigest, requiredBinding.RuntimeMetadataViewDigest,
		}
	}
	if requiredBinding == nil && len(exclude) > 0 {
		parts := make([]string, 0, len(exclude))
		for _, lane := range exclude {
			if strings.TrimSpace(lane.parentObject) == "" || strings.TrimSpace(lane.parentRecordID) == "" {
				continue
			}
			parts = append(parts, "(ParentObject = ? AND ParentRecordId = ?)")
			claimArgs = append(claimArgs, lane.parentObject, lane.parentRecordID)
		}
		if len(parts) > 0 {
			excludeClause += "\n   AND NOT (" + strings.Join(parts, " OR ") + ")"
		}
	}
	if requiredBinding == nil {
		claimQuery = fmt.Sprintf(claimQuery, excludeClause)
	}
	err = exec.QueryRowContext(ctx, claimQuery, claimArgs...).Scan(&parentObject, &parentRecordID)
	lane := rollupDeltaParentLane{parentObject: parentObject, parentRecordID: parentRecordID}
	if err == sql.ErrNoRows {
		return 0, rollupDeltaParentLane{}, nil
	}
	if err != nil {
		return 0, rollupDeltaParentLane{}, appErrors.NewInternalError("failed to claim rollup delta parent lane", err)
	}

	laneArgs, err := rollupDeltaLaneArgs(requiredBinding, parentObject, parentRecordID, batchSize)
	if err != nil {
		return 0, lane, appErrors.NewInternalError("failed to derive rollup delta lane scope", err)
	}
	rows, err := exec.QueryContext(ctx, rollupDeltaLaneLoadSQL(requiredBinding), laneArgs...)
	if err != nil {
		return 0, lane, appErrors.NewInternalError("failed to load rollup delta lane", err)
	}
	defer rows.Close()

	var ids []string
	fieldSet := map[string]struct{}{}
	for rows.Next() {
		var id string
		var fieldName string
		if err := rows.Scan(&id, &fieldName); err != nil {
			return 0, lane, appErrors.NewInternalError("failed to scan rollup delta row", err)
		}
		ids = append(ids, id)
		fieldSet[fieldName] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return 0, lane, appErrors.NewInternalError("failed to iterate rollup delta lane", err)
	}
	if len(ids) == 0 {
		return 0, lane, nil
	}
	fieldNames := make([]string, 0, len(fieldSet))
	for fieldName := range fieldSet {
		fieldNames = append(fieldNames, fieldName)
	}
	sort.Strings(fieldNames)

	canApplyDirect, err := ps.CanApplyDerivedCountRollupDeltas(ctx, parentObject, fieldNames...)
	if err != nil {
		processed, err := ps.markRollupDeltaFailure(ctx, exec, ids, err)
		return processed, lane, err
	}
	if canApplyDirect && ps.metadata != nil && ps.metadata.HasRollupChildDependencies(ctx, parentObject) {
		canApplyDirect = false
	}

	systemUser, err := ps.LoadSystemUserSession(ctx)
	if err != nil {
		processed, err := ps.markRollupDeltaFailure(ctx, exec, ids, appErrors.NewInternalError("failed to load rollup system user", err))
		return processed, lane, err
	}
	if systemUser == nil {
		processed, err := ps.markRollupDeltaFailure(ctx, exec, ids, appErrors.NewInternalError("rollup system user not found", nil))
		return processed, lane, err
	}
	if canApplyDirect {
		rollupDeltaParentLockAttempts.WithLabelValues("direct", "attempt").Inc()
		if err := ps.LockDerivedRecord(ctx, parentObject, parentRecordID); err != nil {
			rollupDeltaParentLockAttempts.WithLabelValues("direct", "error").Inc()
			processed, err := ps.markRollupDeltaFailure(ctx, exec, ids, err)
			return processed, lane, err
		}
		rollupDeltaParentLockAttempts.WithLabelValues("direct", "success").Inc()
		updates, reconcileErr := reconcile(ctx, parentObject, parentRecordID, fieldNames)
		if reconcileErr != nil {
			processed, markErr := ps.markRollupDeltaFailure(ctx, exec, ids, reconcileErr)
			return processed, lane, markErr
		}
		err = ps.applyDerivedCountRollupValuesDirect(ctx, parentObject, parentRecordID, updates, systemUser)
	} else {
		rollupDeltaParentLockAttempts.WithLabelValues("metadata_update", "attempt").Inc()
		err = ps.UpdateDerivedCountRollupValues(ctx, parentObject, parentRecordID, fieldNames, systemUser, func(lockedCtx context.Context) (models.SObject, error) {
			return reconcile(lockedCtx, parentObject, parentRecordID, fieldNames)
		})
		if err != nil {
			rollupDeltaParentLockAttempts.WithLabelValues("metadata_update", "error").Inc()
		} else {
			rollupDeltaParentLockAttempts.WithLabelValues("metadata_update", "success").Inc()
		}
	}
	if err != nil {
		processed, err := ps.markRollupDeltaFailure(ctx, exec, ids, err)
		return processed, lane, err
	}

	placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",")
	args := make([]any, 0, 2+len(ids))
	args = append(args, rollupDeltaStatusProcessed, time.Now().UTC())
	for _, id := range ids {
		args = append(args, id)
	}
	if ps.rollupDeltaWorker != nil {
		// The business effect and acknowledgement share this transaction. A
		// successful independently-owned worker therefore retains no unbounded
		// payload history; poison rows remain explicit until operator repair.
		if _, err := exec.ExecContext(ctx, fmt.Sprintf(`DELETE FROM _System_RollupDelta WHERE Id IN (%s)`, placeholders), args[2:]...); err != nil {
			return 0, lane, appErrors.NewInternalError("failed to acknowledge completed Runtime rollup deltas", err)
		}
	} else if _, err := exec.ExecContext(ctx, fmt.Sprintf(`
UPDATE _System_RollupDelta
   SET Status = ?, ProcessedAt = ?, NextAttemptAt = NULL, LastModifiedDate = NOW()
 WHERE Id IN (%s)`, placeholders), args...); err != nil {
		return 0, lane, appErrors.NewInternalError("failed to mark rollup deltas processed", err)
	}
	return len(ids), lane, nil
}

func (ps *PersistenceService) markRollupDeltaFailure(ctx context.Context, exec persistence.Executor, ids []string, cause error) (int, error) {
	if len(ids) == 0 {
		return 0, cause
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",")
	args := make([]any, 0, 5+len(ids))
	args = append(args, truncateRollupDeltaError(cause), time.Now().UTC())
	statusUpdate := ""
	if worker := ps.rollupDeltaWorker; worker != nil {
		statusUpdate = `
       Status = CASE WHEN RetryCount + 1 >= ? THEN 'failed' ELSE Status END,
       NextAttemptAt = CASE WHEN RetryCount + 1 >= ? THEN NULL ELSE DATE_ADD(NOW(), INTERVAL ? SECOND) END,`
		args = append(args, worker.max, worker.max, rollupDeltaRetryDelay())
	} else {
		statusUpdate = `
       NextAttemptAt = DATE_ADD(NOW(), INTERVAL ? SECOND),`
		args = append(args, rollupDeltaRetryDelay())
	}
	for _, id := range ids {
		args = append(args, id)
	}
	if _, err := exec.ExecContext(ctx, fmt.Sprintf(`
UPDATE _System_RollupDelta
   SET RetryCount = RetryCount + 1,
       LastError = ?,
       LastErrorAt = ?,
%s
       LastModifiedDate = NOW()
 WHERE Id IN (%s)`, statusUpdate, placeholders), args...); err != nil {
		return 0, appErrors.NewInternalError("failed to record rollup delta failure", err)
	}
	return 0, nil
}

func rollupDeltaRetryDelay() int {
	return 1
}

func truncateRollupDeltaError(err error) string {
	if err == nil {
		return ""
	}
	msg := err.Error()
	if len(msg) > 4000 {
		return msg[:4000]
	}
	return msg
}

func (ps *PersistenceService) drainRollupDeltaTick(ctx context.Context, batchSize int, reconcile countRollupReconciler) {
	started := time.Now()
	result := "empty"
	defer func() {
		rollupDeltaWorkerDrainDurationSeconds.WithLabelValues(result).Observe(time.Since(started).Seconds())
	}()
	visited := make([]rollupDeltaParentLane, 0)
	for {
		select {
		case <-ctx.Done():
			result = "canceled"
			return
		default:
		}
		processed, lane, err := ps.processRollupDeltaBatch(ctx, batchSize, reconcile, visited)
		if lane.parentObject != "" && lane.parentRecordID != "" {
			visited = append(visited, lane)
		}
		if err != nil {
			if lane.parentObject != "" && lane.parentRecordID != "" {
				log.Printf("rollup delta worker error parent_lane_hash=%s: %v", lane.safeHash(), err)
			} else {
				log.Printf("rollup delta worker error: %v", err)
			}
			result = "error"
			return
		}
		if processed > 0 {
			result = "processed"
		}
		if processed == 0 && (lane.parentObject == "" || lane.parentRecordID == "") {
			return
		}
	}
}

func observeRollupDeltaParentLaneRows(lane rollupDeltaParentLane, processed int, err error) {
	if lane.parentObject == "" || processed <= 0 {
		return
	}
	result := "processed"
	if err != nil {
		result = "failed"
	}
	rollupDeltaParentLaneRowsProcessed.WithLabelValues(result).Add(float64(processed))
}

func (lane rollupDeltaParentLane) safeHash() string {
	if lane.parentObject == "" || lane.parentRecordID == "" {
		return ""
	}
	h := sha256.New()
	fmt.Fprintf(h, "rollup-delta-parent-lane\x00%s\x00%s", lane.parentObject, lane.parentRecordID)
	sum := hex.EncodeToString(h.Sum(nil))
	if len(sum) > 16 {
		return sum[:16]
	}
	return sum
}

func (ps *PersistenceService) StartRollupDeltaWorker(interval time.Duration, batchSize int, reconcile countRollupReconciler) (stop func()) {
	if ps == nil {
		return func() {}
	}
	if interval <= 0 {
		interval = 100 * time.Millisecond
	}
	stopCh := make(chan struct{})
	doneCh := make(chan struct{})
	go func() {
		defer close(doneCh)
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-stopCh:
				return
			case <-ticker.C:
				tickCtx, cancel := rollupDeltaWorkerTickContext()
				go func() {
					select {
					case <-stopCh:
						cancel()
					case <-tickCtx.Done():
					}
				}()
				ps.drainRollupDeltaTick(tickCtx, batchSize, reconcile)
				cancel()
			}
		}
	}()
	return func() {
		close(stopCh)
		<-doneCh
	}
}

func rollupDeltaWorkerTickContext() (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.Background(), rollupDeltaWorkerTickTimeout)
}