// Package automation contains the flow execution engine and associated
// runtime state types.
package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"github.com/nexuscrm/shared/pkg/models"
)

// FlowExecutionIdentity is the trusted runtime provenance of the Flow currently
// executing on this context. It is derived from FlowDefinition/Flow rows loaded
// by the executor, not from Flow metadata inputParameters.
type FlowExecutionIdentity struct {
	APIName                string
	DefinitionID           string
	VersionID              string
	OwnerPack              string
	DefinitionIsCustomized bool
	VersionSource          string
}

// connectorInboundCompanionSubflowDelegation is a runtime-only, one-hop grant
// from an exact connector-inbound companion Flow to one directly declared
// child Flow. It never comes from Flow inputs or persisted metadata.
type connectorInboundCompanionSubflowDelegation struct {
	flowAPIName string
	ownerPack   string
}

// flowExecutionState holds the per-execution runtime-internal fields that
// were previously stored as map[string]interface{} entries in
// interview.ContextData under private keys such as __r_cache, _objectType,
// _loopDepth, and the historical _loop_* loop runtime cache keys.
//
// The struct is heap-allocated once per interview execution by
// NewFlowRuntimeContext and shared by pointer across all FlowRuntimeContext
// clones that derive from the same root.  clone() copies the pointer value (not
// the struct), so all derived contexts operate on the same mutable state —
// matching the semantics of the previous interview.ContextData shared-map
// approach.
//
// Exported-level read and write go through the typed accessors on
// FlowRuntimeContext; code outside this file must not access the struct fields
// directly.
type flowExecutionState struct {
	// rCache is the per-interview lazy-load cache for __r cross-object
	// relationship traversal. Keys are "{objectType}:{recordId}".
	// Lazily initialised on first write via FlowRuntimeContext.RCache().
	rCache map[string]models.SObject

	// recordLookupCache is a per-interview cache for deterministic Flow
	// RecordLookup reads. It is intentionally scoped to the current runtime
	// execution state so it cannot become a cross-request business-data source
	// of truth.
	recordLookupCache map[string]*models.QueryResult

	// recordLookupBaselines preserves the immutable field values returned by
	// RecordLookup, keyed by object and record ID. Collection RecordUpdate uses
	// this provenance to distinguish fields that were merely read from fields
	// that the Flow actually changed.
	recordLookupBaselines map[string]models.SObject

	// objectType is the root trigger object type for $Record / $Record__Old /
	// $Record__Prior resolution (previously stored as _objectType in
	// interview.ContextData).
	objectType string

	// loopDepth tracks the current Loop nesting depth so loadRelatedRecord
	// can emit an N+1 warning when a relationship is resolved inside a loop
	// body (previously stored as _loopDepth as float64 in interview.ContextData).
	loopDepth int

	// loopIndices holds per-loop runtime state that was previously stored in
	// interview.ContextData under _loop_<LoopName>_index. Keeping it here
	// prevents caller-supplied flow inputs from pre-seeding loop execution
	// state while leaving ContextData as the sole authority for loop
	// collectionReference values between iterations.
	loopIndices map[string]int

	// preparedRecordUpdates stores post-before-save prepared update payloads
	// keyed by "{objectApiName}|{recordId}". previewRecordUpdate writes here,
	// and a later finalization action can resume the shared update pipeline
	// without depending on a hardcoded flow variable name.
	preparedRecordUpdates map[string]*models.PreparedRecordUpdate

	// retriable is true when a child flow (via executeFlowAction or
	// executeSubflow) signaled that the current execution should be treated as
	// a retriable failure. The inbound worker checks this flag on successful
	// completion and calls MarkFailed instead of MarkProcessed, leaving the
	// event in the queue for retry after prerequisites land.
	//
	// This is a typed runtime field, not a magic flow variable name, so it
	// cannot collide with user-authored flow variables or inputs.
	retriable bool
}

// postUpsertRegistry is a per-execution store for engine-confirmed upserted
// records. It lives at the top level of FlowRuntimeContext (not inside
// flowExecutionState) so the pointer is preserved across PushCallStack subflow
// resets — all derived contexts share the same registry through clone().
//
// The registry accumulates snapshots per object (a list, not last-write-wins)
// so the register action can detect ambiguity when two different records of the
// same object are upserted in the same execution.
type postUpsertRegistry struct {
	records        map[string][]models.SObject // keyed by lowercased object name
	writtenObjects map[string]struct{}         // keyed by lowercased object name
	// mutationSeen marks that this execution attempt performed record DML.
	// It is kept with the post-upsert registry because CloneWithFreshPostUpsert
	// is the retry/root-dispatch reset boundary for mutation-sensitive state.
	mutationSeen bool
}

type activeFlowVersionCache struct {
	mu       sync.Mutex
	versions map[string]*models.FlowVersion // keyed by lowercased FlowDefinition.DeveloperName
}

func newActiveFlowVersionCache() *activeFlowVersionCache {
	return &activeFlowVersionCache{versions: make(map[string]*models.FlowVersion)}
}

func newPostUpsertRegistry() *postUpsertRegistry {
	return &postUpsertRegistry{
		records:        make(map[string][]models.SObject),
		writtenObjects: make(map[string]struct{}),
	}
}

// add appends a defensive copy of record under the given object.
func (r *postUpsertRegistry) add(object string, record models.SObject) {
	key := strings.ToLower(strings.TrimSpace(object))
	if key == "" {
		return
	}
	r.records[key] = append(r.records[key], cloneRuntimeSObject(record))
}

func (r *postUpsertRegistry) markWrite(object string) {
	if r == nil {
		return
	}
	key := strings.ToLower(strings.TrimSpace(object))
	if key == "" {
		return
	}
	if r.writtenObjects == nil {
		r.writtenObjects = make(map[string]struct{})
	}
	r.writtenObjects[key] = struct{}{}
}

func (r *postUpsertRegistry) hasWrite(object string) bool {
	if r == nil {
		return false
	}
	key := strings.ToLower(strings.TrimSpace(object))
	if key == "" {
		return false
	}
	_, ok := r.writtenObjects[key]
	return ok
}

func (r *postUpsertRegistry) markMutation() {
	if r != nil {
		r.mutationSeen = true
	}
}

func (r *postUpsertRegistry) hasMutation() bool {
	return r != nil && r.mutationSeen
}

