manager.go

// Copyright (c) 2025 Ehab Terra
// SPDX-License-Identifier: MIT

package workflow

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand/v2"
	"time"
)

// defFingerprintKey is the reserved context key under which the Manager persists
// a workflow definition's fingerprint. It is written into the stored context on
// save and stripped from the live context on load, so it never appears in a
// workflow's user-visible context or guard environment.
const defFingerprintKey = "__workflow_def_fingerprint"

// defShapeKey is the reserved context key under which the Manager persists the
// definition's structural shape (place names + per-transition record hashes,
// a few hundred bytes). It is what turns a later fingerprint mismatch into a
// DefinitionDiff for the migration handler. Written on save, stripped on load,
// exactly like defFingerprintKey.
const defShapeKey = "__workflow_def_shape"

// DefinitionMismatch describes a stored-vs-supplied definition mismatch, as
// handed to the WithDefinitionMigration handler.
type DefinitionMismatch struct {
	WorkflowID string

	// StoredFingerprint is the fingerprint stamped on the instance's last
	// save; empty for a row that was not written by this library's save path.
	// CurrentFingerprint is the fingerprint of the definition supplied now.
	StoredFingerprint  string
	CurrentFingerprint string

	// Diff is the structural difference from the definition the instance was
	// last saved under to the one supplied now — what was added, removed, or
	// rewired, by name. It is nil when the stored state carries no shape (it
	// predates shape stamping, or was not written by this library): nil means
	// "no information", not "no change".
	Diff *DefinitionDiff
}

// DefinitionMigrationFunc is called by the Manager when a persisted instance's
// stored definition fingerprint differs from the definition supplied to load it.
// Returning nil lets the load proceed (the caller has confirmed the change is
// safe, or has migrated the marking); returning an error aborts the load with
// that error. The mismatch carries both fingerprints and, when the stored
// state includes a shape, the structural DefinitionDiff — so approval can be
// a policy (e.g. mismatch.Diff.Additive()) rather than blind trust.
type DefinitionMigrationFunc func(ctx context.Context, mismatch DefinitionMismatch) error

// Manager handles workflow instances and their persistence.
//
// The Manager reserves the context keys "__workflow_def_fingerprint" and
// "__workflow_def_shape" for the definition fingerprint and structural shape
// it stamps on every save: user values stored under those keys are
// overwritten on save and stripped on load.
type Manager struct {
	registry *Registry
	storage  Storage

	// listeners holds the dynamic listeners for all managed workflows. It is
	// concurrency-safe: listeners may be added or removed while managed
	// workflows fire transitions on other goroutines.
	listeners listenerSet

	// useCache controls whether loaded instances are cached in the registry.
	// Fresh loads are the default: they are correct in every deployment shape,
	// and optimistic concurrency protects the saves. Caching is an explicit
	// single-process optimization (WithRegistryCache) — a cached copy can be
	// stale versus the database the moment another replica saves.
	useCache bool

	// onDefinitionMismatch, if set, is consulted when a loaded instance's stored
	// fingerprint differs from the definition's; nil means mismatches are errors.
	onDefinitionMismatch DefinitionMigrationFunc

	// effects resolves the effect names transitions declare. Nil when the host
	// never called WithEffectRegistry.
	effects *EffectRegistry
}

// ManagerOption configures a Manager.
type ManagerOption func(*Manager)

// WithRegistryCache makes the Manager serve loaded instances from the registry
// instead of reading fresh from storage on every load. It is a single-process
// optimization: a cached copy can be stale the moment another replica saves,
// so never enable it in multi-replica deployments. The registry grows with
// every distinct instance loaded; long-lived processes should EvictWorkflow
// after finishing with an instance.
func WithRegistryCache() ManagerOption {
	return func(m *Manager) { m.useCache = true }
}

// WithDefinitionMigration installs a handler consulted when a persisted
// instance's definition fingerprint differs from the definition supplied to load
// it. Without it, such a load fails with ErrDefinitionMismatch.
func WithDefinitionMigration(fn DefinitionMigrationFunc) ManagerOption {
	return func(m *Manager) { m.onDefinitionMismatch = fn }
}

// WithEffectRegistry installs the registry that resolves the effects a
// definition's transitions declare (see Transition.SetEffects). Without it,
// declared effects are inert: Execute reports them as unresolvable rather than
// silently skipping writes the definition promised.
//
// Effects declared on the transitions that fired run inside the state-save
// transaction, in declared order, after any effect passed via WithTxSideEffect.
// After-commit effects run once the transaction has committed.
func WithEffectRegistry(reg *EffectRegistry) ManagerOption {
	return func(m *Manager) { m.effects = reg }
}

// NewManager creates a new workflow manager.
func NewManager(registry *Registry, storage Storage, opts ...ManagerOption) *Manager {
	m := &Manager{
		registry: registry,
		storage:  storage,
	}
	for _, opt := range opts {
		opt(m)
	}
	return m
}

// LoadWorkflow loads a workflow instance fresh from storage. With
// WithRegistryCache enabled, a cached instance is returned if present and the
// loaded instance is cached for next time.
func (m *Manager) LoadWorkflow(ctx context.Context, id string, definition *Definition) (*Workflow, error) {
	if m.useCache {
		if wf, err := m.registry.Workflow(id); err == nil {
			// The definition check must hold on cache hits too: a cached
			// instance built from a different definition is exactly as unsafe
			// as loading a persisted one against it.
			if err := checkCachedDefinition(wf, definition); err != nil {
				return nil, err
			}
			return wf, nil
		}
	}

	wf, err := m.loadFromStorage(ctx, id, definition)
	if err != nil {
		return nil, err
	}

	if m.useCache {
		if err := m.registry.AddWorkflow(wf); err != nil {
			return nil, fmt.Errorf("failed to add workflow to registry: %w", err)
		}
	}
	return wf, nil
}

