package automation

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"

	"github.com/nexuscrm/backend/internal/domain/ports"
	"github.com/nexuscrm/backend/internal/features/audit"
	"github.com/nexuscrm/backend/internal/features/metadata"
	apperrors "github.com/nexuscrm/backend/pkg/errors"
	"github.com/nexuscrm/backend/pkg/formula"
	"github.com/nexuscrm/backend/pkg/logging"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// identifierRegex validates API names and IDs to prevent SQL injection
// Allows alphanumeric, underscores, and hyphens (for UUIDs)
var identifierRegex = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)

// flowInterviewPersistModeKey is a private context key used to opt a flow
// execution in or out of FlowInterview persistence. The default (absent key)
// is persist=true so REST-invoked, scheduled, and subflow-invoked runs keep
// writing audit rows for backward compatibility. Only the record-triggered
// hot path explicitly opts out via WithTransientFlowInterviews.
type flowInterviewPersistModeKey struct{}

// WithTransientFlowInterviews marks the given context so that any flow
// executed through ExecuteFlow / executeFlowVersion under it uses an
// in-memory transient FlowInterview and skips the two DB round-trips
// that would otherwise insert and update the row. Subflows inherit the
// mode automatically because they propagate the same context.
//
// The record-triggered listener wraps its context with this helper before
// firing RecordBeforeSave / RecordAfterSave / RecordBeforeDelete flows,
// which is where the observed ~160s-per-save latency was originating.
// Scheduled flows and REST-invoked flows intentionally do NOT call this —
// their audit rows are still useful for operator visibility.
func WithTransientFlowInterviews(ctx context.Context) context.Context {
	return context.WithValue(ctx, flowInterviewPersistModeKey{}, false)
}

// shouldPersistFlowInterview returns true when the current flow execution
// should write its FlowInterview row to the database (default), and false
// when the caller opted into transient execution via
// WithTransientFlowInterviews.
func shouldPersistFlowInterview(ctx context.Context) bool {
	if v, ok := ctx.Value(flowInterviewPersistModeKey{}).(bool); ok {
		return v
	}
	return true
}

// beforeSaveFlowKey is a private context key set by the record-triggered
// listener for RecordBeforeSave / RecordBeforeDelete events. It signals to
// executor_record_ops that implicit-$Record recordUpdate should mutate the
// in-flight record in memory (so the listener can merge changes back into
// the insert/update payload) rather than issuing a DB UPDATE for a row
// that either does not exist yet (BeforeCreate) or is in the middle of an
// open transaction (BeforeUpdate).
type beforeSaveFlowKey struct{}

// atomicFlowBoundaryKey marks execution that is currently running inside the
// Flow engine's RunAtomically wrapper. This is narrower than "a DB
// transaction exists on ctx": many non-action code paths already run inside a
// transaction, but only an explicit atomic-flow boundary needs the extra
// preflight rule that child flows must not contain Wait / emailAlert either.
//
// Inherited behavior matters because subflows use ExecuteFlow recursively. A
// non-atomic child invoked from an atomic parent still runs inside the same
// open transaction (PersistenceProvider.RunInTransaction reuses the outer tx),
// so allowing the child to pause on Wait or send emailAlert would still hold
// that atomic transaction across I/O / resume boundaries.
type atomicFlowBoundaryKey struct{}

// WithBeforeSaveFlow marks the context as executing under a RecordBeforeSave
// trigger. Only the FlowTriggerListener's before-save branch should call
// this helper. AfterSave, scheduled, and REST-invoked runs must leave the
// key absent so recordUpdate persists via DB UPDATE as normal.
func WithBeforeSaveFlow(ctx context.Context) context.Context {
	return context.WithValue(ctx, beforeSaveFlowKey{}, true)
}

// isBeforeSaveFlow reports whether the current execution was marked as
// a before-save record-triggered flow via WithBeforeSaveFlow.
func isBeforeSaveFlow(ctx context.Context) bool {
	v, _ := ctx.Value(beforeSaveFlowKey{}).(bool)
	return v
}