// get returns defensive copies of all records stored for the given object.
func (r *postUpsertRegistry) get(object string) []models.SObject {
	key := strings.ToLower(strings.TrimSpace(object))
	if key == "" {
		return nil
	}
	src := r.records[key]
	if len(src) == 0 {
		return nil
	}
	out := make([]models.SObject, len(src))
	for i, s := range src {
		out[i] = cloneRuntimeSObject(s)
	}
	return out
}

func (r *activeFlowVersionCache) get(apiName string) (*models.FlowVersion, bool) {
	if r == nil {
		return nil, false
	}
	key := activeFlowVersionCacheKey(apiName)
	if key == "" {
		return nil, false
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	version, ok := r.versions[key]
	if !ok || version == nil {
		return nil, false
	}
	return cloneRuntimeFlowVersion(version), true
}

func (r *activeFlowVersionCache) set(apiName string, version *models.FlowVersion) {
	if r == nil || version == nil {
		return
	}
	key := activeFlowVersionCacheKey(apiName)
	if key == "" {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.versions[key] = cloneRuntimeFlowVersion(version)
}

func activeFlowVersionCacheKey(apiName string) string {
	return strings.ToLower(strings.TrimSpace(apiName))
}

// InterviewPersistMode controls whether a FlowInterview row is written to the
// database during flow execution. This replaces the flowInterviewPersistModeKey
// context.WithValue payload, which stored a bool (false = transient, absent =
// persist). The default zero value (InterviewPersistModeSave) preserves
// backward-compatible behavior: REST-invoked, scheduled, and subflow-invoked
// runs persist their audit rows unless the caller explicitly opts out.
type InterviewPersistMode int

const (
	// InterviewPersistModeSave (zero value) directs executeFlowVersion to
	// INSERT and UPDATE the FlowInterview row. This is the backward-compatible
	// default for REST-invoked, scheduled, and subflow-invoked flows.
	InterviewPersistModeSave InterviewPersistMode = iota

	// InterviewPersistModeTransient directs executeFlowVersion to use an
	// in-memory FlowInterview and skip both DB round-trips. The record-triggered
	// listener wraps its context with this mode to eliminate the ~160s-per-save
	// latency observed when FlowInterview rows were being written on every
	// record-triggered flow execution.
	InterviewPersistModeTransient
)

// FlowRuntimeContext carries per-execution state that was previously threaded
// through context.WithValue. It replaces five private context keys:
//
//   - flowInterviewPersistModeKey (payload: bool)           → PersistMode field
//   - beforeSaveFlowKey (payload: bool)                     → beforeSaveFlow field
//   - atomicFlowBoundaryKey (payload: bool)                 → atomicBoundary field
//   - flowCallStackKey (payload: []string)                  → callStack field
//   - recordTriggerStackKey (payload: map[string]struct{})  → recordTriggerStack field
//
// All builder methods return a new *FlowRuntimeContext; the receiver is never
// mutated in place. Slice and map fields are copied on write, ensuring
// goroutine safety across concurrent subflow invocations.
//
// FlowRuntimeContext is a companion to context.Context, not a replacement for
// it. Every function that previously took ctx context.Context continues to take
// it (for cancellation and deadline propagation). The typed fields here carry
// only flow-execution-specific state.
//
// Nil receivers are safe for all accessor methods; they return documented zero
// values.
type FlowRuntimeContext struct {
	// ctx is the underlying context.Context carried alongside. Stored for
	// convenience; callers may also carry it separately.
	ctx context.Context

	// user is the user session associated with this execution. May be nil for
	// system-initiated flows.
	user *models.UserSession

	// persistMode controls whether executeFlowVersion writes FlowInterview rows
	// to the database. Zero value (InterviewPersistModeSave) preserves existing
	// behavior. Set to InterviewPersistModeTransient by the record-triggered
	// listener hot path via WithInterviewPersistMode.
	//
	// Replaces flowInterviewPersistModeKey context.WithValue.
	persistMode InterviewPersistMode

	// beforeSaveFlow is true when the execution was initiated from a
	// RecordBeforeSave / RecordBeforeDelete trigger. Element executors check
	// this flag to decide whether implicit-$Record recordUpdate should mutate
	// the in-flight record in memory instead of issuing a DB UPDATE.
	//
	// Replaces beforeSaveFlowKey context.WithValue.
	beforeSaveFlow bool

	// atomicBoundary is true when execution is running inside the Flow
	// engine's RunAtomically wrapper. This is narrower than "a DB transaction
	// exists on ctx": it specifically guards against child flows containing
	// Wait elements or email-I/O action calls, which would hold the atomic
	// transaction open across I/O or resume boundaries.
	//
	// Replaces atomicFlowBoundaryKey context.WithValue.
	atomicBoundary bool

	// previewMode is true when the execution was initiated via
	// FlowExecutor.PreviewRecordCreateFields (the no-DML preview path).
	// Element executors and the audit writer check this flag to suppress
	// every side-effect channel that previewPersistence cannot intercept on
	// its own:
	//   - audit row emission (pkgAuditWriter writes _System_FlowAuditLog)
	//   - executeRecordUpdate's in-place $Record mutation branch (returns
	//     before reaching persistence.Update)
	//   - executeWait pause persistence (pauseForWait writes FlowInterview
	//     directly via persistence.Insert/Update)
	previewMode bool

	// currentFlow records the trusted FlowDefinition/Flow provenance for the
	// currently executing Flow. Action executors use this as an authorization
	// root for pack-owned metadata reads instead of trusting Flow input values.
	currentFlow    FlowExecutionIdentity
	hasCurrentFlow bool

	// connectorInboundCompanionSubflow is replaced at every native Subflow
	// boundary. A child may consume the grant for its own protected execution
	// and inbound finalizer, but cannot pass it to a grandchild.
	connectorInboundCompanionSubflow connectorInboundCompanionSubflowDelegation

	// callStack holds the flow API names currently executing in this goroutine's
	// call chain. Used to detect direct recursion (A→A) and cycles (A→B→A) when
	// an actionType="flow" action is dispatched.
	//
	// Replaces flowCallStackKey context.WithValue (payload: []string).
	// Always a copy-on-write slice; PushCallStack never mutates this field.
	callStack []string

	// recordTriggerStack holds legacy (object, recordID) pairs currently inside
	// an executeRecordTriggerFlows call on this goroutine. New record-trigger
	// dispatch uses the 3-arg FlowDefinition-aware guard stored in
	// executionState.recordTriggerFlows.
	//
	// Replaces recordTriggerStackKey context.WithValue (payload: map[string]struct{}).
	// Always a copy-on-write map; PushRecordTrigger never mutates this field.
	recordTriggerStack map[string]struct{}

	// recordTriggerFlows holds FlowDefinition-aware record-trigger dispatch keys.
	// It intentionally lives outside executionState so executeFlowMetadata's
	// per-interview executionState reset preserves reentrancy markers for
	// recordUpdate / recordCreate republishes inside the running flow.
	// Always a copy-on-write map; PushRecordTriggerFlow never mutates this field.
	recordTriggerFlows map[string]struct{}

	// recordTriggerRoot marks that the current execution entered the Flow
	// engine through FlowTriggerListener for a record-triggered root dispatch.
	// recordTriggerRootCallDepth stores the flow call-stack depth observed at
	// that root handoff so subflows can be distinguished from the root even
	// when the listener fires inside an existing parent flow chain.
	recordTriggerRoot          bool
	recordTriggerRootCallDepth int
	recordTriggerNewSnapshot   models.SObject
	recordTriggerOldSnapshot   models.SObject

	// executionState holds mutable per-interview runtime fields (rCache,
	// objectType, loopDepth, loopIndices). The pointer is shared across all
	// clones so in-place mutations (e.g. IncrLoopDepth during loop execution)
	// are visible to the original context and any derived contexts — exactly
	// the same sharing semantics as the previous interview.ContextData map.
	executionState *flowExecutionState

	// postUpsert is the engine-owned registry of records confirmed by a
	// successful persistence.Insert or persistence.Update in this execution.
	// The pointer lives here (NOT inside executionState) so it survives
	// PushCallStack's per-subflow executionState reset — clone() shallow-copies
	// the pointer, so parent and child contexts share the same registry.
	// Initialized by NewFlowRuntimeContext; never nil on a properly constructed
	// context.
	postUpsert *postUpsertRegistry

	// activeFlowVersions memoizes FlowDefinition DeveloperName -> active Flow row
	// within one request/execution tree. It stores metadata facts only; business
	// records and lock-sensitive reads are not cached here.
	activeFlowVersions *activeFlowVersionCache
}

// NewFlowRuntimeContext constructs a minimal FlowRuntimeContext. All five
// payload fields are initialised to their zero / absent values:
//
//   - PersistMode: InterviewPersistModeSave (write FlowInterview rows)
//   - BeforeSaveFlow: false (not a before-save trigger)
//   - AtomicBoundary: false (not inside an atomic-flow boundary)
//   - CallStack: empty (no subflow calls in progress)
//   - RecordTriggerStack: empty (no record-trigger re-entrancy in progress)
//
// Callers use the builder methods (WithInterviewPersistMode, WithBeforeSaveFlow,
// etc.) to layer in the required flags before passing the context to
// executeFlowVersion or the element executors.
func NewFlowRuntimeContext(ctx context.Context, user *models.UserSession) *FlowRuntimeContext {
	return &FlowRuntimeContext{
		ctx:                ctx,
		user:               user,
		executionState:     &flowExecutionState{},
		postUpsert:         newPostUpsertRegistry(),
		activeFlowVersions: newActiveFlowVersionCache(),
	}
}

// Context returns the underlying context.Context carried by this
// FlowRuntimeContext. Safe to call on a nil receiver (returns context.Background()).
func (c *FlowRuntimeContext) Context() context.Context {
	if c == nil || c.ctx == nil {
		return context.Background()
	}
	return c.ctx
}

// User returns the user session associated with this execution. May be nil.
// Safe to call on a nil receiver.
func (c *FlowRuntimeContext) User() *models.UserSession {
	if c == nil {
		return nil
	}
	return c.user
}

// ----------------------------------------------------------------------------
// InterviewPersistMode
// ----------------------------------------------------------------------------

// WithInterviewPersistMode returns a new *FlowRuntimeContext with the persist
// mode set to mode. The receiver is not modified.
//
// Use InterviewPersistModeTransient on the record-triggered hot path to skip
// FlowInterview DB round-trips. Use InterviewPersistModeSave (zero value, the
// default) for REST-invoked, scheduled, and subflow-invoked flows.
func (c *FlowRuntimeContext) WithInterviewPersistMode(mode InterviewPersistMode) *FlowRuntimeContext {
	n := c.clone()
	n.persistMode = mode
	return n
}

// GetInterviewPersistMode returns the current persist mode. Safe to call on a
// nil receiver, which returns InterviewPersistModeSave (the zero / default value).
func (c *FlowRuntimeContext) GetInterviewPersistMode() InterviewPersistMode {
	if c == nil {
		return InterviewPersistModeSave
	}
	return c.persistMode
}

// ShouldPersistInterview reports true when the FlowInterview row should be
// written to the database. This mirrors the semantics of the old
// shouldPersistFlowInterview(ctx) helper.
//
// Returns true when c is nil (the zero / default behavior is to persist).
func (c *FlowRuntimeContext) ShouldPersistInterview() bool {
	if c == nil {
		return true
	}
	return c.persistMode == InterviewPersistModeSave
}

// CachedActiveFlowVersion returns the active FlowVersion loaded earlier in this
// request/execution tree for the given FlowDefinition DeveloperName.
func (c *FlowRuntimeContext) CachedActiveFlowVersion(apiName string) (*models.FlowVersion, bool) {
	if c == nil || c.activeFlowVersions == nil {
		return nil, false
	}
	return c.activeFlowVersions.get(apiName)
}

// SetCachedActiveFlowVersion stores an active FlowVersion for reuse by later
// dispatches of the same FlowDefinition DeveloperName in this execution tree.
func (c *FlowRuntimeContext) SetCachedActiveFlowVersion(apiName string, version *models.FlowVersion) {
	if c == nil {
		return
	}
	if c.activeFlowVersions == nil {
		c.activeFlowVersions = newActiveFlowVersionCache()
	}
	c.activeFlowVersions.set(apiName, version)
}

// ----------------------------------------------------------------------------
// BeforeSaveFlow
// ----------------------------------------------------------------------------

// WithBeforeSaveFlow returns a new *FlowRuntimeContext with the before-save
// marker set. The receiver is not modified.
//
// Only the FlowTriggerListener's before-save branch should call this.
// AfterSave, scheduled, and REST-invoked flows must not set this flag.
func (c *FlowRuntimeContext) WithBeforeSaveFlow(marker bool) *FlowRuntimeContext {
	n := c.clone()
	n.beforeSaveFlow = marker
	return n
}

// IsBeforeSaveFlow reports whether the current execution is a before-save
// record-triggered flow. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) IsBeforeSaveFlow() bool {
	if c == nil {
		return false
	}
	return c.beforeSaveFlow
}

// ----------------------------------------------------------------------------
// AtomicBoundary
// ----------------------------------------------------------------------------

// WithAtomicBoundary returns a new *FlowRuntimeContext with the atomic
// boundary marker set. The receiver is not modified.
//
// Only FlowExecutor.RunAtomically should set this. It prevents child flows
// from containing Wait elements or email-I/O action calls (which would hold the
// open transaction across I/O or resume boundaries).
func (c *FlowRuntimeContext) WithAtomicBoundary(marker bool) *FlowRuntimeContext {
	n := c.clone()
	n.atomicBoundary = marker
	return n
}

// IsWithinAtomicBoundary reports whether execution is inside the Flow engine's
// RunAtomically wrapper. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) IsWithinAtomicBoundary() bool {
	if c == nil {
		return false
	}
	return c.atomicBoundary
}