// checkCachedDefinition verifies a registry-cached instance was built from the
// same definition the caller supplied (pointer fast path, fingerprint slow
// path). Unlike a storage load there is no migration path here — the cached
// instance is live, so a mismatch is always an error.
func checkCachedDefinition(wf *Workflow, definition *Definition) error {
	cached := wf.Definition()
	if cached == definition {
		return nil
	}
	if cached != nil && definition != nil && cached.Fingerprint() == definition.Fingerprint() {
		return nil
	}
	return fmt.Errorf("%w: cached instance %q was built from a different definition", ErrDefinitionMismatch, wf.Name())
}

// loadFromStorage builds a workflow instance from persisted state without
// touching the registry. It verifies the definition fingerprint (consulting the
// migration handler on mismatch) and then validates every loaded place against
// the definition. Execute uses it so each retry runs against fresh, validated
// state.
func (m *Manager) loadFromStorage(ctx context.Context, id string, definition *Definition) (*Workflow, error) {
	return m.loadWith(ctx, id, definition, fingerprintMigrate, m.readState)
}

// fingerprintCheck selects how a load reacts to the stored definition
// fingerprint. The ordinary path consults the migration handler; the
// transaction-scoped path cannot, because the handler is host code that may
// write to storage and would deadlock against the transaction we hold — so it
// resolves the question before opening the scope and tells the load which
// answer it got.
type fingerprintCheck int

const (
	// fingerprintMigrate consults the migration handler on a mismatch.
	fingerprintMigrate fingerprintCheck = iota
	// fingerprintStrict makes a mismatch a hard error: it means the definition
	// changed under us after the pre-scope check, not something to approve now.
	fingerprintStrict
	// fingerprintSkip is for a mismatch a handler already approved before the
	// scope opened.
	fingerprintSkip
)

// loadWith is loadFromStorage over a caller-supplied state reader, so the
// transaction-scoped path can read through its own transaction (see
// loadForCycle) while sharing every validation.
//
// check selects how a stored-fingerprint mismatch is handled; see
// fingerprintCheck.
func (m *Manager) loadWith(
	ctx context.Context,
	id string,
	definition *Definition,
	check fingerprintCheck,
	read func(context.Context, string) (Marking, map[string]any, int64, error),
) (*Workflow, error) {
	loaded, wfContext, version, err := read(ctx, id)
	if err != nil {
		return nil, err
	}

	// Verify the definition fingerprint FIRST — before validating places — so a
	// mismatch consults the migration handler even when the stale marking
	// references places the new definition no longer has (the very case
	// migration exists for). After the handler approves, reload: a handler that
	// migrated the persisted state expects the load to observe its rewrite, not
	// clobber it with the pre-migration snapshot on the next save.
	switch check {
	case fingerprintMigrate:
		migrated, merr := m.checkFingerprint(ctx, id, definition, wfContext)
		if merr != nil {
			return nil, merr
		}
		if migrated {
			if loaded, wfContext, version, err = read(ctx, id); err != nil {
				return nil, fmt.Errorf("reloading after definition migration: %w", err)
			}
		}
	case fingerprintStrict:
		if stored, _ := wfContext[defFingerprintKey].(string); stored != definition.Fingerprint() {
			return nil, fmt.Errorf("%w: instance %q stored fingerprint %s, definition fingerprint %s",
				ErrDefinitionMismatch, id, stored, definition.Fingerprint())
		}
	case fingerprintSkip:
		// A handler already approved this mismatch before the scope opened.
	}
	// Strip the fingerprint and shape so they never reach the workflow's
	// user-visible context or guard environment.
	delete(wfContext, defFingerprintKey)
	delete(wfContext, defShapeKey)

	// Validate EVERY loaded place against the definition, not just the first:
	// a stale marking referencing a place removed from the definition must fail
	// loudly rather than load into an instance that can never fire.
	places := loaded.Places()
	for _, p := range places {
		if !definition.Place(p) {
			return nil, fmt.Errorf("%w: loaded marking references place %q not in the definition", ErrDefinitionMismatch, p)
		}
	}

	// A marking with ZERO places is valid: a pure token-pool net's places are
	// all legitimately empty between batches, and its persisted state must
	// round-trip (see NewWorkflowFromMarking).
	var wf *Workflow
	if len(places) == 0 {
		wf, err = NewWorkflowFromMarking(id, definition, loaded)
	} else {
		wf, err = NewWorkflow(id, definition, places[0])
	}
	if err != nil {
		return nil, fmt.Errorf("failed to create workflow: %w", err)
	}
	wf.SetManager(m)
	wf.context = wfContext // Set the loaded context (fingerprint already stripped)
	wf.setVersion(version) // Track the loaded concurrency version (0 if unversioned)

	// Adopt the full loaded marking (preserves colored tokens, not just presence).
	if err := wf.SetMarking(loaded); err != nil {
		return nil, fmt.Errorf("failed to set loaded marking: %w", err)
	}
	return wf, nil
}

// readState loads a workflow's persisted marking, context, and
// optimistic-concurrency version.
func (m *Manager) readState(ctx context.Context, id string) (Marking, map[string]any, int64, error) {
	loaded, wfContext, version, err := m.storage.LoadState(ctx, id)
	if err != nil {
		return nil, nil, 0, fmt.Errorf("failed to load workflow state: %w", err)
	}
	return loaded, wfContext, version, nil
}

