ProcessManager.hs

-- | Stateful coordination across aggregates: the process manager (saga).
--
-- A 'ProcessManager' reacts to an incoming event by stepping its /own/ state
-- machine — a private \"manager\" event stream, keyed by a correlation id — and,
-- in the same turn, dispatching commands to /target/ aggregates and scheduling
-- timers. It is the stateful counterpart of "Keiro.Router": where a router
-- resolves targets from a read model and holds no state, a process manager
-- folds the events it has seen into durable manager state and decides what to
-- do next from that state.
--
-- Every write — the manager-state append, each dispatched command, each timer
-- — is keyed by a 'deterministicCommandId' derived from
-- @(name, correlation, source event id, emit index)@. The manager pre-checks
-- each id with 'eventAlreadyIn' and folds the store's duplicate rejection into
-- a benign 'PMCommandDuplicate' \/ 'PMStateDuplicate'. Replaying the same
-- source event therefore appends nothing new, which is what makes the worker
-- crash-safe under at-least-once delivery.
--
-- Use 'runProcessManagerOnce' to react to a single event, or
-- 'runProcessManagerWorker' to run the manager as a live subscription over a
-- Shibuya adapter. Worker acks are finalized exactly once: successful and duplicate
-- dispatches ack 'AckOk', transient store failures retry, systemic deterministic
-- failures halt, rejection-class failures follow 'RejectedCommandPolicy', and
-- undecodable messages follow the configured 'PoisonPolicy'.
--
-- === Rejected commands and saga history
--
-- 'RejectedHalt' is the safe default: the subscription stops without advancing,
-- so an operator cannot miss the failure. 'RejectedDeadLetter' instead writes a
-- durable "Keiro.DeadLetter.DispatchDeadLetter" and acknowledges the source
-- event; 'RejectedSkip' acknowledges and records only the metric. Prefer making a
-- target command total in its Keiki transducer (for example, an explicit no-op
-- transition for a benign business rejection) so no policy escape hatch is
-- needed.
--
-- For a process manager, dead-lettering a target dispatch creates an important
-- history split: the manager's own state stream has recorded its reaction, but
-- the target command never applied. Keiro cannot append a generic correction to
-- the manager stream because an event unknown to the manager's transducer would
-- make that stream fail replay. If saga history must reflect the failure, model a
-- domain-specific @DispatchFailed@-style command and event in the manager and
-- drive it from 'Keiro.DeadLetter.listDispatchDeadLetters' through an operator
-- runbook or automation. Otherwise, the dead-letter row is the durable witness.
-- Manager-state rejection itself has no such split because no manager event was
-- appended; its record uses emit index @-1@.
--
-- === Bounded retries and source-event dead letters
--
-- On a Kiroku-backed Shibuya adapter, a transient failure finalizes 'AckRetry',
-- but retries are bounded by the Kiroku subscription's @RetryPolicy@. Its
-- @retryMaxAttempts@ counts total deliveries and defaults to five. When the bound
-- is exhausted, Kiroku records the /source event/ in @kiroku.dead_letters@ with
-- structured reason kind @max_attempts_exceeded@ and atomically advances the
-- checkpoint. The manager will not see that event again unless an operator
-- replays it through @Keiro.DeadLetter.Replay@. Install
-- 'Keiro.Telemetry.kirokuEventBridge' on Kiroku's @eventHandler@ to observe the
-- terminal transition.
--
-- The adapter's @KirokuAdapterConfig@ does not currently expose @retryPolicy@, so
-- that path uses Kiroku's default bound. The sharded path is configurable:
-- 'Keiro.Subscription.Shard.Worker.runShardedSubscriptionGroupAck' forwards
-- 'Keiro.Subscription.Shard.Worker.ShardedWorkerOptions.retryPolicy' into the
-- same Kiroku acknowledgement ladder. A shard handler that dispatches commands
-- can use 'decideForFailures' and 'isRejectionClass' to classify outcomes, then
-- map the decision to its existing @ShardAck@ reply.
--
-- === Correlation, ordering, and transaction boundaries
--
-- Events from one originating stream retain that stream's append order. Kiroku
-- consumer groups hash the originating stream id to a stable member, so sharding
-- does not split one stream across readers. Events from /different/ streams that
-- 'correlate' to the same manager instance have no business-order guarantee,
-- however. Their append transactions race for global positions and, under
-- sharding, the streams may be processed concurrently by different members. A
-- retry on one member does not stop another member from advancing.
--
-- For example, both @payment-ORD1@'s @PaymentCaptured@ and @shipment-ORD1@'s
-- @ShipmentAllocated@ may correlate to order @ORD1@. The manager must accept
-- @PaymentCaptured@ then @ShipmentAllocated@ /and/ the reverse, normally with
-- states that record one fact while waiting for the other. An unsharded Kiroku
-- subscription processes its observed global order serially, including retries,
-- but that order still reflects append timing rather than a domain sequence.
--
-- Rule of thumb: 'correlate' may join streams freely, but every such join must be
-- order-insensitive. When a strict sequence is required, enforce it in the
-- manager's own state machine — for example with an explicit no-op/waiting state
-- and a timer-driven retry, or a modeled rejection handled by the dead-letter
-- policy — never by assuming delivery order.
--
-- The persistence boundary is also intentionally smaller than one whole saga
-- reaction. A manager event and its timers commit together when the manager
-- command appends; a timer-only no-op reaction schedules its timers in a separate
-- transaction. Each target command and its inline projections then commit in
-- their own transaction. A crash or rejected target can therefore leave durable
-- manager history without every target write. Deterministic ids make source-event
-- replay fill the missing writes without duplicating completed ones; the rejected
-- command section above describes the case that is deliberately acknowledged
-- instead.
module Keiro.ProcessManager
  ( -- * Definition
    ProcessManager (..),
    ProcessManagerAction (..),
    PMCommand (..),

    -- * Results
    ProcessManagerResult (..),
    PMCommandResult (..),
    PMStateResult (..),

    -- * Running
    PoisonPolicy (..),
    RejectedCommandPolicy (..),
    DispatchFailure (..),
    WorkerOptions (..),
    defaultWorkerOptions,
    isTransientStoreError,
    isTransientCommandError,
    isRejectionClass,
    decideForFailures,
    ackForCommandError,
    runProcessManagerOnce,
    runProcessManagerWorkerWith,
    runProcessManagerWorker,

    -- * Idempotency primitives
    deterministicCommandId,
    eventAlreadyIn,
    confirmBenignDuplicate,
  )