// ----------------------------------------------------------------------------
// PreviewMode
// ----------------------------------------------------------------------------

// WithPreviewMode returns a new *FlowRuntimeContext with the preview-mode
// marker set. The receiver is not modified.
//
// Only PreviewRecordCreateFields should call this. It instructs the audit
// writer, executeRecordUpdate, and executeWait to refuse any DB-touching
// path that previewPersistence cannot already intercept on its own.
func (c *FlowRuntimeContext) WithPreviewMode(marker bool) *FlowRuntimeContext {
	n := c.clone()
	n.previewMode = marker
	return n
}

// IsPreviewMode reports whether the current execution is a no-DML preview
// run. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) IsPreviewMode() bool {
	if c == nil {
		return false
	}
	return c.previewMode
}

// ----------------------------------------------------------------------------
// CallStack (flowCallStackKey replacement)
// ----------------------------------------------------------------------------

// WithCurrentFlow returns a new *FlowRuntimeContext with the current Flow
// identity set. Blank string fields are normalised so authorization callers can
// compare provenance deterministically.
func (c *FlowRuntimeContext) WithCurrentFlow(identity FlowExecutionIdentity) *FlowRuntimeContext {
	n := c.clone()
	identity.APIName = strings.TrimSpace(identity.APIName)
	identity.DefinitionID = strings.TrimSpace(identity.DefinitionID)
	identity.VersionID = strings.TrimSpace(identity.VersionID)
	identity.OwnerPack = strings.TrimSpace(identity.OwnerPack)
	identity.VersionSource = strings.TrimSpace(identity.VersionSource)
	n.currentFlow = identity
	n.hasCurrentFlow = true
	return n
}