func withAtomicFlowBoundary(ctx context.Context) context.Context {
	return context.WithValue(ctx, atomicFlowBoundaryKey{}, true)
}

func isWithinAtomicFlowBoundary(ctx context.Context) bool {
	v, _ := ctx.Value(atomicFlowBoundaryKey{}).(bool)
	return v
}

const scheduledFlowQueryLimit = 1000

// statusCodeInvalidFlow is returned by executeFlowVersion when flow metadata
// fails pre-execution validation (see validateAtomicFlow).
const statusCodeInvalidFlow = "INVALID_FLOW"

// validateAtomicFlow rejects flow shapes incompatible with wrapping execution
// in a single DB transaction: Wait elements (would pause the flow with the
// tx still open) and emailAlert action calls (synchronous SMTP send inside
// the tx holds locks across external I/O). Non-atomic flows are unaffected.
func validateAtomicFlow(meta *models.FlowMetadata) error {
	return validateAtomicFlowForExecution(context.Background(), meta)
}

// validateAtomicFlowForExecution is the ctx-aware validator used at runtime.
// A flow is "effectively atomic" when either:
//   - its own metadata declares RunAtomically=true, or
//   - it is executing under an inherited atomic parent boundary.
//
// The second case closes the child-flow hole where a non-atomic subflow with
// Wait / emailAlert could still run inside the parent's open transaction.
func validateAtomicFlowForExecution(ctx context.Context, meta *models.FlowMetadata) error {
	if meta == nil {
		return nil
	}
	if !meta.RunAtomically {
		if !isWithinAtomicFlowBoundary(ctx) {
			return nil
		}
	}
	if len(meta.Waits) > 0 {
		return apperrors.NewValidationError("runAtomically",
			"flows executing atomically cannot contain Wait elements; a paused interview would hold the database transaction open")
	}
	for _, action := range meta.ActionCalls {
		if strings.EqualFold(action.ActionType, "emailAlert") {
			return apperrors.NewValidationError("runAtomically",
				"flows executing atomically cannot contain emailAlert action calls; SMTP send would hold the database transaction open across external I/O")
		}
	}
	return nil
}

// validateIdentifier checks if a string is a safe identifier (prevents SQL injection)
func validateIdentifier(id string) error {
	if id == "" {
		return apperrors.NewValidationError("identifier", "identifier cannot be empty")
	}
	if len(id) > 255 {
		return apperrors.NewValidationError("identifier", "identifier too long")
	}
	if !identifierRegex.MatchString(id) {
		return apperrors.NewValidationError("identifier", fmt.Sprintf("invalid identifier format: %s", id))
	}
	return nil
}

// FlowExecutor is the core flow execution engine.
// Method implementations are split across domain-specific files:
//   - executor_elements.go    — element processing (Assignment, Decision, Loop, Screen, Wait, etc.)
//   - executor_record_ops.go  — record CRUD operations (Create, Update, Delete, Get, Lookup)
//   - executor_actions.go     — action invocations and subflow execution
//   - executor_helpers.go     — variable resolution, condition evaluation, formula helpers
type FlowExecutor struct {
	metadata      *metadata.MetadataService
	actions       *ActionService
	interviews    *FlowInterviewService
	persistence   PersistenceProvider
	query         ports.QueryService
	fieldMappings FieldMappingApplier
	logger        *logging.Logger

	// approvals is injected post-construction via SetApprovalService to break
	// the mutual dependency with ApprovalService (which already holds a
	// FlowExecutor to run approval-triggered flows). Used by the Flow engine
	// to dispatch actionType="emailAlert" via the shared WorkflowAlert path.
	// May be nil in tests; callers must nil-guard.
	approvals *ApprovalService

	// flowVersionFn is a test seam that bypasses the FlowDefinition/Flow
	// query path when a parent flow invokes a subflow. Production code
	// leaves this nil and the real query-service lookup runs; unit tests
	// point it at an in-memory pack so they can exercise subflow
	// composition without a live DB.
	flowVersionFn func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error)

	// flowPackOwnedFn is a test seam for the L1 standard-action internal
	// access boundary. Production determines ownership from FlowDefinition +
	// Flow.Source; unit tests can override it without standing up those rows.
	flowPackOwnedFn func(ctx context.Context, apiName string, user *models.UserSession) (bool, error)

	// convertedLookupsFn is a test seam for executeStampConvertedAnchors.
	// Production code leaves this nil; unit tests override it to avoid
	// wiring a full MetadataService when exercising the action.
	convertedLookupsFn func(ctx context.Context, objectAPIName string) []models.FieldMetadata

	// fieldMetaLookup is a test seam used by resolveNestedReferenceCtx to
	// translate foreign-key fields to their referenced object type. Production
	// code leaves this nil and falls back to e.metadata.GetField; unit tests
	// override it to avoid wiring a full MetadataService.
	fieldMetaLookup func(objectAPIName, fieldAPIName string) *models.FieldMetadata

	// formulaEngine evaluates FlowFormula resources at interview init and on
	// every resume (so TODAY/NOW refresh). Initialized once per executor
	// (NewFlowExecutor) so the engine's compiled regex/function tables are
	// reused across all flow runs.
	formulaEngine *formula.Engine

	// fieldDefaultResolver applies stage 3 (ActionFieldDefault) of the
	// recordCreate default pipeline. Lazy-initialized in NewFlowExecutor
	// against formulaEngine. Tests may override this field directly to
	// inject a fake resolver without standing up an action-field-default
	// row set on ctx.
	fieldDefaultResolver FieldDefaultResolver
}