// checkFingerprint compares the stored definition fingerprint against the
// supplied definition. A missing stored fingerprint (pre-fingerprint instance)
// passes... to the migration handler, like any mismatch. A mismatch is an
// error unless a migration handler approves it; the returned bool reports
// whether a handler was consulted and approved (the caller then reloads
// state, since the handler may have rewritten it).
func (m *Manager) checkFingerprint(ctx context.Context, id string, definition *Definition, wfContext map[string]any) (migrated bool, err error) {
	stored, ok := wfContext[defFingerprintKey].(string)
	if !ok || stored == "" {
		// Every save stamps the fingerprint, so a persisted instance without
		// one was not written by this library's save path. Treat it exactly
		// like a mismatch: fail loudly, or route to the migration handler
		// (which receives an empty StoredFingerprint).
		stored = ""
	}
	current := definition.Fingerprint()
	if stored == current {
		return false, nil
	}
	if m.onDefinitionMismatch != nil {
		mismatch := DefinitionMismatch{
			WorkflowID:         id,
			StoredFingerprint:  stored,
			CurrentFingerprint: current,
			Diff:               storedShapeDiff(wfContext, definition),
		}
		if err := m.onDefinitionMismatch(ctx, mismatch); err != nil {
			return false, err
		}
		return true, nil
	}
	return false, fmt.Errorf("%w: instance %q stored fingerprint %s, definition fingerprint %s", ErrDefinitionMismatch, id, stored, current)
}

// storedShapeDiff parses the shape stamped on the instance's last save and
// diffs it against the supplied definition. It returns nil when the stored
// state carries no (parseable) shape — pre-shape instances, or rows not
// written by this library — leaving the mismatch's Diff as "no information".
func storedShapeDiff(wfContext map[string]any, definition *Definition) *DefinitionDiff {
	raw, ok := wfContext[defShapeKey].(string)
	if !ok || raw == "" {
		return nil
	}
	var stored definitionShape
	if err := json.Unmarshal([]byte(raw), &stored); err != nil {
		return nil
	}
	return diffShapes(stored, definition.shape())
}

// contextForSave returns a copy of ctxData stamped with the definition
// fingerprint and structural shape, ready to hand to storage. The live
// workflow context is never mutated (the caller passes a snapshot copy).
func contextForSave(ctxData map[string]any, definition *Definition) map[string]any {
	if ctxData == nil {
		ctxData = make(map[string]any, 2)
	}
	if definition != nil {
		ctxData[defFingerprintKey] = definition.Fingerprint()
		if shape, err := json.Marshal(definition.shape()); err == nil {
			ctxData[defShapeKey] = string(shape)
		}
	}
	return ctxData
}

// SaveWorkflow saves a workflow instance state to storage.
//
// The save is guarded by the workflow's current version: if another writer
// saved first, it returns ErrConflict and the workflow's version is left
// unchanged so the caller can reload and retry.
func (m *Manager) SaveWorkflow(ctx context.Context, id string, wf *Workflow) error {
	// Snapshot marking and context under the workflow's lock: handing the live
	// state to the storage layer would race concurrent transitions and
	// SetContext calls while it marshals.
	marking, ctxData := wf.snapshotState()
	ctxData = contextForSave(ctxData, wf.Definition())
	due := m.dueForSave(wf.Definition(), marking)
	newVersion, err := m.persistState(ctx, id, marking, ctxData, wf.Version(), due)
	if err != nil {
		return err
	}
	wf.setVersion(newVersion)
	return nil
}

// dueForSave computes an instance's next-due wall-clock time from the same
// marking snapshot that is about to be persisted, returning nil when no timer is
// running. Deriving it from the persisted snapshot (rather than the live
// workflow) keeps the stored due index consistent with the stored marking.
//
// It returns nil immediately when the backend does not maintain a due index (not
// a DueStorage), avoiding a wasted deadline scan on every save for backends that
// would ignore the result anyway.
func (m *Manager) dueForSave(definition *Definition, marking Marking) *time.Time {
	if _, ok := m.storage.(DueStorage); !ok {
		return nil
	}
	if t, ok := nextDue(definition, marking); ok {
		return &t
	}
	return nil
}

// persistState saves a marking and context, always maintaining the due index
// when the backend is a DueStorage — so the index can never go stale, whatever
// save path a caller takes. It returns the new version.
func (m *Manager) persistState(ctx context.Context, id string, marking Marking, ctxData map[string]any, expectedVersion int64, due *time.Time) (int64, error) {
	if ds, ok := m.storage.(DueStorage); ok {
		return ds.SaveStateWithDue(ctx, id, marking, ctxData, expectedVersion, due)
	}
	return m.storage.SaveState(ctx, id, marking, ctxData, expectedVersion)
}

// ExecuteOption configures a single Manager.Execute call.
type ExecuteOption func(*executeConfig)

type executeConfig struct {
	effects     []TxSideEffect
	maxAttempts int

	// firedEffects are FireDue-scoped tx side effects (WithFireDueTxSideEffect):
	// like effects, but each receives the steps the FireDue pass fired.
	// firedSteps is the provider FireDue installs so the wrapped effects can
	// read the current attempt's steps at commit time; it is nil under plain
	// Execute, which rejects firedEffects rather than silently dropping the
	// payload.
	firedEffects []FireDueTxSideEffect
	firedSteps   func() []FiredStep
}

// WithMaxRetries sets how many times Execute retries the whole
// load-fn-save cycle when the save hits an optimistic-concurrency conflict.
// The default is 5. Raise it for instances with many concurrent writers.
func WithMaxRetries(attempts int) ExecuteOption {
	return func(c *executeConfig) {
		if attempts > 0 {
			c.maxAttempts = attempts
		}
	}
}