// CurrentFlow returns the trusted runtime identity for the currently executing
// Flow, if one has been threaded by executeFlowVersion / ResumeInterview.
func (c *FlowRuntimeContext) CurrentFlow() (FlowExecutionIdentity, bool) {
	if c == nil || !c.hasCurrentFlow {
		return FlowExecutionIdentity{}, false
	}
	return c.currentFlow, true
}

func (c *FlowRuntimeContext) withConnectorInboundCompanionSubflowDelegation(flowAPIName, ownerPack string) *FlowRuntimeContext {
	n := c.clone()
	n.connectorInboundCompanionSubflow = connectorInboundCompanionSubflowDelegation{
		flowAPIName: strings.TrimSpace(flowAPIName),
		ownerPack:   strings.TrimSpace(ownerPack),
	}
	return n
}

func (c *FlowRuntimeContext) connectorInboundCompanionSubflowAuthorizes(identity FlowExecutionIdentity) bool {
	if c == nil || strings.TrimSpace(identity.APIName) == "" || strings.TrimSpace(identity.OwnerPack) == "" {
		return false
	}
	delegation := c.connectorInboundCompanionSubflow
	return strings.EqualFold(strings.TrimSpace(delegation.flowAPIName), strings.TrimSpace(identity.APIName)) &&
		strings.EqualFold(strings.TrimSpace(delegation.ownerPack), strings.TrimSpace(identity.OwnerPack))
}

// CurrentFlowOwnerPack returns the trusted owner pack for the currently
// executing Flow. It is empty when no pack-owned provenance is available.
func (c *FlowRuntimeContext) CurrentFlowOwnerPack() string {
	if c == nil || !c.hasCurrentFlow {
		return ""
	}
	ownerPack := strings.TrimSpace(c.currentFlow.OwnerPack)
	if ownerPack == "" || c.currentFlow.DefinitionIsCustomized || c.currentFlow.VersionSource != flowSourceStandard {
		return ""
	}
	return ownerPack
}

// WithRecordTriggerRoot returns a new *FlowRuntimeContext marked as a
// record-trigger root dispatch that entered at the provided call-stack depth.
// FlowTriggerListener sets this once per root dispatch before invoking the
// matched flow set.
func (c *FlowRuntimeContext) WithRecordTriggerRoot(callDepth int) *FlowRuntimeContext {
	n := c.clone()
	n.recordTriggerRoot = true
	n.recordTriggerRootCallDepth = callDepth
	return n
}

// HasRecordTriggerRoot reports whether the current execution was marked as a
// record-trigger root dispatch. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) HasRecordTriggerRoot() bool {
	if c == nil {
		return false
	}
	return c.recordTriggerRoot
}

// RecordTriggerRootCallDepth returns the call-stack depth captured at the
// record-trigger root handoff. Safe to call on a nil receiver (returns 0).
func (c *FlowRuntimeContext) RecordTriggerRootCallDepth() int {
	if c == nil {
		return 0
	}
	return c.recordTriggerRootCallDepth
}

