Router.hs

{- | Stateless, content-based routing of events to commands.

A 'Router' is the stateless sibling of
'Keiro.ProcessManager.ProcessManager': for each incoming event it resolves
a data-dependent set of target streams /effectfully/ (typically via a
read-model query) and dispatches one command to each. Dispatch is
idempotent per resolved target identity: every command is appended under a
target-name-keyed deterministic id, and store-level duplicate rejections are
confirmed against that target before becoming a benign
'PMCommandDuplicate'. Redelivery therefore deduplicates every target resolved
again, regardless of target order. Because resolution is effectful, the target
set may drift between attempts; dispatches accumulate as the union of those
attempts, so callers that require one exact set must keep resolution stable for
a source event.

Use 'runRouterOnce' to dispatch a single event, or 'runRouterWorker' to run
the router as a live subscription draining a Shibuya adapter. Its retry and
source-event dead-letter contract is the same bounded Kiroku contract described
by "Keiro.ProcessManager": five total deliveries by default, followed by a
@kiroku.dead_letters@ write and atomic checkpoint advance.

A router's 'key' can join events from different source streams just as a
process manager's @correlate@ function can. The same ordering rule applies:
same-stream order is preserved, but different streams have no relative
business-order guarantee and may run concurrently under sharding. Keep routed
logic order-insensitive; see "Keiro.ProcessManager" for the worked example.
Each resolved target command (with its inline projections) commits in its own
transaction, so fan-out is idempotent rather than all-target atomic.
-}
module Keiro.Router (
    -- * Definition
    Router (..),
    RouterResult (..),

    -- * Idempotency
    deterministicRouterCommandId,

    -- * Running
    runRouterOnce,
    runRouterWorkerWith,
    runRouterWorker,
)
where

import Data.ByteString qualified as ByteString
import Data.ByteString.Char8 qualified as ByteString.Char8
import Data.Coerce (coerce)
import Data.Map.Strict qualified as Map
import Data.Text qualified as Text
import Data.Text.Encoding qualified as Text.Encoding
import Data.Traversable (mapAccumL)
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 (..), RunCommandOptions)
import Keiro.DeadLetter (DispatcherKind (..))
import Keiro.EventStream (EventStream)
import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated)
import Keiro.Prelude
import Keiro.ProcessManager (
    DispatchFailure (..),
    PMCommand (..),
    PMCommandResult (..),
    PoisonPolicy (..),
    WorkerOptions (..),
    ackForCommandError,
    confirmBenignDuplicate,
    decideForFailures,
    defaultWorkerOptions,
    deterministicCommandId,
    eventAlreadyIn,
 )
import Keiro.Projection (InlineProjection, runCommandWithProjections)
import Keiro.Stream (Stream)
import Keiro.Telemetry (recordDispatchDuplicate, recordDispatchFailed, recordDispatchPoison)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Effect.Resource (KirokuStoreResource)
import Kiroku.Store.Error (StoreError (..))
import Kiroku.Store.Types (EventId (..), RecordedEvent, StreamName (..))
import Shibuya.Adapter (Adapter (..))
import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))
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 (filter, fromIntegral, length, snd, zip, (+))