// WithTxSideEffect registers a write committed atomically with the state save —
// the crash-consistent way to append an audit/history record or an outbox row
// for a firing. Requires the Manager's storage to implement
// TransactionalStorage (the SQLite and Postgres backends do); Execute fails
// otherwise rather than silently losing atomicity.
//
// Effects registered here differ from listener side effects: an effect shares
// the save's transaction (state and effect commit or roll back together), while
// a listener's external side effects happen immediately and are not undone by a
// later conflict or crash. The canonical use is a history record built inside
// fn (capturing the fired transition) and written by the effect:
//
//	var record *history.TransitionRecord
//	err := mgr.Execute(ctx, id, def,
//	    func(wf *workflow.Workflow) error {
//	        if err := wf.ApplyTransition("approve"); err != nil {
//	            return err
//	        }
//	        record = &history.TransitionRecord{WorkflowID: id, Transition: "approve", ...}
//	        return nil
//	    },
//	    workflow.WithTxSideEffect(func(ctx context.Context, tx any) error {
//	        return hist.SaveTransitionTx(ctx, tx.(*sql.Tx), record)
//	    }),
//	)
func WithTxSideEffect(effect TxSideEffect) ExecuteOption {
	return func(c *executeConfig) { c.effects = append(c.effects, effect) }
}

// FiredStep describes one timer firing inside a FireDue pass: the transition
// that fired and the marking on each side of it — what a host needs to write
// an audit/history record for the firing.
type FiredStep struct {
	Transition string
	Before     []Place
	After      []Place
}

// FireDueTxSideEffect is a write committed atomically with a FireDue save,
// receiving the steps the pass fired. See WithFireDueTxSideEffect.
type FireDueTxSideEffect func(ctx context.Context, tx any, steps []FiredStep) error

// WithFireDueTxSideEffect registers a write committed atomically with the
// FireDue save that receives the transitions the pass fired — the
// exactly-once way to record timer firings. A plain WithTxSideEffect cannot
// serve here: FireDue returns the fired names only after its save commits, so
// a host writing history from the return value has an at-least-once crash
// window between the two. This effect runs inside the save's transaction with
// the fired steps in hand, so the state change and its audit record commit or
// roll back together, exactly like an interactive Execute fire:
//
//	fired, err := mgr.FireDue(ctx, id, def, now,
//	    workflow.WithFireDueTxSideEffect(func(ctx context.Context, tx any, steps []workflow.FiredStep) error {
//	        for _, s := range steps {
//	            if err := hist.SaveTransitionTx(ctx, tx.(*sql.Tx), &history.TransitionRecord{
//	                WorkflowID: id, Transition: s.Transition, Actor: "timer", CreatedAt: now,
//	            }); err != nil {
//	                return err
//	            }
//	        }
//	        return nil
//	    }))
//
// The effect is skipped when the pass fired nothing (a due-index self-heal
// save has no steps to record). Like WithTxSideEffect it requires a
// TransactionalStorage backend, and it is only meaningful under FireDue —
// plain Execute rejects it, since only FireDue knows what fired.
func WithFireDueTxSideEffect(effect FireDueTxSideEffect) ExecuteOption {
	return func(c *executeConfig) { c.firedEffects = append(c.firedEffects, effect) }
}