// PushCallStack returns a new *FlowRuntimeContext with flowApiName appended to
// the call stack. The receiver is not modified; the underlying slice is always
// copied so that concurrent or nested callers do not share storage.
//
// Safe to call on a nil receiver (treats the receiver as an empty context).
// Does not enforce the cycle-detection or depth-cap rules; that logic stays in
// withFlowCall (executor_actions.go) until T21c wires them together.
func (c *FlowRuntimeContext) PushCallStack(flowApiName string) *FlowRuntimeContext {
	n := c.clone() // nil-safe: clone handles nil c
	var prev []string
	if c != nil {
		prev = c.callStack
	}
	next := make([]string, len(prev)+1)
	copy(next, prev)
	next[len(prev)] = flowApiName
	n.callStack = next
	// Each subflow starts a new FlowInterview and must track its own objectType,
	// rCache, and loopDepth independently of the parent interview.
	n.executionState = &flowExecutionState{}
	return n
}

// PopCallStack returns a new *FlowRuntimeContext with the top-most entry
// removed from the call stack, plus the popped entry and a boolean indicating
// whether a value was available. The receiver is not modified.
//
// If the call stack is empty, returns (receiver-clone, "", false).
func (c *FlowRuntimeContext) PopCallStack() (*FlowRuntimeContext, string, bool) {
	if c == nil || len(c.callStack) == 0 {
		var zero *FlowRuntimeContext
		if c != nil {
			zero = c.clone()
		} else {
			zero = &FlowRuntimeContext{}
		}
		return zero, "", false
	}
	n := c.clone()
	top := c.callStack[len(c.callStack)-1]
	prev := c.callStack[:len(c.callStack)-1]
	next := make([]string, len(prev))
	copy(next, prev)
	n.callStack = next
	return n, top, true
}

// GetCallStack returns a defensive copy of the current call stack. Mutations
// to the returned slice do not affect the FlowRuntimeContext. Safe to call on
// a nil receiver (returns nil).
func (c *FlowRuntimeContext) GetCallStack() []string {
	if c == nil || len(c.callStack) == 0 {
		return nil
	}
	out := make([]string, len(c.callStack))
	copy(out, c.callStack)
	return out
}

// CallStackDepth returns the number of entries currently on the call stack.
// Safe to call on a nil receiver (returns 0).
func (c *FlowRuntimeContext) CallStackDepth() int {
	if c == nil {
		return 0
	}
	return len(c.callStack)
}

// HasFlowInCallStack reports whether flowApiName (case-insensitive) is already
// present anywhere in the call stack. Used for cycle detection.
// Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) HasFlowInCallStack(flowApiName string) bool {
	if c == nil {
		return false
	}
	for _, name := range c.callStack {
		if strings.EqualFold(name, flowApiName) {
			return true
		}
	}
	return false
}

// ----------------------------------------------------------------------------
// RecordTriggerStack (recordTriggerStackKey replacement)
// ----------------------------------------------------------------------------

// recordTriggerKey builds the legacy canonical map key for
// (objectApiName, recordID). Lower-cases the object component to match the
// existing withRecordTrigger behavior.
func recordTriggerKey(objectApiName, recordID string) string {
	return strings.ToLower(objectApiName) + ":" + recordID
}

// recordTriggerFlowKey builds the canonical key for a record-triggered
// FlowDefinition dispatch: (KeyPrefix, recordID, FlowDefinitionId).
func recordTriggerFlowKey(keyPrefix, recordID, flowDefinitionID string) string {
	return strings.ToLower(strings.TrimSpace(keyPrefix)) + ":" + recordID + ":" + strings.ToLower(strings.TrimSpace(flowDefinitionID))
}

// PushRecordTrigger returns a new *FlowRuntimeContext with the
// (objectApiName, recordID) pair added to the record trigger stack, plus a
// boolean `alreadyPresent`. The receiver is not modified; the underlying map
// is always copied.
//
// If recordID is empty, the receiver is returned unchanged with alreadyPresent=false,
// matching the semantics of the old withRecordTrigger helper (BeforeCreate has
// no stable ID and must not be guarded).
func (c *FlowRuntimeContext) PushRecordTrigger(objectApiName, recordID string) (*FlowRuntimeContext, bool) {
	if recordID == "" {
		return c.clone(), false // nil-safe: clone handles nil c
	}
	key := recordTriggerKey(objectApiName, recordID)
	existing := c.getRecordTriggerStack()
	if _, present := existing[key]; present {
		return c.clone(), true // nil-safe: clone handles nil c
	}
	next := make(map[string]struct{}, len(existing)+1)
	for k := range existing {
		next[k] = struct{}{}
	}
	next[key] = struct{}{}
	n := c.clone()
	n.recordTriggerStack = next
	return n, false
}

// PushRecordTriggerFlow records one FlowDefinition dispatch for a stable
// record-trigger key.
//
// Empty recordID remains unguarded because before-create has no stable record
// identity. Empty FlowDefinitionId is also unguarded; production metadata rows
// should always carry one, and callers use a deterministic fallback before
// reaching this helper.
func (c *FlowRuntimeContext) PushRecordTriggerFlow(keyPrefix, recordID, flowDefinitionID string) (*FlowRuntimeContext, bool) {
	if strings.TrimSpace(recordID) == "" || strings.TrimSpace(flowDefinitionID) == "" {
		return c.clone(), false
	}

	key := recordTriggerFlowKey(keyPrefix, recordID, flowDefinitionID)
	existing := c.getRecordTriggerFlows()
	if _, present := existing[key]; present {
		return c.clone(), true
	}
	next := make(map[string]struct{}, len(existing)+1)
	for k := range existing {
		next[k] = struct{}{}
	}
	next[key] = struct{}{}
	n := c.clone()
	n.recordTriggerFlows = next
	return n, false
}

// HasRecordTriggerFlow reports whether the given FlowDefinition dispatch key
// has already been seen in the current event chain.
func (c *FlowRuntimeContext) HasRecordTriggerFlow(keyPrefix, recordID, flowDefinitionID string) bool {
	if strings.TrimSpace(recordID) == "" || strings.TrimSpace(flowDefinitionID) == "" {
		return false
	}
	key := recordTriggerFlowKey(keyPrefix, recordID, flowDefinitionID)
	_, present := c.getRecordTriggerFlows()[key]
	return present
}

// HasRecordTrigger reports whether (objectApiName, recordID) is currently
// in-flight on this execution path. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) HasRecordTrigger(objectApiName, recordID string) bool {
	if recordID == "" {
		return false
	}
	key := recordTriggerKey(objectApiName, recordID)
	existing := c.getRecordTriggerStack()
	_, present := existing[key]
	return present
}