// lookupFieldMeta returns FieldMetadata for (object, field), honoring the
// test override when set. Centralizing the access here keeps the production
// path unchanged when the override is nil.
func (e *FlowExecutor) lookupFieldMeta(objectAPIName, fieldAPIName string) *models.FieldMetadata {
	if e.fieldMetaLookup != nil {
		return e.fieldMetaLookup(objectAPIName, fieldAPIName)
	}
	if e.metadata == nil {
		return nil
	}
	return e.metadata.GetField(objectAPIName, fieldAPIName)
}

// lookupObjectSchema returns the full ObjectMetadata for objectAPIName,
// nil-tolerant for test executors and cold-start cache misses. Centralized
// so all callers that need schema-aware FK resolution (buildFormulaFetcher,
// resolveNestedReferenceCtx, resolveNestedReferenceCtxWithPresence) consult
// the metadata service the same way (PR #2057 round-6 F1).
//
// Returning nil is safe: every downstream caller of the schema-aware
// resolver path falls back to convention-only Tier 2/3/4 when schema is
// nil, preserving pre-#2057 automation behavior for executors without a
// wired metadata service.
func (e *FlowExecutor) lookupObjectSchema(ctx context.Context, objectAPIName string) *models.ObjectMetadata {
	if e == nil || e.metadata == nil || objectAPIName == "" {
		return nil
	}
	return e.metadata.GetSchema(ctx, objectAPIName)
}

// NewFlowExecutor creates a new FlowExecutor. fieldMappings may be nil;
// when a RecordCreate element declares a FieldMapping ref the executor
// returns a fault without the applier, preventing silent omissions.
func NewFlowExecutor(
	metadataService *metadata.MetadataService,
	actions *ActionService,
	interviews *FlowInterviewService,
	persistence PersistenceProvider,
	query ports.QueryService,
	fieldMappings FieldMappingApplier,
	logger *logging.Logger,
) *FlowExecutor {
	if logger == nil {
		logger = logging.New(context.Background())
	}
	eng := formula.NewEngine()
	return &FlowExecutor{
		metadata:             metadataService,
		actions:              actions,
		interviews:           interviews,
		persistence:          persistence,
		query:                query,
		fieldMappings:        fieldMappings,
		logger:               logger,
		formulaEngine:        eng,
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(eng),
	}
}

// SetApprovalService wires an ApprovalService into the executor post-construction.
// Required for actionType="emailAlert" dispatch; other flow features work without it.
func (e *FlowExecutor) SetApprovalService(approvals *ApprovalService) {
	e.approvals = approvals
}