// Execute atomically advances a persisted instance: it loads the instance fresh
// from storage, runs fn against it (fn typically applies one or more
// transitions), and saves it back under optimistic concurrency. If the save
// conflicts with a concurrent writer (ErrConflict), the whole cycle retries on
// fresh state up to a bounded number of times.
//
// This is the recommended way to react to an external event (a webhook, a UI
// action, a timer) because it removes the load / fire / save / retry boilerplate
// and always operates on current state — even with registry caching disabled or
// across replicas. fn must be safe to re-run, since a conflict re-invokes it on
// reloaded state; a transition that is no longer enabled on reload returns
// ErrNotEnabled from within fn, which Execute surfaces to the caller.
//
// RETRY RESETS: any variable fn's closure writes (fired-transition names,
// step records, before/after markings) must be RE-INITIALIZED at the top of
// fn — a retry otherwise appends to the previous attempt's values, and the
// bug stays silent until real concurrency triggers a conflict:
//
//	var fired []string
//	err := mgr.Execute(ctx, id, def, func(wf *workflow.Workflow) error {
//	    fired = nil // reset: Execute may re-run fn on ErrConflict
//	    if err := wf.ApplyTransition("approve"); err != nil {
//	        return err
//	    }
//	    fired = append(fired, "approve")
//	    return nil
//	})
//
// Side-effect semantics: fn (and any listeners it triggers) may run more than
// once across retries, and a listener's external side effects are not rolled
// back by a later conflict — make them idempotent, or perform them after
// Execute returns. Writes that must be crash-consistent with the state change
// (history records, outbox rows) belong in a WithTxSideEffect option, which
// commits them in the same transaction as the save. See
// docs/guides/PRODUCTION_RECIPES.md for the full set of retry/crash recipes.
func (m *Manager) Execute(ctx context.Context, id string, definition *Definition, fn func(*Workflow) error, opts ...ExecuteOption) error {
	var cfg executeConfig
	for _, opt := range opts {
		opt(&cfg)
	}

	// FireDue-scoped effects are only meaningful when a steps provider was
	// installed (by FireDue itself): plain Execute has no fired steps to hand
	// them, so reject rather than silently invoke with nothing. Wrapping them
	// into plain effects BEFORE the transactional-support check keeps the
	// atomicity requirement applying to them too.
	if len(cfg.firedEffects) > 0 {
		if cfg.firedSteps == nil {
			return fmt.Errorf("WithFireDueTxSideEffect is only valid with FireDue: %w", errors.ErrUnsupported)
		}
		for _, fe := range cfg.firedEffects {
			cfg.effects = append(cfg.effects, func(ctx context.Context, tx any) error {
				steps := cfg.firedSteps()
				if len(steps) == 0 {
					return nil // a self-heal save with no firing needs no record
				}
				return fe(ctx, tx, steps)
			})
		}
	}

	// Atomic side effects need transactional support; fail loudly rather than
	// silently dropping the atomicity guarantee.
	// Declared effects need the same transactional storage as WithTxSideEffect.
	// Whether the definition declares any is knowable up front, before anything
	// fires, so the requirement is checked here rather than mid-transaction.
	declaresEffects := definitionDeclaresEffects(definition)
	if declaresEffects && m.effects == nil {
		return fmt.Errorf("definition declares transition effects but the Manager has no registry: %w "+
			"(build it with workflow.WithEffectRegistry)", errors.ErrUnsupported)
	}

	var ts TransactionalStorage
	var tds TransactionalDueStorage
	if len(cfg.effects) > 0 || declaresEffects {
		var ok bool
		if ts, ok = m.storage.(TransactionalStorage); !ok {
			return fmt.Errorf("atomic side effects (WithTxSideEffect or declared transition effects) "+
				"require a TransactionalStorage backend: %w", errors.ErrUnsupported)
		}
		// A backend that maintains a due index but cannot update it inside the
		// state+effect transaction would leave the index silently corrupt for a
		// timed definition (state and effect commit, but the due column is not
		// touched). Fail loudly rather than drift — mirroring the missing-
		// TransactionalStorage error above.
		if _, isDue := m.storage.(DueStorage); isDue {
			if tds, ok = m.storage.(TransactionalDueStorage); !ok && definitionHasTimers(definition) {
				return fmt.Errorf("WithTxSideEffect on a DueStorage backend with a timed definition requires a TransactionalDueStorage backend "+
					"(implement SaveStateInTxWithDue so the due index commits atomically with state and effects): %w", errors.ErrUnsupported)
			}
		}
	}

	// Transaction-scoped guards invert the usual order: instead of firing in
	// memory and opening a transaction at the save, the WHOLE cycle runs inside
	// one transaction, so a guard can read host state as of the state it is
	// about to commit against. It needs a backend that can lend out its
	// transaction; say so loudly rather than evaluating the guard on stale data.
	plan := executePlan{cfg: &cfg, declaresEffects: declaresEffects, ts: ts, tds: tds}
	if definitionHasTxGuards(definition) {
		var ok bool
		if plan.tss, ok = m.storage.(TxScopedStorage); !ok {
			return fmt.Errorf("definition has transaction-scoped guards, which require a TxScopedStorage backend "+
				"(one that can run the whole load-fire-save cycle in a transaction it lends out): %w", errors.ErrUnsupported)
		}
		// Same reasoning as the TransactionalDueStorage check above: committing
		// state without updating the due column in the same transaction leaves
		// the index silently corrupt.
		if _, isDue := m.storage.(DueStorage); isDue {
			if plan.tsds, ok = m.storage.(TxScopedDueStorage); !ok && definitionHasTimers(definition) {
				return fmt.Errorf("transaction-scoped guards on a DueStorage backend with a timed definition require "+
					"a TxScopedDueStorage backend (implement SaveStateScopedWithDue): %w", errors.ErrUnsupported)
			}
		}
		// The migration handler is host code that may write to storage, which
		// would deadlock against the transaction we are about to open on the
		// same rows. Consult it BEFORE the scope; in the common case (no handler
		// registered) this costs nothing, because the in-scope check catches a
		// mismatch identically.
		plan.check = fingerprintStrict
		if m.onDefinitionMismatch != nil {
			approved, err := m.migrateBeforeScope(ctx, id, definition)
			if err != nil {
				return err
			}
			if approved {
				plan.check = fingerprintSkip
			}
		}
	}

	maxAttempts := cfg.maxAttempts
	if maxAttempts <= 0 {
		maxAttempts = 5
	}
	var lastErr error
	for attempt := range maxAttempts {
		// A conflict storm must not outlive the caller: honor cancellation even
		// if the storage backend ignores ctx.
		if err := ctx.Err(); err != nil {
			return err
		}
		// Back off with jitter before each retry so N concurrent writers on one
		// instance de-synchronize instead of starving each other out.
		if attempt > 0 {
			jitter := time.Duration(rand.Int64N(int64(4 * time.Millisecond)))
			time.Sleep(time.Duration(attempt)*2*time.Millisecond + jitter)
		}

		afterCommit, err := m.runCycle(ctx, id, definition, fn, &plan)
		if err != nil {
			if errors.Is(err, ErrConflict) {
				lastErr = err
				continue // reload fresh and retry
			}
			return err
		}

		// Keep the registry from serving a now-stale cached copy.
		if m.useCache {
			_ = m.registry.RemoveWorkflow(id)
		}
		// The state transaction has committed. After-commit effects run now,
		// deliberately outside it — and so AT-LEAST-ONCE: a crash here loses
		// them, and an error from one is returned without undoing the commit.
		return runAfterCommit(ctx, afterCommit)
	}
	return fmt.Errorf("%w: giving up after %d attempts", lastErr, maxAttempts)
}

// executePlan is the per-Execute decision about how one attempt persists: which
// optional storage capabilities were resolved, and whether the definition needs
// the whole cycle wrapped in a transaction. It is computed once, before the
// retry loop, so every attempt behaves identically.
type executePlan struct {
	cfg             *executeConfig
	declaresEffects bool

	ts  TransactionalStorage
	tds TransactionalDueStorage

	// tss is non-nil when the definition has transaction-scoped guards; tsds is
	// additionally non-nil when the backend also maintains the due index.
	tss  TxScopedStorage
	tsds TxScopedDueStorage

	// check is how the in-scope load treats a stored-fingerprint mismatch.
	check fingerprintCheck
}