// GetRecordTriggerStack returns a defensive copy of the current record trigger
// set as a slice of "objectApiName:recordID" keys. Mutations to the returned
// slice do not affect the FlowRuntimeContext. Safe to call on a nil receiver
// (returns nil).
func (c *FlowRuntimeContext) GetRecordTriggerStack() []string {
	if c == nil || len(c.recordTriggerStack) == 0 {
		return nil
	}
	out := make([]string, 0, len(c.recordTriggerStack))
	for k := range c.recordTriggerStack {
		out = append(out, k)
	}
	return out
}

// getRecordTriggerStack returns the internal map, nil-safe.
func (c *FlowRuntimeContext) getRecordTriggerStack() map[string]struct{} {
	if c == nil {
		return nil
	}
	return c.recordTriggerStack
}

func (c *FlowRuntimeContext) getRecordTriggerFlows() map[string]struct{} {
	if c == nil {
		return nil
	}
	return c.recordTriggerFlows
}

// WithRecordTriggerNewSnapshot stores a defensive copy of the trigger-new
// record image captured by FlowTriggerListener before flow elements can mutate
// the live $Record map.
func (c *FlowRuntimeContext) WithRecordTriggerNewSnapshot(record models.SObject) *FlowRuntimeContext {
	n := c.clone()
	n.recordTriggerNewSnapshot = cloneRuntimeSObject(record)
	return n
}

// RecordTriggerNewSnapshot returns a defensive copy of the trigger-new record
// image captured at record-trigger entry. Nil means no immutable trigger image
// is available for this execution.
func (c *FlowRuntimeContext) RecordTriggerNewSnapshot() models.SObject {
	if c == nil || c.recordTriggerNewSnapshot == nil {
		return nil
	}
	return cloneRuntimeSObject(c.recordTriggerNewSnapshot)
}

// WithRecordTriggerOldSnapshot stores a defensive copy of the trigger-old
// record image captured by FlowTriggerListener before flow elements can mutate
// the live $Record__Old map.
func (c *FlowRuntimeContext) WithRecordTriggerOldSnapshot(record models.SObject) *FlowRuntimeContext {
	n := c.clone()
	n.recordTriggerOldSnapshot = cloneRuntimeSObject(record)
	return n
}

// RecordTriggerOldSnapshot returns a defensive copy of the trigger-old record
// image captured at record-trigger entry. Nil means no immutable old image is
// available for this execution.
func (c *FlowRuntimeContext) RecordTriggerOldSnapshot() models.SObject {
	if c == nil || c.recordTriggerOldSnapshot == nil {
		return nil
	}
	return cloneRuntimeSObject(c.recordTriggerOldSnapshot)
}

// ----------------------------------------------------------------------------
// executionState accessors (ObjectType, LoopDepth, RCache, LoopState)
//
// These fields correspond to runtime-internal state that was previously stored
// in interview.ContextData under raw string keys such as __r_cache,
// _objectType, _loopDepth, and _loop_*.
// Moving them to typed fields removes the raw-string coupling and ensures they
// never appear in a persisted interview JSON snapshot or a caller-writable
// input map.
//
// All three mutate the shared flowExecutionState in place (the pointer is
// shared across all FlowRuntimeContext clones derived from the same root),
// matching the previous behaviour where all code shared the same
// interview.ContextData map reference.
// ----------------------------------------------------------------------------

// ObjectType returns the root trigger object type for $Record / $Record__Old /
// $Record__Prior resolution. Returns "" when the execution state has not been
// populated. Safe to call on a nil receiver.
func (c *FlowRuntimeContext) ObjectType() string {
	if c == nil || c.executionState == nil {
		return ""
	}
	return c.executionState.objectType
}

// SetObjectType sets the root trigger object type.  Calling on a nil receiver
// is a no-op (defensive; callers should always hold a non-nil context).
func (c *FlowRuntimeContext) SetObjectType(objectType string) {
	if c == nil || c.executionState == nil {
		return
	}
	c.executionState.objectType = objectType
}

// LoopDepth returns the current Loop nesting depth.  Returns 0 when the
// execution state has not been populated or no loop is active.
// Safe to call on a nil receiver.
func (c *FlowRuntimeContext) LoopDepth() int {
	if c == nil || c.executionState == nil {
		return 0
	}
	return c.executionState.loopDepth
}

// IncrLoopDepth increments the Loop nesting depth by 1.  Calling on a nil
// receiver is a no-op.
func (c *FlowRuntimeContext) IncrLoopDepth() {
	if c == nil || c.executionState == nil {
		return
	}
	c.executionState.loopDepth++
}

// DecrLoopDepth decrements the Loop nesting depth by 1, clamping at 0.
// Calling on a nil receiver is a no-op.
func (c *FlowRuntimeContext) DecrLoopDepth() {
	if c == nil || c.executionState == nil {
		return
	}
	if c.executionState.loopDepth > 0 {
		c.executionState.loopDepth--
	}
}

// RCache returns (or lazily initialises) the per-interview __r relationship
// resolution cache.  Keys are "{objectType}:{recordId}".
//
// Returns nil on a nil receiver (no interview context to cache for).
// All non-nil FlowRuntimeContext instances that share the same root share
// the same cache map.
func (c *FlowRuntimeContext) RCache() map[string]models.SObject {
	if c == nil || c.executionState == nil {
		return nil
	}
	if c.executionState.rCache == nil {
		c.executionState.rCache = make(map[string]models.SObject)
	}
	return c.executionState.rCache
}

// CachedRecordLookupResult returns a defensive copy of a cached Flow
// RecordLookup result for key. The cache is per-interview/runtime-state only.
func (c *FlowRuntimeContext) CachedRecordLookupResult(key string) (*models.QueryResult, bool) {
	if c == nil || c.executionState == nil || c.executionState.recordLookupCache == nil || key == "" {
		return nil, false
	}
	result, ok := c.executionState.recordLookupCache[key]
	if !ok {
		return nil, false
	}
	return cloneRuntimeQueryResult(result), true
}

// SetCachedRecordLookupResult stores a defensive copy of result for key.
func (c *FlowRuntimeContext) SetCachedRecordLookupResult(key string, result *models.QueryResult) {
	if c == nil || c.executionState == nil || key == "" || result == nil {
		return
	}
	if c.executionState.recordLookupCache == nil {
		c.executionState.recordLookupCache = make(map[string]*models.QueryResult)
	}
	c.executionState.recordLookupCache[key] = cloneRuntimeQueryResult(result)
}