// runPossiblyAtomic invokes fn directly or wraps it in
// persistence.RunInTransaction. The persistence layer re-uses an outer
// transaction when ctx already carries one, so nested atomic subflows share
// the parent's rollback boundary instead of opening independent savepoints.
func (e *FlowExecutor) runPossiblyAtomic(ctx context.Context, atomic bool, fn func(context.Context) error) error {
	if !atomic {
		return fn(ctx)
	}
	return e.persistence.RunInTransaction(ctx, func(txCtx context.Context) error {
		return fn(withAtomicFlowBoundary(txCtx))
	})
}

// ExecuteFlow executes a flow by API name (Actions API entry point)
func (e *FlowExecutor) ExecuteFlow(ctx context.Context, flowApiName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	// Design §3.5: stamp Entrypoint=FLOW onto ctx so any record mutations
	// performed downstream (RecordCreate/Update/Delete elements) can recover
	// the correct AuditEvent.Entrypoint via audit.EntrypointFromContext(ctx).
	// The emitter still requires callers to set Entrypoint on the AuditEvent
	// struct explicitly (§3.1 CH-4) — this is transport only.
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)

	// Find the active flow version by API name
	flowVersion, err := e.getActiveFlowVersion(ctx, flowApiName, user)
	if err != nil {
		return nil, err
	}
	if flowVersion == nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("Flow '%s' not found or not active", flowApiName),
			}},
		}, nil
	}
	var result *models.FlowExecutionResult
	ctx, result = e.authorizeProtectedStandardFlow(ctx, flowApiName, flowVersion)
	if result != nil {
		return result, nil
	}

	return e.executeFlowVersion(ctx, flowVersion, flowApiName, inputs, user)
}

func (e *FlowExecutor) authorizeProtectedStandardFlow(ctx context.Context, flowApiName string, flowVersion *models.FlowVersion) (context.Context, *models.FlowExecutionResult) {
	if flowVersion == nil || flowVersion.DefinitionIsCustomized || flowVersion.Source != flowSourceStandard {
		return ctx, nil
	}
	if constants.HasStandardActionProtectedFlowExecution(ctx) {
		return ctx, nil
	}
	if constants.StandardActionExecutionOwnsRootFlow(ctx, flowApiName) {
		return constants.WithStandardActionProtectedFlowExecution(ctx, flowApiName), nil
	}
	return ctx, &models.FlowExecutionResult{
		ActionName: flowApiName,
		IsSuccess:  false,
		Errors: []models.FlowExecutionError{{
			StatusCode: "FLOW_EXECUTION_ERROR",
			Message:    fmt.Sprintf("Flow %q is protected and can only run through its standard action", flowApiName),
		}},
	}
}