where

import Data.Coerce (coerce)
import Data.Text qualified as Text
import Data.UUID qualified as UUID
import Data.UUID.V5 qualified as UUID.V5
import Effectful (Eff, IOE, (:>))
import Effectful.Error.Static (Error, tryError)
import GHC.Stack (HasCallStack)
import Keiki.Core (BoolAlg, RegFile)
import Keiro.Command (CommandError (..), CommandResult, RunCommandOptions, commandErrorClass, runCommandWithSql)
import Keiro.DeadLetter (DispatchDeadLetter (..), DispatcherKind (..), recordDispatchDeadLetter)
import Keiro.EventStream (EventStream)
import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated)
import Keiro.Prelude
import Keiro.Projection (InlineProjection, runCommandWithProjections)
import Keiro.Stream (Stream)
import Keiro.Telemetry (KeiroMetrics, recordDispatchDeadLettered, recordDispatchDuplicate, recordDispatchFailed, recordDispatchPoison)
import Keiro.Timer (TimerRequest, scheduleTimerTx)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Effect.Resource (KirokuStoreResource)
import Kiroku.Store.Error (StoreError (..))
import Kiroku.Store.Read (eventExistsInStream)
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (EventId (..), RecordedEvent)
import Kiroku.Store.Types qualified as StoreTypes
import Shibuya.Adapter (Adapter (..))
import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..), RetryDelay (..))
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (Ingested (..))
import Shibuya.Core.Types (Attempt (..), Envelope (..))
import Streamly.Data.Fold qualified as Fold
import Streamly.Data.Stream qualified as Streamly
import Prelude (any, filter, fromIntegral, length, not, uncurry, zip, (&&), (+))