{- | A stateless, content-based router (in the Enterprise Integration Patterns
sense): for each incoming event it resolves a data-dependent set of target
streams /effectfully/ and dispatches one command to each.

This is the stateless counterpart of 'Keiro.ProcessManager.ProcessManager'. It
has no manager state stream, no @correlate@, and no self-directed command. Its
sole new capability over the process manager is that target resolution runs in
@Eff es@ — typically a read-model query via 'Keiro.ReadModel.runQuery' — so the
fan-out set can be /looked up/ rather than computed purely from the event.

Dispatch is idempotent by construction: each target command is appended under a
deterministic identifier derived from @(name, key input, source event id,
resolved target stream name, occurrence)@ (see
'deterministicRouterCommandId'), pre-checked with 'eventAlreadyIn', and the
store's @DuplicateEvent@ rejection is confirmed against the target stream
before it is treated as benign. A redelivery deduplicates every target it
resolves again even if target order or membership changed. A target resolved
only on an earlier attempt keeps its immutable dispatch, and a newly resolved
target is dispatched on the later attempt; the cumulative set is therefore the
union of attempt outputs. Keep 'resolve' stable for a source event when the
exact recipient set matters.

Each dispatch also runs 'targetProjections' for the target aggregate in the same
append transaction. The function receives the concrete target stream so callers can
build projections closed over stream-local keys. Return @[]@ to preserve
append-only dispatch.
-}
data Router input targetPhi targetRs targetState targetCi targetCo es = Router
    { name :: !Text
    -- ^ Stable identifier; part of every dispatched command's deterministic id.
    , key :: !(input -> Text)
    -- ^ Correlation string for the source event (e.g. the transaction id).
    , resolve :: !(input -> Eff es [PMCommand targetCi])
    {- ^ The effectful seam: compute the data-dependent target set, typically
    @runQuery readModel q@.
    -}
    , targetEventStream :: !(ValidatedEventStream targetPhi targetRs targetState targetCi targetCo)
    -- ^ The aggregate every resolved command is dispatched to.
    , targetProjections :: !(Stream targetCi -> [InlineProjection targetCo])
    {- ^ Inline projections for the target aggregate, run in the same transaction
    as each dispatched command's append. Return @[]@ for append-only dispatch.
    -}
    }
    deriving stock (Generic)

{- | The outcome of a single 'runRouterOnce' invocation: one
'PMCommandResult' per resolved target, in resolution order.

Unlike 'Keiro.ProcessManager.ProcessManagerResult' there is no manager-state
result, because a router has no state stream. A failed dispatch surfaces as a
'PMCommandFailed' element rather than an outer 'Either'.
-}
newtype RouterResult target = RouterResult
    { commandResults :: [PMCommandResult target]
    }
    deriving stock (Generic, Eq, Show)

{- | Derive a stable, collision-resistant 'EventId' for a router dispatch from
@(router name, key input, source event id, resolved target stream name,
occurrence)@ via a v5 UUID.

Unlike 'Keiro.ProcessManager.deterministicCommandId' (which the process manager
still uses, soundly, because its command list is a pure function of the input),
the router keys the id by the target's identity rather than its position in the
resolved list: 'resolve' is effectful, so a redelivery may see the same targets
in a different order or a drifted set, and a positional id would then point at
the wrong target. The @occurrence@ is the index among commands in the same
resolve batch that address the same target stream (0 for the first), so
resolving the same target twice in one batch still yields distinct ids.

The v5 name encodes every text field as length-prefixed UTF-8. This avoids both
delimiter ambiguity when names contain colons and character truncation for
non-ASCII names.
-}
deterministicRouterCommandId :: Text -> Text -> EventId -> StreamName -> Int -> EventId
deterministicRouterCommandId routerName correlationId sourceEventId targetStreamName occurrence =
    EventId
        $ UUID.V5.generateNamed UUID.V5.namespaceURL
        $ ByteString.unpack
        $ ByteString.concat
        $ fmap
            encodeField
            [ "keiro"
            , "router"
            , routerName
            , correlationId
            , UUID.toText (coerce sourceEventId)
            , coerce targetStreamName
            , Text.pack (show occurrence)
            ]
  where
    encodeField field =
        let bytes = Text.Encoding.encodeUtf8 field
         in ByteString.concat
                [ ByteString.Char8.pack (show (ByteString.length bytes))
                , ByteString.singleton 58
                , bytes
                ]