func (e *FlowExecutor) executeFlowVersion(ctx context.Context, flowVersion *models.FlowVersion, flowApiName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	if flowVersion == nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("Flow '%s' not found or not active", flowApiName),
			}},
		}, nil
	}

	// Use hydrated metadata (HydrateMetadata is called by the data layer)
	flowMeta := flowVersion.Metadata

	if err := validateAtomicFlowForExecution(ctx, &flowMeta); err != nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: statusCodeInvalidFlow,
				Message:    err.Error(),
			}},
		}, nil
	}

	// Decide persistence mode once per run. The default (true) preserves
	// prior behaviour for REST-invoked, scheduled, and subflow-invoked
	// flows — all of which benefit from an audit trail. Record-triggered
	// flows opt out via WithTransientFlowInterviews in the listener,
	// avoiding 2 * N DB round-trips on the hot save path.
	persist := shouldPersistFlowInterview(ctx)

	// Create flow interview (persistent or transient per the mode above).
	var interview *models.FlowInterview
	var createErr error
	if persist {
		interview, createErr = e.interviews.CreateInterview(ctx, flowVersion.Id, flowApiName, user)
	} else {
		interview, createErr = e.interviews.CreateTransientInterview(flowVersion.Id, flowApiName, user)
	}
	if createErr != nil {
		return nil, apperrors.NewInternalError("failed to create flow interview", createErr)
	}

	// Initialize context with input variables
	interview.ContextData = initializeFlowContext(&flowMeta, inputs)

	// Inject $Profile / $User / $UserRole so Decision conditions that reference
	// e.g. $Profile.Name via resolveNestedReferenceCtx resolve correctly.
	// These are transient keys; marshalPersistableContext strips them on persist.
	injectUserGlobalsIntoContext(interview.ContextData, user)

	// Check entry conditions (for record-triggered flows). Use the Ctx
	// variant so Start filters like `WhatId StartsWith "006"` on
	// Salesforce-imported flows resolve against UUID IDs via sfKeyPrefixMatch.
	if !e.checkEntryConditionsCtx(ctx, &flowMeta, interview.ContextData) {
		// Record doesn't meet entry conditions - skip execution.
		if persist {
			if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
				e.logger.Warn("Failed to mark skipped interview as finished: " + completeErr.Error())
			}
		}
		return &models.FlowExecutionResult{
			ActionName:   flowApiName,
			IsSuccess:    true,
			OutputValues: map[string]interface{}{"_skipped": true},
		}, nil
	}

	// The interview row is created above, outside any transaction, so the
	// audit trail survives rollback of an atomic run.
	var outputs map[string]interface{}
	err := e.runPossiblyAtomic(ctx, flowMeta.RunAtomically, func(runCtx context.Context) error {
		var runErr error
		outputs, runErr = e.executeFlowMetadata(runCtx, interview, &flowMeta, user)
		return runErr
	})
	if err != nil {
		// Log the error with enough context to debug without the DB row.
		// For transient runs this is the only record of the failure.
		e.logger.Error(fmt.Sprintf("flow '%s' (version %s) execution failed: %s", flowApiName, flowVersion.Id, err.Error()), err)
		if persist {
			if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusError, user); completeErr != nil {
				e.logger.Error("Failed to mark interview as error", completeErr)
			}
		}

		// SF-native statusCode propagates through FlowCustomErrorFault when the
		// author declared one on the customError's customErrorMessage. Absent
		// → fall back to the generic "FLOW_ERROR" sentinel.
		statusCode := "FLOW_ERROR"
		message := err.Error()
		var customFault *FlowCustomErrorFault
		if errors.As(err, &customFault) {
			message = customFault.Message
			if customFault.StatusCode != "" {
				statusCode = customFault.StatusCode
			}
		}

		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: statusCode,
				Message:    message,
			}},
		}, nil
	}

	// If the flow paused on a Wait element, leave the interview in Paused
	// status for the scheduler to resume — do NOT overwrite with Finished.
	// Note: a transient interview that hit a Wait was promoted to a
	// persistent Paused row inside pauseForWait (IsTransient cleared).
	paused := interview.InterviewStatus == models.FlowInterviewStatusPaused

	// Mark interview as complete (persistent path only; transient runs
	// have no row to update; paused runs must be left for resume).
	if persist && !paused {
		if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
			e.logger.Error("Failed to mark interview as finished", completeErr)
		}
	}

	return &models.FlowExecutionResult{
		ActionName:   flowApiName,
		IsSuccess:    true,
		OutputValues: outputs,
	}, nil
}

func (e *FlowExecutor) buildScheduledRunInputs(ctx context.Context, flowVersion *models.FlowVersion, user *models.UserSession) ([]map[string]interface{}, error) {
	if flowVersion == nil || flowVersion.Metadata.Start == nil || flowVersion.Metadata.Start.Object == "" {
		return []map[string]interface{}{nil}, nil
	}
	if e.query == nil {
		return nil, apperrors.NewInternalError("scheduled flow query service unavailable", nil)
	}

	flowMeta := flowVersion.Metadata
	filterExpr := e.buildScheduledStartFilter(nil, flowMeta.Start.Filters, flowMeta.Start.FilterLogic)
	result, err := e.query.QueryWithFilter(ctx, flowMeta.Start.Object, filterExpr, user, "", "", scheduledFlowQueryLimit)
	if err != nil {
		return nil, err
	}

	inputs := make([]map[string]interface{}, 0, len(result.Records))
	objectName := flowMeta.Start.Object
	for _, record := range result.Records {
		// Stamp attributes.type so downstream RecordUpdate elements with
		// object=null can resolve the target object name.
		if attrs, ok := record["attributes"].(map[string]interface{}); ok {
			if _, hasType := attrs["type"]; !hasType {
				attrs["type"] = objectName
			}
		} else {
			record["attributes"] = map[string]interface{}{"type": objectName}
		}
		runInputs := map[string]interface{}{
			"$Record": record,
		}
		if e.checkEntryConditionsCtx(ctx, &flowMeta, runInputs) {
			inputs = append(inputs, runInputs)
		}
	}

	return inputs, nil
}