-- | A process manager wiring together a manager state machine and the target
-- aggregate it drives.
--
-- * 'name' — stable identity; part of every deterministic write id.
-- * 'correlate' — derives the correlation key for an input event, selecting
--   which manager instance handles it.
-- * 'eventStream' — the manager's own 'EventStream'; its events record the
--   saga's progress.
-- * 'streamFor' — maps a correlation key to the manager's 'Stream' handle.
-- * 'targetEventStream' — the aggregate that dispatched commands are sent to.
-- * 'targetProjections' — inline projections for the target aggregate, run in
--   the same transaction as each dispatched command's append. Return @[]@ for
--   append-only dispatch.
-- * 'handle' — the pure reaction: given an input event, produce the
--   manager-state command, the target commands to dispatch, and any timers to
--   schedule.
data ProcessManager input phi rs s ci co targetPhi targetRs targetState targetCi targetCo = ProcessManager
  { name :: !Text,
    correlate :: !(input -> Text),
    eventStream :: !(ValidatedEventStream phi rs s ci co),
    streamFor :: !(Text -> Stream (EventStream phi rs s ci co)),
    targetEventStream :: !(ValidatedEventStream targetPhi targetRs targetState targetCi targetCo),
    -- | Inline projections for the target aggregate, run in the same transaction
    --     as each dispatched command's append. Return @[]@ for append-only dispatch.
    targetProjections :: !(Stream targetCi -> [InlineProjection targetCo]),
    handle :: !(input -> ProcessManagerAction ci targetCi)
  }
  deriving stock (Generic)

-- | What a process manager decides to do for one input event: advance its own
-- state with 'command', dispatch zero or more target 'commands', and schedule
-- zero or more 'timers'. All three are applied atomically with crash-safe
-- idempotency by 'runProcessManagerOnce'.
data ProcessManagerAction ci targetCi = ProcessManagerAction
  { command :: !ci,
    commands :: ![PMCommand targetCi],
    timers :: ![TimerRequest]
  }
  deriving stock (Generic)

-- | A single command addressed to a specific target stream.
data PMCommand targetCi = PMCommand
  { target :: !(Stream targetCi),
    command :: !targetCi
  }
  deriving stock (Generic, Eq, Show)

-- | Outcome of one dispatched target command.
data PMCommandResult target
  = -- | The command appended events (carries the 'CommandResult').
    PMCommandAppended !(CommandResult target)
  | -- | The command was already applied (idempotent replay); carries the
    --       deterministic id that already existed.
    PMCommandDuplicate !EventId
  | -- | The command failed in the named target stream; the worker classifies
    --       the error. Transient failures retry, rejection-class failures follow
    --       'RejectedCommandPolicy', and systemic deterministic failures halt.
    PMCommandFailed !StoreTypes.StreamName !CommandError
  deriving stock (Generic, Eq, Show)

-- | Outcome of the manager's own state append. Unlike 'PMCommandResult' there
-- is no failure case — a manager-state append that genuinely errors aborts the
-- whole reaction via an outer @Left@ 'CommandError'.
data PMStateResult target
  = PMStateAppended !(CommandResult target)
  | PMStateDuplicate !EventId
  deriving stock (Generic, Eq, Show)

-- | The complete result of reacting to one event: how the manager state
-- advanced, the outcome of each dispatched command in order, and how many
-- timers were scheduled.
data ProcessManagerResult managerTarget commandTarget = ProcessManagerResult
  { managerResult :: !(PMStateResult managerTarget),
    commandResults :: ![PMCommandResult commandTarget],
    timersScheduled :: !Int
  }
  deriving stock (Generic, Eq, Show)