// RememberRecordLookupBaselines records defensive copies of the projected
// rows exposed to Flow variables. Repeated lookups merge their field closures
// for the same record while replacing values with the newest observed ones.
func (c *FlowRuntimeContext) RememberRecordLookupBaselines(object string, records []models.SObject) {
	if c == nil || c.executionState == nil {
		return
	}
	for _, record := range records {
		id, _ := record["Id"].(string)
		key := flowRecordLookupBaselineKey(object, id)
		if key == "" {
			continue
		}
		if c.executionState.recordLookupBaselines == nil {
			c.executionState.recordLookupBaselines = make(map[string]models.SObject)
		}
		baseline := c.executionState.recordLookupBaselines[key]
		if baseline == nil {
			baseline = make(models.SObject, len(record))
		}
		for field, value := range record {
			baseline[field] = value
		}
		c.executionState.recordLookupBaselines[key] = baseline
	}
}

// RecordLookupBaseline returns a defensive copy of lookup provenance for one
// record. False means the Flow constructed the record instead of reading it.
func (c *FlowRuntimeContext) RecordLookupBaseline(object, id string) (models.SObject, bool) {
	if c == nil || c.executionState == nil || c.executionState.recordLookupBaselines == nil {
		return nil, false
	}
	baseline, ok := c.executionState.recordLookupBaselines[flowRecordLookupBaselineKey(object, id)]
	if !ok {
		return nil, false
	}
	return cloneRuntimeSObject(baseline), true
}

// AdvanceRecordLookupBaseline applies a successful mutation to the provenance
// snapshot so a later update in the same Flow diffs against the latest value.
func (c *FlowRuntimeContext) AdvanceRecordLookupBaseline(object, id string, updates models.SObject) {
	if c == nil || c.executionState == nil || len(updates) == 0 {
		return
	}
	baseline, ok := c.executionState.recordLookupBaselines[flowRecordLookupBaselineKey(object, id)]
	if !ok {
		return
	}
	for field, value := range updates {
		if models.IsRecordMetadataKey(field) || field == "Id" {
			continue
		}
		baseline[field] = value
	}
}

// ClearRecordLookupCache drops cached Flow RecordLookup rows after an in-flow
// mutation. Clearing is deliberately coarse: correctness beats hit rate.
func (c *FlowRuntimeContext) ClearRecordLookupCache() {
	if c == nil || c.executionState == nil {
		return
	}
	c.executionState.recordLookupCache = nil
}

// MarkRecordMutation records that this execution attempt performed record DML
// and clears lookup caches that may no longer be valid.
func (c *FlowRuntimeContext) MarkRecordMutation() {
	if c == nil {
		return
	}
	if c.postUpsert == nil {
		c.postUpsert = newPostUpsertRegistry()
	}
	c.postUpsert.markMutation()
	c.ClearRecordLookupCache()
}

// HasRecordMutation reports whether this execution attempt has performed any
// record DML. It is intentionally coarser than post-upsert records so deletes
// and non-published writes still block detached, cross-request batching.
func (c *FlowRuntimeContext) HasRecordMutation() bool {
	if c == nil || c.postUpsert == nil {
		return false
	}
	return c.postUpsert.hasMutation()
}

// LoopIndex returns the cached iteration index for loopName and whether a
// runtime-owned index is present.
func (c *FlowRuntimeContext) LoopIndex(loopName string) (int, bool) {
	if c == nil || c.executionState == nil || loopName == "" || c.executionState.loopIndices == nil {
		return 0, false
	}
	idx, ok := c.executionState.loopIndices[loopName]
	return idx, ok
}

// SetLoopIndex stores the runtime-owned iteration index for loopName.
func (c *FlowRuntimeContext) SetLoopIndex(loopName string, idx int) {
	if c == nil || c.executionState == nil || loopName == "" {
		return
	}
	if c.executionState.loopIndices == nil {
		c.executionState.loopIndices = make(map[string]int)
	}
	c.executionState.loopIndices[loopName] = idx
}

// ClearLoopState removes any runtime-owned loop index for loopName.
func (c *FlowRuntimeContext) ClearLoopState(loopName string) {
	if c == nil || c.executionState == nil || loopName == "" {
		return
	}
	if c.executionState.loopIndices != nil {
		delete(c.executionState.loopIndices, loopName)
	}
}

// PreparedRecordUpdate returns a clone of the prepared update payload stored
// for (objectAPIName, recordID), if any.
func (c *FlowRuntimeContext) PreparedRecordUpdate(objectAPIName string, recordID string) *models.PreparedRecordUpdate {
	if c == nil || c.executionState == nil || c.executionState.preparedRecordUpdates == nil {
		return nil
	}
	key := preparedRecordUpdateKey(objectAPIName, recordID)
	if key == "" {
		return nil
	}
	prepared := c.executionState.preparedRecordUpdates[key]
	return prepared.Clone()
}

// SetPreparedRecordUpdate stores a clone of prepared for (objectAPIName,
// recordID). Calling on a nil receiver is a no-op.
func (c *FlowRuntimeContext) SetPreparedRecordUpdate(objectAPIName string, recordID string, prepared *models.PreparedRecordUpdate) {
	if c == nil || c.executionState == nil || prepared == nil {
		return
	}
	key := preparedRecordUpdateKey(objectAPIName, recordID)
	if key == "" {
		return
	}
	if c.executionState.preparedRecordUpdates == nil {
		c.executionState.preparedRecordUpdates = make(map[string]*models.PreparedRecordUpdate)
	}
	c.executionState.preparedRecordUpdates[key] = prepared.Clone()
}

// SetRetriable marks the current execution as retriable. Child flows call this
// when they encounter a transient condition (e.g. missing prerequisite record)
// that should cause the inbound worker to retry the event instead of marking
// it processed. The flag is stored in executionState so it survives
// PushCallStack subflow boundaries and is visible to the root execution.
//
// Calling on a nil receiver is a no-op.
func (c *FlowRuntimeContext) SetRetriable() {
	if c == nil || c.executionState == nil {
		return
	}
	c.executionState.retriable = true
}