// ResumeInterview resumes a paused flow interview. On entry the interview
// is in Paused state with CurrentElement pointing at the Wait (or other
// pauseable) element. This method:
//
//  1. Loads the interview + FlowVersion (re-hydrates Metadata).
//  2. Re-evaluates any time-sensitive context so expressions that depend
//     on $Flow.CurrentDate / $Flow.CurrentDateTime (TODAY / NOW) pick up
//     the resume-time value, not the pause-time value.
//  3. Clears PausedUntil and flips InterviewStatus back to Running atomically.
//  4. Re-enters the executor loop at CurrentElement. On that re-entry the
//     Wait element's conditions are re-evaluated — so a catch-up resume
//     (PausedUntil already in the past at pause time) completes without
//     further delay.
//  5. If the resumed execution pauses again (e.g. a second Wait element),
//     leave the interview Paused; otherwise mark Finished.
//
// TODO(lane-a-formula): once the formula resource wiring lands (Phase 2.3),
// re-run formula.Engine.Evaluate here against the refreshed context so
// formulas referencing TODAY()/NOW() are recomputed at resume time rather
// than pause time.
func (e *FlowExecutor) ResumeInterview(ctx context.Context, interviewId string, user *models.UserSession) error {
	// Scheduled/paused flow resumptions run outside the original REST ctx;
	// stamp Entrypoint=FLOW so downstream record mutations produce correct
	// AuditEvent rows. See ExecuteFlow comment for the transport contract.
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)

	interview, err := e.interviews.GetInterview(ctx, interviewId, user)
	if err != nil {
		return err
	}
	if interview == nil {
		return apperrors.NewNotFoundError("flow interview", interviewId)
	}
	if interview.InterviewStatus != models.FlowInterviewStatusPaused {
		// Already claimed by another scheduler instance or manually resumed.
		return nil
	}

	// Atomic claim: use ClaimPausedInterview which does a conditional
	// UPDATE ... WHERE InterviewStatus = 'Paused'. Returns false if
	// another scheduler already claimed this interview.
	claimed, err := e.interviews.ClaimPausedInterview(ctx, interview.Id, user)
	if err != nil {
		return err
	}
	if !claimed {
		return nil // another scheduler won the race
	}

	// Get flow version to get metadata
	flowVersion, err := e.getFlowVersionById(ctx, interview.FlowVersionId, user)
	if err != nil {
		return err
	}
	if flowVersion == nil {
		return apperrors.NewNotFoundError("flow version", interview.FlowVersionId)
	}

	// Use hydrated metadata (HydrateMetadata is called by the data layer)
	flowMeta := flowVersion.Metadata

	// Refresh time-of-day context so any downstream Wait / condition check
	// that reads $Flow.CurrentDateTime or $Flow.CurrentDate sees resume
	// time rather than pause time.
	if interview.ContextData == nil {
		interview.ContextData = make(map[string]interface{})
	}
	now := time.Now()
	interview.ContextData["$Flow.CurrentDateTime"] = now.Format(time.RFC3339)
	interview.ContextData["$Flow.CurrentDate"] = now.Format("2006-01-02")

	// Re-inject user globals so resumed flows's Decision conditions still see
	// the current user's $Profile / $User / $UserRole. The keys are transient
	// and were stripped when the interview was persisted at pause time.
	injectUserGlobalsIntoContext(interview.ContextData, user)

	// ClaimPausedInterview already flipped status to Running.
	interview.InterviewStatus = models.FlowInterviewStatusRunning
	interview.PausedUntil = nil

	// Resume execution from current element.
	_, err = e.executeFlowMetadata(ctx, interview, &flowMeta, user)
	if err != nil {
		if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusError, user); completeErr != nil {
			e.logger.Error("Failed to mark interview as error during resume", completeErr)
		}
		return err
	}

	// If the resumed run paused again (another Wait element), leave it Paused.
	if interview.InterviewStatus == models.FlowInterviewStatusPaused {
		return nil
	}

	if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
		e.logger.Error("Failed to mark interview as finished during resume", completeErr)
	}
	return nil
}