-- | What a worker does with a message its decoder cannot parse.
data PoisonPolicy es msg
  = PoisonHalt
  | PoisonSkip !(Envelope msg -> Eff es ())
  | PoisonDeadLetter !(Envelope msg -> Eff es ())

-- | What a worker does when every failed dispatch is a rejection-class error.
data RejectedCommandPolicy
  = -- | Halt without acknowledging so the source event replays. This is the default.
    RejectedHalt
  | -- | Persist a durable dispatch dead letter and acknowledge the source event.
    RejectedDeadLetter
  | -- | Acknowledge and count the rejection without persisting a record.
    RejectedSkip
  deriving stock (Generic, Eq, Show)

-- | One failed dispatch with the target identity needed by worker policy.
data DispatchFailure = DispatchFailure
  { emitIndex :: !Int,
    targetStreamName :: !StoreTypes.StreamName,
    commandError :: !CommandError
  }
  deriving stock (Generic, Eq, Show)

-- | Worker-level knobs shared by the process-manager and router workers.
data WorkerOptions es msg = WorkerOptions
  { poisonPolicy :: !(PoisonPolicy es msg),
    rejectedCommandPolicy :: !RejectedCommandPolicy,
    transientRetryDelay :: !RetryDelay,
    metrics :: !(Maybe KeiroMetrics)
  }
  deriving stock (Generic)

defaultWorkerOptions :: WorkerOptions es msg
defaultWorkerOptions =
  WorkerOptions
    { poisonPolicy = PoisonHalt,
      rejectedCommandPolicy = RejectedHalt,
      transientRetryDelay = RetryDelay 5,
      metrics = Nothing
    }

isTransientStoreError :: StoreError -> Bool
isTransientStoreError = \case
  ConnectionLost {} -> True
  PoolAcquisitionTimeout -> True
  ConnectionError {} -> True
  WrongExpectedVersion {} -> True
  StreamAlreadyExists {} -> True
  EmptyAppendBatch {} -> False
  StreamNotFound {} -> False
  ReservedStreamName {} -> False
  StreamNameTooLong {} -> False
  DuplicateEvent {} -> False
  EventAlreadyLinked {} -> False
  LinkSourceEventMissing {} -> False
  UnexpectedServerError {} -> False

isTransientCommandError :: CommandError -> Bool
isTransientCommandError = \case
  StoreFailed err -> isTransientStoreError err
  RetryExhausted _ err -> isTransientStoreError err
  ConflictFixpoint _ err -> isTransientStoreError err
  HydrationDecodeFailed {} -> False
  HydrationReplayFailed {} -> False
  HydrationGapDetected {} -> False
  CommandRejected -> False
  CommandAmbiguous {} -> False
  EncodeFailed {} -> False

-- | Whether a command error is a per-command rejection covered by worker policy.
isRejectionClass :: CommandError -> Bool
isRejectionClass = \case
  CommandRejected -> True
  CommandAmbiguous {} -> True
  _ -> False

-- | Classify a group of failed dispatches and choose one acknowledgement.
--
-- Systemic deterministic errors always halt. Any transient error retries the
-- whole source event. Only an all-rejection group reaches the configured
-- 'RejectedCommandPolicy'. Dead-letter writes are idempotent under redelivery.
decideForFailures ::
  (IOE :> es, Store :> es) =>
  WorkerOptions es msg ->
  DispatcherKind ->
  Text ->
  Text ->
  RecordedEvent ->
  Int ->
  [DispatchFailure] ->
  Eff es AckDecision