// txScope is an open transaction that one cycle runs inside, or nil for the
// ordinary path where the fire happens in memory and the transaction opens only
// at the save.
type txScope struct {
	tss  TxScopedStorage
	tsds TxScopedDueStorage
	tx   any

	// check is how the in-scope load treats the stored fingerprint, decided
	// before the scope opened (see migrateBeforeScope).
	check fingerprintCheck
}

// runCycle performs one load → fire → save attempt, opening a transaction
// around the whole of it when the definition has transaction-scoped guards. It
// returns the after-commit effects the caller must run once the state has
// actually committed.
func (m *Manager) runCycle(
	ctx context.Context,
	id string,
	definition *Definition,
	fn func(*Workflow) error,
	p *executePlan,
) ([]pendingAfterCommit, error) {
	if p.tss == nil {
		return m.cycle(ctx, nil, id, definition, fn, p)
	}

	var afterCommit []pendingAfterCommit
	err := p.tss.BeginScope(ctx, func(ctx context.Context, tx any) error {
		var err error
		afterCommit, err = m.cycle(ctx, &txScope{tss: p.tss, tsds: p.tsds, tx: tx, check: p.check}, id, definition, fn, p)
		return err
	})
	if err != nil {
		// The scope rolled back, so nothing this attempt produced was committed
		// — including the after-commit effects it had resolved.
		return nil, err
	}
	return afterCommit, nil
}

// cycle is the body of one attempt. sc is nil on the ordinary path; when it is
// set, the load, the guards, the save, and the effects all run inside sc.tx.
func (m *Manager) cycle(
	ctx context.Context,
	sc *txScope,
	id string,
	definition *Definition,
	fn func(*Workflow) error,
	p *executePlan,
) ([]pendingAfterCommit, error) {
	wf, err := m.loadForCycle(ctx, sc, id, definition)
	if err != nil {
		return nil, err
	}

	if sc != nil {
		// Bind for the duration of the fire only. A guard reached through a
		// workflow value that outlived the scope must not find a committed
		// transaction waiting for it.
		wf.bindTx(sc.tx)
		defer wf.bindTx(nil)
	}
	if err := fn(wf); err != nil {
		return nil, err
	}

	marking, ctxData := wf.snapshotState()
	ctxData = contextForSave(ctxData, definition)
	due := m.dueForSave(definition, marking)

	// Resolve the effects declared by whatever actually fired this attempt.
	// Draining after fn (and after the snapshot) means a retried attempt
	// starts from an empty log, so an abandoned attempt's effects can never
	// leak into the one that commits.
	var afterCommit []pendingAfterCommit
	effects := p.cfg.effects
	if p.declaresEffects {
		steps := wf.drainFired()
		txEffects, pending, rerr := m.resolveEffects(id, definition, steps, ctxData)
		if rerr != nil {
			return nil, rerr
		}
		effects = append(append([]TxSideEffect(nil), p.cfg.effects...), txEffects...)
		afterCommit = pending
	}

	switch {
	case sc != nil:
		// Save into the very transaction the guards read from, then run the
		// effects in it too. Same atomicity guarantee as SaveStateInTx — the
		// difference is only who opened the transaction.
		if sc.tsds != nil {
			_, err = sc.tsds.SaveStateScopedWithDue(ctx, sc.tx, id, marking, ctxData, wf.Version(), due)
		} else {
			_, err = sc.tss.SaveStateScoped(ctx, sc.tx, id, marking, ctxData, wf.Version())
		}
		for _, effect := range effects {
			if err != nil {
				break
			}
			err = effect(ctx, sc.tx)
		}
	case p.ts != nil:
		// Keep the due index current even on the transactional path (state +
		// side effect commit together) when the backend supports it. A partial
		// due backend (DueStorage but not TransactionalDueStorage) with a timed
		// definition was already rejected before the loop.
		if p.tds != nil {
			_, err = p.tds.SaveStateInTxWithDue(ctx, id, marking, ctxData, wf.Version(), due, effects...)
		} else {
			_, err = p.ts.SaveStateInTx(ctx, id, marking, ctxData, wf.Version(), effects...)
		}
	default:
		_, err = m.persistState(ctx, id, marking, ctxData, wf.Version(), due)
	}
	if err != nil {
		return nil, err
	}
	return afterCommit, nil
}

// loadForCycle loads the instance for one attempt, reading through the scope's
// transaction when there is one so the marking and version a firing decides on
// come from the same snapshot its guards read.
func (m *Manager) loadForCycle(ctx context.Context, sc *txScope, id string, definition *Definition) (*Workflow, error) {
	if sc == nil {
		return m.loadFromStorage(ctx, id, definition)
	}
	return m.loadWith(ctx, id, definition, sc.check, func(ctx context.Context, id string) (Marking, map[string]any, int64, error) {
		loaded, wfContext, version, err := sc.tss.LoadStateScoped(ctx, sc.tx, id)
		if err != nil {
			return nil, nil, 0, fmt.Errorf("failed to load workflow state: %w", err)
		}
		return loaded, wfContext, version, nil
	})
}

// migrateBeforeScope runs the definition-fingerprint check, and therefore the
// migration handler, outside any transaction of ours. It reports whether a
// mismatch was found AND approved, so the in-scope load knows not to re-check
// against a fingerprint the handler deliberately accepted. See the call site.
func (m *Manager) migrateBeforeScope(ctx context.Context, id string, definition *Definition) (bool, error) {
	_, wfContext, _, err := m.readState(ctx, id)
	if err != nil {
		return false, err
	}
	return m.checkFingerprint(ctx, id, definition, wfContext)
}