// IsRetriable reports whether the current execution has been marked as
// retriable by a child flow. Safe to call on a nil receiver (returns false).
func (c *FlowRuntimeContext) IsRetriable() bool {
	if c == nil || c.executionState == nil {
		return false
	}
	return c.executionState.retriable
}

// AddPostUpsertRecord appends a defensive copy of record to the execution-wide
// registry under the given object API name (lowercased). Calling on a nil
// receiver is a no-op.
//
// Unlike the v1 SetPostUpsertRecord (last-write-wins), this method accumulates
// all upserted records so the register action can detect ambiguity when two
// different records of the same object are persisted in one execution.
//
// The registry lives on the FlowRuntimeContext pointer, not inside
// executionState, so it survives PushCallStack subflow resets and is visible
// across the whole execution tree.
func (c *FlowRuntimeContext) AddPostUpsertRecord(object string, record models.SObject) {
	if c == nil {
		return
	}
	c.MarkRecordMutation()
	if c.postUpsert == nil {
		c.postUpsert = newPostUpsertRegistry()
	}
	c.postUpsert.add(object, record)
}

// MarkPostUpsertWrite records that the engine successfully wrote object during
// this flow execution. It is intentionally separate from AddPostUpsertRecord:
// an inbound register action uses it to distinguish a legitimate no-write path
// from a lost engine-owned snapshot after a successful write.
func (c *FlowRuntimeContext) MarkPostUpsertWrite(object string) {
	if c == nil {
		return
	}
	if c.postUpsert == nil {
		c.postUpsert = newPostUpsertRegistry()
	}
	c.postUpsert.markWrite(object)
}

// HasPostUpsertWrite reports whether the engine successfully wrote object in
// this flow execution. Safe to call on a nil receiver.
func (c *FlowRuntimeContext) HasPostUpsertWrite(object string) bool {
	return c != nil && c.postUpsert != nil && c.postUpsert.hasWrite(object)
}

// PostUpsertRecords returns defensive copies of all engine-confirmed record
// snapshots stored for the given object API name. Returns nil when none has
// been stored. Safe to call on a nil receiver.
func (c *FlowRuntimeContext) PostUpsertRecords(object string) []models.SObject {
	if c == nil || c.postUpsert == nil {
		return nil
	}
	return c.postUpsert.get(object)
}

// ErrAmbiguousPostUpsert is returned by executeRegisterConnectorInboundRemoteAuthorship
// when more than one record of the required object was upserted in the same execution.
// The action fails closed — it does not author evidence — because it cannot
// determine which record is the authoritative inbound target.
func ErrAmbiguousPostUpsertf(count int, object string) error {
	return fmt.Errorf("register inbound authorship: ambiguous source record: %d records of object %q upserted in this execution", count, object)
}

// CloneWithFreshPostUpsert returns a new *FlowRuntimeContext that is identical
// to c in every way EXCEPT that it carries a fresh, empty postUpsertRegistry.
// This isolates one root-flow dispatch from sibling root flows that share a
// base ctx (issue #2080 Gate-0 P1-A): each root flow gets its own registry so
// post-upsert records published by root A are not visible to root B's register
// action. Subflows of a given root still share the registry via PushCallStack,
// because PushCallStack calls clone() which shallow-copies the postUpsert pointer.
func (c *FlowRuntimeContext) CloneWithFreshPostUpsert() *FlowRuntimeContext {
	n := c.clone()
	n.postUpsert = newPostUpsertRegistry()
	n.activeFlowVersions = newActiveFlowVersionCache()
	return n
}

// ----------------------------------------------------------------------------
// Internal helpers
// ----------------------------------------------------------------------------

// clone produces a shallow copy of c. The callStack and recordTriggerStack
// header pointers are copied (sharing the underlying arrays), but Push/Pop
// operations always replace them with new allocations, so sharing is safe.
// The executionState pointer is also copied so all clones share the same
// mutable runtime state (rCache, objectType, loopDepth). The postUpsert
// pointer is also shallow-copied so parent and child contexts (including
// subflow contexts created by PushCallStack) share the same registry.
func (c *FlowRuntimeContext) clone() *FlowRuntimeContext {
	if c == nil {
		return &FlowRuntimeContext{
			executionState:     &flowExecutionState{},
			postUpsert:         newPostUpsertRegistry(),
			activeFlowVersions: newActiveFlowVersionCache(),
		}
	}
	n := *c // shallow copy of the struct
	return &n
}

func preparedRecordUpdateKey(objectAPIName string, recordID string) string {
	objectAPIName = strings.ToLower(strings.TrimSpace(objectAPIName))
	recordID = strings.TrimSpace(recordID)
	if objectAPIName == "" || recordID == "" {
		return ""
	}
	return objectAPIName + "|" + recordID
}

func cloneRuntimeSObject(record models.SObject) models.SObject {
	if record == nil {
		return nil
	}
	clone := make(models.SObject, len(record))
	for k, v := range record {
		clone[k] = v
	}
	return clone
}

func flowRecordLookupBaselineKey(object, id string) string {
	object = strings.ToLower(strings.TrimSpace(object))
	id = strings.TrimSpace(id)
	if object == "" || id == "" {
		return ""
	}
	return object + "|" + id
}

func cloneRuntimeSObjects(records []models.SObject) []models.SObject {
	if len(records) == 0 {
		return nil
	}
	out := make([]models.SObject, len(records))
	for i, record := range records {
		out[i] = cloneRuntimeSObject(record)
	}
	return out
}

func cloneRuntimeQueryResult(result *models.QueryResult) *models.QueryResult {
	if result == nil {
		return nil
	}
	clone := *result
	clone.Records = cloneRuntimeSObjects(result.Records)
	return &clone
}

func cloneRuntimeFlowVersion(version *models.FlowVersion) *models.FlowVersion {
	if version == nil {
		return nil
	}
	clone := *version
	if version.MetadataRaw != nil {
		clone.MetadataRaw = append([]byte(nil), version.MetadataRaw...)
	}
	if version.MetadataJSON != "" {
		clone.HydrateMetadata()
		return &clone
	}
	if raw, err := json.Marshal(version.Metadata); err == nil && string(raw) != "{}" {
		var metadata models.FlowMetadata
		if err := json.Unmarshal(raw, &metadata); err == nil {
			clone.Metadata = metadata
		}
	}
	return &clone
}