{- | Resolve the targets for one source event, then dispatch one command per
target with crash-safe, target-identity idempotency.

Unlike 'Keiro.ProcessManager.runProcessManagerOnce', whose pure command list
can safely use positional ids, a router derives each id from the resolved
target stream name and its same-stream occurrence. It skips ids already in the
target stream, otherwise runs 'Keiro.Projection.runCommandWithProjections', and
folds a @DuplicateEvent@ rejection only after confirming the attempted id is in
that target stream.

Returns 'RouterResult' directly (no outer @Either CommandError@) because — unlike
the process manager — there is no manager-state append that can fail before
dispatch.
-}
runRouterOnce ::
    forall input targetPhi targetRs targetState targetCi targetCo es.
    ( HasCallStack
    , IOE :> es
    , Store :> es
    , Error StoreError :> es
    , KirokuStoreResource :> es
    , BoolAlg targetPhi (RegFile targetRs, targetCi)
    , Eq targetCo
    ) =>
    RunCommandOptions ->
    Router input targetPhi targetRs targetState targetCi targetCo es ->
    RecordedEvent ->
    input ->
    Eff es (RouterResult (EventStream targetPhi targetRs targetState targetCi targetCo))
runRouterOnce options router sourceEvent input = do
    let correlationId = (router ^. #key) input
    commands <- (router ^. #resolve) input
    let named =
            [ (streamNameOf command, command)
            | command <- commands
            ]
        annotated = snd (mapAccumL occurrenceStep Map.empty (zip [0 ..] named))
        occurrenceStep seen (legacyIndex, (targetStreamName, command)) =
            let occurrence = Map.findWithDefault 0 targetStreamName seen
             in ( Map.insert targetStreamName (occurrence + 1) seen
                , (legacyIndex, occurrence, targetStreamName, command)
                )
    results <-
        traverse
            (dispatchCommand correlationId (sourceEvent ^. #eventId))
            annotated
    pure (RouterResult results)
  where
    streamNameOf command =
        ((unvalidated (router ^. #targetEventStream)) ^. #resolveStreamName)
            (retarget (command ^. #target))

    dispatchCommand correlationId sourceEventId (legacyIndex, occurrence, targetStreamName, command) = do
        let commandId =
                deterministicRouterCommandId
                    (router ^. #name)
                    correlationId
                    sourceEventId
                    targetStreamName
                    occurrence
            -- Transition: dispatches written by keiro versions that derived
            -- positional ids must still dedup across the upgrade. Remove in a
            -- later release after the compatibility window closes.
            legacyCommandId =
                deterministicCommandId
                    (router ^. #name)
                    correlationId
                    sourceEventId
                    legacyIndex
            targetOptions = options & #eventIds .~ [commandId]
            targetEventStream = router ^. #targetEventStream
            targetStream = retarget (command ^. #target)
        commandAlreadyProcessed <- eventAlreadyIn options targetStreamName commandId
        legacyAlreadyProcessed <-
            if commandAlreadyProcessed
                then pure False
                else eventAlreadyIn options targetStreamName legacyCommandId
        if commandAlreadyProcessed
            then pure (PMCommandDuplicate commandId)
            else
                if legacyAlreadyProcessed
                    then pure (PMCommandDuplicate legacyCommandId)
                    else do
                        outcome <-
                            runCommandWithProjections
                                targetOptions
                                targetEventStream
                                targetStream
                                (command ^. #command)
                                ((router ^. #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 'Router' as a live subscription over a Shibuya 'Adapter'.

Mirrors 'Keiro.ProcessManager.runProcessManagerWorker': it drains the adapter's
message stream, decoding each message to a @(RecordedEvent, input)@ pair and
dispatching it through 'runRouterOnce'.

Ack policy (see this plan's Decision Log):

  * a message that fails to decode follows the configured 'PoisonPolicy'
    (default: 'AckHalt' @HaltFatal@);
  * otherwise, after dispatch, if every 'PMCommandResult' is
    'PMCommandAppended' or 'PMCommandDuplicate' the message finalizes 'AckOk';
  * if any dispatch is 'PMCommandFailed', transient failures finalize
    'AckRetry', systemic deterministic failures finalize @AckHalt
    (HaltFatal …)@, and rejection-class failures follow
    'RejectedCommandPolicy'.

Benign domain rejections (a target aggregate refusing a "check" command because
no edge matches) must be modeled as /total/ transitions in the keiki transducer
(an ε-complement self-loop) so they never surface as 'PMCommandFailed' and
therefore never wedge the worker. When the rejection is genuinely
data-dependent, 'RejectedDeadLetter' persists a queryable
"Keiro.DeadLetter.DispatchDeadLetter" and acknowledges the source event;
'RejectedSkip' acknowledges and records only the metric. 'RejectedHalt' remains
the default.

The worker invokes each ingested message's 'Shibuya.Core.AckHandle.AckHandle'
@finalize@ exactly once with the decision, so the decision reaches the adapter.
Use 'runRouterWorkerWith' to override poison-message handling, rejected-command
handling, transient retry delay, or dispatch metrics. On a Kiroku-backed
adapter, 'AckRetry' remains bounded by the subscription @RetryPolicy@ (five
total deliveries by default). Exhaustion records the source event in
@kiroku.dead_letters@ with reason kind @max_attempts_exceeded@ and advances the
checkpoint. @KirokuAdapterConfig@ does not currently expose @retryPolicy@;
install 'Keiro.Telemetry.kirokuEventBridge' on Kiroku's @eventHandler@ to observe
the terminal event. The configurable sharded path forwards
@ShardedWorkerOptions.retryPolicy@ through the same Kiroku ladder; see
"Keiro.Subscription.Shard.Worker".
-}
runRouterWorker ::
    forall msg input targetPhi targetRs targetState targetCi targetCo es.
    ( HasCallStack
    , IOE :> es
    , Store :> es
    , Error StoreError :> es
    , KirokuStoreResource :> es
    , BoolAlg targetPhi (RegFile targetRs, targetCi)
    , Eq targetCo
    ) =>
    RunCommandOptions ->
    Router input targetPhi targetRs targetState targetCi targetCo es ->
    Adapter es msg ->
    (msg -> Maybe (RecordedEvent, input)) ->
    Eff es ()
runRouterWorker =
    runRouterWorkerWith defaultWorkerOptions

runRouterWorkerWith ::
    forall msg input targetPhi targetRs targetState targetCi targetCo es.
    ( HasCallStack
    , IOE :> es
    , Store :> es
    , Error StoreError :> es
    , KirokuStoreResource :> es
    , BoolAlg targetPhi (RegFile targetRs, targetCi)
    , Eq targetCo
    ) =>
    WorkerOptions es msg ->
    RunCommandOptions ->
    Router input targetPhi targetRs targetState targetCi targetCo es ->
    Adapter es msg ->
    (msg -> Maybe (RecordedEvent, input)) ->
    Eff es ()
runRouterWorkerWith workerOptions options router 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 "router worker could not decode message" env
            Just (recorded, input) -> do
                let correlationId = (router ^. #key) input
                    attemptCount = envelopeAttemptCount env
                outcome <- tryError @StoreError (runRouterOnce options router recorded input)
                case outcome of
                    Left (_, storeErr) -> do
                        recordDispatchFailed (workerOptions ^. #metrics) 1
                        pure (ackForCommandError (workerOptions ^. #transientRetryDelay) (StoreFailed storeErr))
                    Right (RouterResult results) -> ackDecisionFor recorded correlationId attemptCount results
        finalizeAck decision
        pure decision

    ackDecisionFor :: RecordedEvent -> Text -> Int -> [PMCommandResult target] -> Eff es AckDecision
    ackDecisionFor sourceEvent correlationId attemptCount results = do
        let duplicateCount = commandDuplicateCount results
            failures =
                [ DispatchFailure emitIndex targetStreamName err
                | (emitIndex, PMCommandFailed targetStreamName err) <- zip [0 ..] results
                ]
        recordDispatchDuplicate (workerOptions ^. #metrics) duplicateCount
        recordDispatchFailed (workerOptions ^. #metrics) (fromIntegral (length failures))
        decideForFailures
            workerOptions
            DispatcherRouter
            (router ^. #name)
            correlationId
            sourceEvent
            attemptCount
            failures

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

    decideForPoison :: Text -> Envelope msg -> Eff es AckDecision
    decideForPoison 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))

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