// GetWorkflow gets a workflow instance, loading it fresh from storage (or from
// the registry when WithRegistryCache is enabled). It is an alias of
// LoadWorkflow.
func (m *Manager) GetWorkflow(ctx context.Context, id string, definition *Definition) (*Workflow, error) {
	return m.LoadWorkflow(ctx, id, definition)
}

// CreateWorkflow creates a new workflow instance and saves it to storage
func (m *Manager) CreateWorkflow(ctx context.Context, id string, definition *Definition, initialPlace Place) (*Workflow, error) {
	wf, err := NewWorkflow(id, definition, initialPlace)
	if err != nil {
		return nil, fmt.Errorf("failed to create workflow: %w", err)
	}
	return m.saveCreated(ctx, id, definition, wf)
}

// CreateWorkflowFromMarking creates and saves a new workflow instance whose
// starting state is the given marking: several places, data-carrying (colored)
// tokens, or — for a pure token-pool net — no marked places at all (an empty
// marking is valid and persists as such; see NewWorkflowFromMarking).
func (m *Manager) CreateWorkflowFromMarking(ctx context.Context, id string, definition *Definition, initial Marking) (*Workflow, error) {
	wf, err := NewWorkflowFromMarking(id, definition, initial)
	if err != nil {
		return nil, fmt.Errorf("failed to create workflow: %w", err)
	}
	return m.saveCreated(ctx, id, definition, wf)
}

// saveCreated persists a freshly constructed instance. The save inserts at
// version 1 and fails with ErrConflict if a workflow with this id already
// exists. A timer-bearing workflow's initial marking is already stamped, so
// its first deadline is indexed from creation.
func (m *Manager) saveCreated(ctx context.Context, id string, definition *Definition, wf *Workflow) (*Workflow, error) {
	wf.SetManager(m)

	marking, ctxData := wf.snapshotState()
	ctxData = contextForSave(ctxData, definition)
	due := m.dueForSave(definition, marking)
	newVersion, err := m.persistState(ctx, id, marking, ctxData, 0, due)
	if err != nil {
		return nil, fmt.Errorf("failed to save initial state: %w", err)
	}
	wf.setVersion(newVersion)

	if m.useCache {
		if err := m.registry.AddWorkflow(wf); err != nil {
			return nil, fmt.Errorf("failed to add workflow to registry: %w", err)
		}
	}
	return wf, nil
}

// DeleteWorkflow removes a workflow instance and its state
func (m *Manager) DeleteWorkflow(ctx context.Context, id string) error {
	// Remove from registry (ignore error if workflow not found)
	_ = m.registry.RemoveWorkflow(id)

	// Remove from storage
	return m.storage.DeleteState(ctx, id)
}

// EvictWorkflow drops a workflow from the in-memory registry cache without
// touching storage. Call it after saving when a long-lived Manager would
// otherwise accumulate instances or serve a stale cached copy; a subsequent
// GetWorkflow/LoadWorkflow then reads fresh from storage. With registry caching
// disabled it is a no-op.
func (m *Manager) EvictWorkflow(id string) {
	_ = m.registry.RemoveWorkflow(id)
}

// ListWorkflowIDs returns the IDs of persisted workflows, ordered by ID, using
// the backend's ListableStorage support. It returns an error wrapping
// errors.ErrUnsupported if the storage backend does not implement
// ListableStorage.
func (m *Manager) ListWorkflowIDs(ctx context.Context, opts ListOptions) ([]string, error) {
	ls, ok := m.storage.(ListableStorage)
	if !ok {
		return nil, fmt.Errorf("storage backend does not implement ListableStorage: %w", errors.ErrUnsupported)
	}
	return ls.ListIDs(ctx, opts)
}

// ListPlaceTokens returns every token currently resting in the given place
// across ALL persisted workflow instances — the cross-instance read-model for
// shared token pools (e.g. "every payable expense in the system"), answered
// by one indexed query instead of loading every instance. It requires the
// backend's TokenQueryStorage support and returns an error wrapping
// errors.ErrUnsupported otherwise.
func (m *Manager) ListPlaceTokens(ctx context.Context, place Place, opts ListOptions) ([]PlacedToken, error) {
	ts, ok := m.storage.(TokenQueryStorage)
	if !ok {
		return nil, fmt.Errorf("storage backend does not implement TokenQueryStorage: %w", errors.ErrUnsupported)
	}
	return ts.ListPlaceTokens(ctx, place, opts)
}

// ListDue returns the IDs of persisted instances whose next-due time is at or
// before `before`, ordered by due time ascending — the instances a host cron
// should advance with FireDue. A zero limit means no limit; drain a batch with
// FireDue before rescanning, or page by raising `before`.
//
// It requires a DueStorage backend (the SQLite and Postgres backends qualify)
// and returns an error wrapping errors.ErrUnsupported otherwise. This is the
// scan half of the host-driven timer model: pass the host's own clock as
// `before` (typically time.Now) so the whole fleet's deadlines are evaluated
// against one authoritative clock.
func (m *Manager) ListDue(ctx context.Context, before time.Time, limit int) ([]string, error) {
	ds, ok := m.storage.(DueStorage)
	if !ok {
		return nil, fmt.Errorf("storage backend does not implement DueStorage: %w", errors.ErrUnsupported)
	}
	return ds.ListDue(ctx, before, limit)
}