decideForFailures workerOptions dispatcherKind dispatcherName correlationId sourceEvent attemptCount failures =
  case filter isSystemicDeterministic failures of
    failure : _ -> pure (haltFor failure)
    []
      | any (isTransientCommandError . (^. #commandError)) failures ->
          pure (AckRetry (workerOptions ^. #transientRetryDelay))
      | otherwise ->
          case failures of
            [] -> pure AckOk
            _ -> decideRejected
  where
    isSystemicDeterministic failure =
      let err = failure ^. #commandError
       in not (isTransientCommandError err) && not (isRejectionClass err)

    haltFor failure =
      AckHalt (HaltFatal (Text.pack (show (failure ^. #commandError))))

    decideRejected =
      case workerOptions ^. #rejectedCommandPolicy of
        RejectedHalt -> pure (haltFor (headFailure failures))
        RejectedDeadLetter -> do
          traverse_ recordFailure failures
          recordHandled
          pure AckOk
        RejectedSkip -> do
          recordHandled
          pure AckOk

    recordHandled =
      recordDispatchDeadLettered
        (workerOptions ^. #metrics)
        (fromIntegral (length failures))

    recordFailure failure =
      let err = failure ^. #commandError
       in recordDispatchDeadLetter
            DispatchDeadLetter
              { dispatcherKind = dispatcherKind,
                dispatcherName = dispatcherName,
                correlationId = correlationId,
                sourceEventId = sourceEvent ^. #eventId,
                sourceGlobalPosition = sourceEvent ^. #globalPosition,
                emitIndex = failure ^. #emitIndex,
                targetStreamName = failure ^. #targetStreamName,
                errorClass = commandErrorClass err,
                errorDetail = Text.pack (show err),
                attemptCount = max 1 attemptCount
              }

    headFailure = \case
      failure : _ -> failure
      [] -> DispatchFailure (-1) (StoreTypes.StreamName "unknown") CommandRejected

ackForCommandError :: RetryDelay -> CommandError -> AckDecision
ackForCommandError delay err
  | isTransientCommandError err = AckRetry delay
  | otherwise = AckHalt (HaltFatal (Text.pack (show err)))

-- | Derive a stable, collision-resistant 'EventId' for a manager write from
-- @(manager name, correlation id, source event id, emit index)@ via a v5 UUID.
--
-- The same inputs always yield the same id, so a replayed source event
-- produces the same write ids and the store's uniqueness constraint collapses
-- the duplicate. The manager-state append uses an emit index of @-1@ to keep
-- it distinct from the dispatched commands (which start at @0@). This positional
-- index is sound because 'handle' is pure and therefore returns the same command
-- order for the same input. The effectful router uses
-- 'Keiro.Router.deterministicRouterCommandId' instead, retaining this positional
-- id only as a transition probe for pre-upgrade router dispatches.
deterministicCommandId :: Text -> Text -> EventId -> Int -> EventId
deterministicCommandId managerName correlationId sourceEventId emitIndex =
  EventId
    $ UUID.V5.generateNamed UUID.V5.namespaceURL
    $ fmap (fromIntegral . fromEnum)
    $ Text.unpack
    $ Text.intercalate
      ":"
      [ "keiro",
        "process-manager",
        managerName,
        correlationId,
        UUID.toText (eventIdToUuid sourceEventId),
        Text.pack (show emitIndex)
      ]

-- | React to a single source event: advance the manager's state, dispatch
-- its target commands, and schedule its timers — each under a deterministic,
-- idempotent write id.
--
-- The manager-state append and its timers commit in one transaction; each
-- target command is then dispatched (with its inline projections) in its own.
-- A duplicate manager append short-circuits to 'PMStateDuplicate' but still
-- re-runs the dispatch loop, so a crash between the state append and a
-- command dispatch is recovered on replay. Returns @Left@ only when the
-- manager-state append fails for a non-duplicate reason; per-command failures
-- are reported inside 'commandResults'.
runProcessManagerOnce ::
  forall input phi rs s ci co targetPhi targetRs targetState targetCi targetCo es.
  ( HasCallStack,
    IOE :> es,
    Store :> es,
    Error StoreError :> es,
    KirokuStoreResource :> es,
    BoolAlg phi (RegFile rs, ci),
    BoolAlg targetPhi (RegFile targetRs, targetCi),
    Eq co,
    Eq targetCo
  ) =>
  RunCommandOptions ->
  ProcessManager input phi rs s ci co targetPhi targetRs targetState targetCi targetCo ->
  RecordedEvent ->
  input ->
  Eff es (Either CommandError (ProcessManagerResult (EventStream phi rs s ci co) (EventStream targetPhi targetRs targetState targetCi targetCo)))
runProcessManagerOnce options manager sourceEvent input = do
  let correlationId = (manager ^. #correlate) input
      action = (manager ^. #handle) input
      managerStream = (manager ^. #streamFor) correlationId
      managerEventId = deterministicCommandId (manager ^. #name) correlationId (sourceEvent ^. #eventId) (-1)
      managerOptions = options & #eventIds .~ [managerEventId]
      managerStreamName = ((unvalidated (manager ^. #eventStream)) ^. #resolveStreamName) managerStream
  managerAlreadyProcessed <- eventAlreadyIn options managerStreamName managerEventId
  if managerAlreadyProcessed
    then finish correlationId (PMStateDuplicate managerEventId) action
    else do
      managerOutcome <-
        runCommandWithSql
          managerOptions
          (manager ^. #eventStream)
          managerStream
          (action ^. #command)
          (\_ -> traverse_ scheduleTimerTx (action ^. #timers))
      case managerOutcome of
        Left err -> do
          benign <- confirmBenignDuplicate managerStreamName managerEventId err
          if benign
            then finish correlationId (PMStateDuplicate managerEventId) action
            else pure (Left err)
        Right (managerResult, scheduledInAppend) -> do
          -- No-op manager commands do not execute runCommandWithSql's callback,
          -- so schedule timer-only reactions explicitly.
          case scheduledInAppend of
            Nothing -> runTransaction (traverse_ scheduleTimerTx (action ^. #timers))
            Just () -> pure ()
          finish correlationId (PMStateAppended managerResult) action
  where
    finish correlationId managerResult action = do
      commandResults <- dispatchCommands correlationId (sourceEvent ^. #eventId) (action ^. #commands)
      pure
        $ Right
          ProcessManagerResult
            { managerResult = managerResult,
              commandResults = commandResults,
              timersScheduled = length (action ^. #timers)
            }

    dispatchCommands correlationId sourceEventId commands =
      traverse
        (uncurry (dispatchCommand correlationId sourceEventId))
        (zip [0 ..] commands)

    dispatchCommand correlationId sourceEventId emitIndex command = do
      let commandId = deterministicCommandId (manager ^. #name) correlationId sourceEventId emitIndex
          targetOptions = options & #eventIds .~ [commandId]
          targetStream = retarget (command ^. #target)
          targetStreamName = ((unvalidated (manager ^. #targetEventStream)) ^. #resolveStreamName) targetStream
      commandAlreadyProcessed <- eventAlreadyIn options targetStreamName commandId
      if commandAlreadyProcessed
        then pure (PMCommandDuplicate commandId)
        else do
          outcome <-
            runCommandWithProjections
              targetOptions
              (manager ^. #targetEventStream)
              targetStream
              (command ^. #command)
              ((manager ^. #targetProjections) (command ^. #target))
          case outcome of
            Right result -> pure (PMCommandAppended result)
            Left err -> do
              benign <- confirmBenignDuplicate targetStreamName commandId err
              pure $ if benign then PMCommandDuplicate commandId else PMCommandFailed targetStreamName err

    retarget :: Stream targetCi -> Stream (EventStream targetPhi targetRs targetState targetCi targetCo)
    retarget = coerce

-- | Run a process manager as a live subscription draining a Shibuya adapter with
-- 'defaultWorkerOptions'.
--
-- Use 'runProcessManagerWorkerWith' to override poison-message handling, rejected
-- command handling, transient retry delay, or dispatch metrics. Every ingested message's ack handle is
-- finalized exactly once. Successful and duplicate dispatches finalize 'AckOk';
-- transient store failures finalize 'AckRetry'; rejection-class failures follow
-- 'RejectedCommandPolicy'; other deterministic failures finalize 'AckHalt';
-- undecodable messages follow the configured 'PoisonPolicy'. Under
-- 'RejectedDeadLetter', see the module-level saga-history contract before opting
-- in. On a Kiroku-backed adapter, each 'AckRetry' redelivery is bounded by the
-- subscription @RetryPolicy@ (five total deliveries by default). Exhaustion
-- dead-letters the source event in @kiroku.dead_letters@ and advances the
-- checkpoint; @KirokuAdapterConfig@ does not currently expose that bound. Observe
-- the terminal event with 'Keiro.Telemetry.kirokuEventBridge' and replay it with
-- @Keiro.DeadLetter.Replay@ when appropriate.
runProcessManagerWorker ::
  forall msg input phi rs s ci co targetPhi targetRs targetState targetCi targetCo es.
  ( HasCallStack,
    IOE :> es,
    Store :> es,
    Error StoreError :> es,
    KirokuStoreResource :> es,
    BoolAlg phi (RegFile rs, ci),
    BoolAlg targetPhi (RegFile targetRs, targetCi),
    Eq co,
    Eq targetCo
  ) =>
  RunCommandOptions ->
  ProcessManager input phi rs s ci co targetPhi targetRs targetState targetCi targetCo ->
  Adapter es msg ->
  (msg -> Maybe (RecordedEvent, input)) ->
  Eff es ()
runProcessManagerWorker =
  runProcessManagerWorkerWith defaultWorkerOptions

runProcessManagerWorkerWith ::
  forall msg input phi rs s ci co targetPhi targetRs targetState targetCi targetCo es.
  ( HasCallStack,
    IOE :> es,
    Store :> es,
    Error StoreError :> es,
    KirokuStoreResource :> es,
    BoolAlg phi (RegFile rs, ci),
    BoolAlg targetPhi (RegFile targetRs, targetCi),
    Eq co,
    Eq targetCo
  ) =>
  WorkerOptions es msg ->
  RunCommandOptions ->
  ProcessManager input phi rs s ci co targetPhi targetRs targetState targetCi targetCo ->
  Adapter es msg ->
  (msg -> Maybe (RecordedEvent, input)) ->
  Eff es ()
runProcessManagerWorkerWith workerOptions options manager Adapter {source = adapterSource} decodeMessage =
  Streamly.fold Fold.drain
    $ Streamly.mapM handleIngested adapterSource
  where
    handleIngested :: Ingested es msg -> Eff es AckDecision
    handleIngested Ingested {envelope = env@Envelope {payload = message}, ack = AckHandle finalizeAck} = do
      decision <- case decodeMessage message of
        Nothing -> decideForPoison workerOptions "process-manager worker could not decode message" env
        Just (recorded, input) -> do
          let correlationId = (manager ^. #correlate) input
              managerStream = (manager ^. #streamFor) correlationId
              managerStreamName = ((unvalidated (manager ^. #eventStream)) ^. #resolveStreamName) managerStream
              attemptCount = envelopeAttemptCount env
          outcome <- tryError @StoreError (runProcessManagerOnce options manager recorded input)
          case outcome of
            Left (_, storeErr) -> do
              recordDispatchFailed (workerOptions ^. #metrics) 1
              pure (ackForThrownStoreError (workerOptions ^. #transientRetryDelay) storeErr)
            Right (Left err) -> do
              recordDispatchFailed (workerOptions ^. #metrics) 1
              decideForFailures
                workerOptions
                DispatcherProcessManager
                (manager ^. #name)
                correlationId
                recorded
                attemptCount
                [DispatchFailure (-1) managerStreamName err]
            Right (Right result) ->
              ackForResults
                workerOptions
                (manager ^. #name)
                correlationId
                recorded
                attemptCount
                (result ^. #managerResult)
                (result ^. #commandResults)
      finalizeAck decision
      pure decision

ackForThrownStoreError :: RetryDelay -> StoreError -> AckDecision
ackForThrownStoreError delay = ackForCommandError delay . StoreFailed

ackForResults ::
  (IOE :> es, Store :> es) =>
  WorkerOptions es msg ->
  Text ->
  Text ->
  RecordedEvent ->
  Int ->
  PMStateResult managerTarget ->
  [PMCommandResult commandTarget] ->
  Eff es AckDecision
ackForResults workerOptions managerName correlationId sourceEvent attemptCount managerResult commandResults = do
  let duplicateCount = stateDuplicateCount managerResult + commandDuplicateCount commandResults
      failures =
        [ DispatchFailure emitIndex targetStreamName err
        | (emitIndex, PMCommandFailed targetStreamName err) <- zip [0 ..] commandResults
        ]
  recordDispatchDuplicate (workerOptions ^. #metrics) duplicateCount
  recordDispatchFailed (workerOptions ^. #metrics) (fromIntegral (length failures))
  decideForFailures
    workerOptions
    DispatcherProcessManager
    managerName
    correlationId
    sourceEvent
    attemptCount
    failures

stateDuplicateCount :: PMStateResult target -> Int64
stateDuplicateCount = \case
  PMStateDuplicate {} -> 1
  PMStateAppended {} -> 0

commandDuplicateCount :: [PMCommandResult target] -> Int64
commandDuplicateCount =
  fromIntegral . length . filter isDuplicateResult
  where
    isDuplicateResult = \case
      PMCommandDuplicate {} -> True
      _ -> False

envelopeAttemptCount :: Envelope msg -> Int
envelopeAttemptCount env =
  case env ^. #attempt of
    Nothing -> 1
    Just (Attempt attempt) -> fromIntegral attempt + 1

decideForPoison ::
  (IOE :> es) =>
  WorkerOptions es msg ->
  Text ->
  Envelope msg ->
  Eff es AckDecision
decideForPoison workerOptions reason env = do
  recordDispatchPoison (workerOptions ^. #metrics) 1
  case workerOptions ^. #poisonPolicy of
    PoisonHalt -> pure (AckHalt (HaltFatal reason))
    PoisonSkip callback -> do
      callback env
      pure AckOk
    PoisonDeadLetter callback -> do
      callback env
      pure (AckDeadLetter (InvalidPayload reason))

eventIdToUuid :: EventId -> UUID.UUID
eventIdToUuid (EventId uuid) = uuid

-- | Check whether an event with the given id is already present in a live stream.
-- Used as the pre-dispatch idempotency guard so a
-- command that was already applied on a prior (possibly crashed) attempt is
-- recognized as a duplicate before re-running it.
eventAlreadyIn ::
  (Store :> es) =>
  RunCommandOptions ->
  StoreTypes.StreamName ->
  EventId ->
  Eff es Bool
eventAlreadyIn _options streamName eventId =
  eventExistsInStream streamName eventId

-- | Decide whether a failed append is a benign duplicate of the write just
-- attempted: whether @ourId@ is genuinely present in @streamName@.
--
-- Kiroku's @DuplicateEvent@ carries 'Just' the colliding id only when
-- PostgreSQL's detail string parses ('Nothing' otherwise), and because the
-- store's event-id uniqueness is global, even a matching id does not prove the
-- event landed in our stream. A mismatched id is never ours; a matching or
-- missing id is confirmed against the target stream with a point lookup. Callers
-- fold 'True' into their duplicate result and surface 'False' as the original
-- failure.
confirmBenignDuplicate ::
  (Store :> es) =>
  StoreTypes.StreamName ->
  EventId ->
  CommandError ->
  Eff es Bool
confirmBenignDuplicate streamName ourId = \case
  StoreFailed (DuplicateEvent (Just duplicateId))
    | duplicateId == ourId -> eventExistsInStream streamName ourId
  StoreFailed (DuplicateEvent Nothing) -> eventExistsInStream streamName ourId
  _ -> pure False