// executeFlowMetadata executes flow from parsed metadata
func (e *FlowExecutor) executeFlowMetadata(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, user *models.UserSession) (map[string]interface{}, error) {
	// Callable-only flows (InvocableProcess callees) have no start element
	// but do have startElementReference pointing to the first element.
	// Fall back to it so the executor can enter the flow.
	var firstElement string
	if flowMeta.Start != nil && flowMeta.Start.Connector != nil {
		firstElement = flowMeta.Start.Connector.TargetReference
	} else if flowMeta.StartElementReference != "" {
		firstElement = flowMeta.StartElementReference
	} else if flowMeta.Start != nil {
		// AsyncAfterCommit and other scheduled-path flows have no direct Start.Connector.
		// Instead, the entry element is carried by Start.ScheduledPaths[].Connector.
		// Use the first non-empty scheduled path connector as the async entry point.
		for _, sp := range flowMeta.Start.ScheduledPaths {
			if sp.Connector != nil && sp.Connector.TargetReference != "" {
				firstElement = sp.Connector.TargetReference
				break
			}
		}
		if firstElement == "" {
			return nil, apperrors.NewValidationError("start", "flow has no start element or startElementReference")
		}
	} else {
		return nil, apperrors.NewValidationError("start", "flow has no start element or startElementReference")
	}

	// Stash the trigger object type so resolveNestedReferenceCtx can translate
	// `__r` segments to concrete object types via the FK field's ReferenceTo.
	// This key is transient (see flow_interview_service.go UpdateInterview).
	if interview.ContextData != nil {
		if _, present := interview.ContextData[flowContextKeyRootObject]; !present {
			// Primary source: the flow's own trigger object.
			if flowMeta.Start != nil && flowMeta.Start.Object != "" {
				interview.ContextData[flowContextKeyRootObject] = flowMeta.Start.Object
			} else {
				// Fallback for AutoLaunchedFlow subflows invoked with a
				// $Record input: recover the type from the record's
				// attributes.type (stamped by the parent executeSubflow or
				// by a RecordLookup that produced the record).
				if inferred := inferRootObjectType(interview.ContextData); inferred != "" {
					interview.ContextData[flowContextKeyRootObject] = inferred
				}
			}
		}
	}

	// Evaluate declared formula resources against the current ContextData so
	// downstream elements can reference them by name. Called here (rather
	// than in executeFlowVersion) because executeFlowMetadata is also the
	// re-entry point for ResumeInterview — on resume, TODAY()/NOW()-derived
	// formulas must refresh against the current wall clock before the flow
	// resumes execution. See design doc GAP 3 / Phase 2.3.
	e.evaluateFlowFormulas(ctx, interview, flowMeta, user)

	// Start from the first element
	currentElement := firstElement
	if interview.CurrentElement != "" {
		// Resume from paused element
		currentElement = interview.CurrentElement
	}

	completedSteps := 0
	maxIterations := 1000 // Prevent infinite loops

	for i := 0; i < maxIterations && currentElement != ""; i++ {
		interview.CurrentElement = currentElement

		nextElement, err := e.executeElement(ctx, interview, flowMeta, currentElement, user)
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("error executing element '%s'", currentElement), err)
		}

		completedSteps++
		interview.ContextData["_completedSteps"] = float64(completedSteps)

		// Move to next element
		currentElement = nextElement
	}

	// Collect output variables
	outputs := make(map[string]interface{})
	for _, variable := range flowMeta.Variables {
		if variable.IsOutput {
			if val, ok := interview.ContextData[variable.Name]; ok {
				outputs[variable.Name] = val
			}
		}
	}
	if record, ok := interview.ContextData["$Record"]; ok {
		outputs["$Record"] = record
	}
	if oldRecord, ok := interview.ContextData["$Record__Old"]; ok {
		outputs["$Record__Old"] = oldRecord
	}

	return outputs, nil
}