// maxFireDueSteps bounds how many transitions a single FireDue advances — a
// safety valve against a pathological self-re-enabling timer. Because
// SetTimeoutAfter only records positive timeouts, a fired transition's next
// deadline is strictly later than now, so a real definition terminates a firing
// pass long before this bound.
const maxFireDueSteps = 10000

// errFireDueNoSave is an internal sentinel the FireDue fn returns to tell Execute
// "nothing changed worth persisting" — Execute aborts the attempt without saving
// (and without retrying, since it is not ErrConflict), and FireDue translates it
// back to a successful no-op. It is never returned to callers.
var errFireDueNoSave = errors.New("firedue: no save needed")

// FireDue advances a persisted instance by firing every transition whose timer
// has elapsed as of now, returning the names of the transitions that actually
// fired, in firing order.
//
// It is the per-instance half of the host-driven timer model (M4): a host cron
// finds due instances with Manager.ListDue and calls FireDue on each, so a
// fleet-wide "escalate if not approved in 3 days" needs no internal scheduler.
// The state lives in the database and the clock lives in the host, which makes
// the whole mechanism restart-safe by construction.
//
// FireDue loads the instance fresh and pins the workflow clock to now, so tokens
// produced by the firing are stamped with the host's evaluation time and every
// downstream deadline is measured from it (deterministic and testable with a
// fixed clock). It then fires due transitions one at a time, re-evaluating the
// due set after each firing because firing changes the marking; a due transition
// whose guard rejects it — or that an earlier firing in the same pass has since
// disabled — is skipped rather than treated as an error, so only an unexpected
// error aborts.
//
// The save runs under the same optimistic-concurrency retry loop as Execute, so
// several hosts scanning the same fleet cannot clobber each other. FireDue is
// idempotent: once nothing is overdue it fires nothing, and after a firing that
// leaves no running timer the instance drops out of ListDue.
//
// Extra ExecuteOptions (e.g. WithMaxRetries, WithTxSideEffect) are forwarded to
// the underlying Execute; WithFireDueTxSideEffect additionally receives the
// steps this pass fired, inside the save's transaction (see its doc).
func (m *Manager) FireDue(ctx context.Context, id string, definition *Definition, now time.Time, opts ...ExecuteOption) ([]string, error) {
	var fired []string
	var steps []FiredStep
	// Install the steps provider for WithFireDueTxSideEffect: the wrapped
	// effects run after fn within the same attempt, so reading the shared
	// slice hands them exactly this attempt's steps.
	execOpts := make([]ExecuteOption, 0, len(opts)+1)
	execOpts = append(execOpts, opts...)
	execOpts = append(execOpts, func(c *executeConfig) {
		c.firedSteps = func() []FiredStep { return steps }
	})
	err := m.Execute(ctx, id, definition, func(wf *Workflow) error {
		fired = nil // Execute may re-run fn on a conflict retry; start clean.
		steps = nil
		wf.setClock(func() time.Time { return now })
		for range maxFireDueSteps {
			// Fire the first due transition that is actually allowed, then
			// re-evaluate (firing changed the marking). When no due transition
			// fires — the set is empty, or every member is guard-blocked or was
			// disabled earlier in this pass — the pass is done.
			progressed := false
			before := wf.CurrentPlaces()
			for _, t := range wf.Due(now) {
				err := wf.ApplyTransitionWithContext(ctx, t.Name())
				if err != nil {
					if errors.Is(err, ErrTransitionNotAllowed) {
						continue // blocked/disabled: skip, not an error.
					}
					return err
				}
				fired = append(fired, t.Name())
				steps = append(steps, FiredStep{Transition: t.Name(), Before: before, After: wf.CurrentPlaces()})
				progressed = true
				break
			}
			if !progressed {
				break
			}
		}
		if len(fired) == 0 {
			// Nothing fired. Skip the save (no pointless version bump) only when
			// the stored due index already agrees with the live marking — i.e. the
			// instance is legitimately due-but-blocked: a timer is still running and
			// its deadline is at or before now. If instead no timer is running, or
			// the next deadline is in the future, the stored index is stale relative
			// to the marking (e.g. a bypass save), so persist to let it self-heal.
			if next, ok := wf.NextDue(); ok && !next.After(now) {
				return errFireDueNoSave
			}
		}
		return nil
	}, execOpts...)
	if err != nil && !errors.Is(err, errFireDueNoSave) {
		return nil, err
	}
	return fired, nil
}

// AddEventListener adds a dynamic event listener for a specific event type
// It returns a handle that can be used to remove the listener later
func (m *Manager) AddEventListener(eventType EventType, listener EventListener) *ListenerHandle {
	return m.listeners.add(eventType, listener, m)
}

// AddObserver adds a non-blocking observer for a specific event type across
// every workflow this Manager touches: it cannot error, its panics are
// recovered, and it is the only listener kind that receives
// EventGuardRejected. Instrumentation belongs here (see ObserverFunc and
// contrib/otel). It returns a handle for RemoveListener.
func (m *Manager) AddObserver(eventType EventType, observer ObserverFunc) *ListenerHandle {
	return m.listeners.add(eventType, observer, m)
}

// AddGuardEventListener adds a dynamic guard event listener
// It returns a handle that can be used to remove the listener later
func (m *Manager) AddGuardEventListener(listener GuardEventListener) *ListenerHandle {
	return m.listeners.add(EventGuard, listener, m)
}

// RemoveListener removes a listener using its handle
// This is the recommended way to remove listeners as it's reliable and efficient
func (m *Manager) RemoveListener(handle *ListenerHandle) {
	if handle == nil || handle.owner != m {
		return
	}
	m.listeners.remove(handle)
}

// ListenerCount returns the number of listeners registered for eventType.
func (m *Manager) ListenerCount(eventType EventType) int {
	return m.listeners.count(eventType)
}