// executeElement executes a single flow element and returns the next element name.
// Individual element executors are in:
//   - executor_elements.go    (Decision, Assignment, Loop)
//   - executor_record_ops.go  (RecordCreate, RecordUpdate, RecordDelete, RecordLookup)
//   - executor_actions.go     (ActionCall, Subflow)
func (e *FlowExecutor) executeElement(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, elementName string, user *models.UserSession) (string, error) {
	// Find and execute the element by type

	// Check Decisions
	for _, decision := range flowMeta.Decisions {
		if decision.Name == elementName {
			return e.executeDecision(ctx, interview, &decision, user)
		}
	}

	// Check Assignments
	for _, assignment := range flowMeta.Assignments {
		if assignment.Name == elementName {
			return e.executeAssignment(ctx, interview, &assignment, user)
		}
	}

	// Check Record Creates
	for _, create := range flowMeta.RecordCreates {
		if create.Name == elementName {
			return e.executeRecordCreate(ctx, interview, &create, user)
		}
	}

	// Check Record Updates
	for _, update := range flowMeta.RecordUpdates {
		if update.Name == elementName {
			return e.executeRecordUpdate(ctx, interview, &update, user)
		}
	}

	// Check Record Deletes
	for _, del := range flowMeta.RecordDeletes {
		if del.Name == elementName {
			return e.executeRecordDelete(ctx, interview, &del, user)
		}
	}

	// Check Record Lookups
	for _, lookup := range flowMeta.RecordLookups {
		if lookup.Name == elementName {
			return e.executeRecordLookup(ctx, interview, &lookup, user)
		}
	}

	// Check Action Calls
	for _, action := range flowMeta.ActionCalls {
		if action.Name == elementName {
			return e.executeActionCall(ctx, interview, flowMeta, &action, user)
		}
	}

	// Check Subflows
	for _, subflow := range flowMeta.Subflows {
		if subflow.Name == elementName {
			return e.executeSubflow(ctx, interview, &subflow, user)
		}
	}

	// Check Loops
	for _, loop := range flowMeta.Loops {
		if loop.Name == elementName {
			return e.executeLoop(ctx, interview, &loop, user)
		}
	}

	// Check Custom Errors
	for _, customErr := range flowMeta.CustomErrors {
		if customErr.Name == elementName {
			return e.executeCustomError(ctx, interview, &customErr)
		}
	}

	// Check Waits
	for _, wait := range flowMeta.Waits {
		if wait.Name == elementName {
			return e.executeWait(ctx, interview, &wait, user)
		}
	}

	return "", apperrors.NewNotFoundError("flow element", elementName)
}

// FlowCustomErrorFault is returned by executeCustomError when a flow's
// customError element fires. Carries the author-declared statusCode (if any)
// so the dispatcher can surface SF-API-compatible error codes
// (e.g. "ALREADY_CONVERTED", "MALFORMED_ID") in FlowExecutionResult.Errors[]
// instead of the generic "FLOW_ERROR" sentinel.
type FlowCustomErrorFault struct {
	Message    string
	StatusCode string
}

func (f *FlowCustomErrorFault) Error() string { return f.Message }

// executeCustomError handles a custom error element, returning an error to
// block execution. The returned error is a *FlowCustomErrorFault when the
// author declared a statusCode, so executeFlowVersion can propagate the
// SF-native code into the dispatcher response.
func (e *FlowExecutor) executeCustomError(ctx context.Context, interview *models.FlowInterview, customErr *models.FlowCustomError) (string, error) {
	msg := customErr.Label
	statusCode := ""
	if len(customErr.CustomErrorMessages) > 0 {
		msg = customErr.CustomErrorMessages[0].ErrorMessage
		statusCode = customErr.CustomErrorMessages[0].StatusCode
	}
	return "", &FlowCustomErrorFault{Message: msg, StatusCode: statusCode}
}