Replay.hs

{- | Operator replay for source events parked in Kiroku's
@kiroku.dead_letters@ table.

The rows remain Kiroku-owned: replay neither deletes nor marks them. Instead it
re-runs a caller-supplied handler and relies on that handler's idempotency. This
is safe for "Keiro.ProcessManager" and "Keiro.Router" handlers because their
writes use deterministic event identifiers derived from the source event;
already-applied writes collapse to duplicate outcomes on every later replay.

Kiroku exposes dead-letter rows by event id and global position, but does not
currently expose an exact point-read by either value. Global positions are
opaque cursors and must never be decremented. 'replaySubscriptionDeadLetters'
therefore scans the global stream backward using only @0@ and cursors returned
by Kiroku, matches both event id and position, and scans once for the whole
batch. A source event removed by hard deletion is reported as
'ReplaySourceMissing' and is not handed to the caller.

A process-manager handler normally decodes the recorded event, calls
@runProcessManagerOnce@, and classifies a result whose manager state and every
target command are duplicates as 'ReplayedDuplicate'; any append makes it
'ReplayedFresh'. Return @Left detail@ for a decode or command failure. For
example, the classification has this shape:

> case result of
>   Left err -> Left (showText err)
>   Right pmResult
>     | managerAndEveryCommandAreDuplicates pmResult -> Right ReplayedDuplicate
>     | otherwise -> Right ReplayedFresh

Because rows are retained, an operator may safely run the same replay command
again after a partial failure or uncertain client disconnect.
-}
module Keiro.DeadLetter.Replay (
    ReplayOutcome (..),
    ReplayResult (..),
    DeadLetterRecord (..),
    listSubscriptionDeadLetters,
    replaySubscriptionDeadLetters,
)
where

import Data.Int (Int32)
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Vector (Vector)
import Data.Vector qualified as Vector
import Effectful (Eff, (:>))
import Keiro.Prelude
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Read (readAllBackward)
import Kiroku.Store.SQL (DeadLetterRecord (..), readDeadLettersStmt)
import Kiroku.Store.Subscription.Types (SubscriptionName (..))
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (EventId (..), GlobalPosition (..), RecordedEvent)
import "hasql-transaction" Hasql.Transaction qualified as Tx

data ReplayOutcome = ReplayOutcome
    { replayGlobalPosition :: !GlobalPosition
    , replayEventId :: !EventId
    , replayResult :: !ReplayResult
    }
    deriving stock (Generic, Eq, Show)

data ReplayResult
    = ReplayedFresh
    | ReplayedDuplicate
    | ReplayFailed !Text
    | ReplaySourceMissing
    deriving stock (Generic, Eq, Show)

-- | List one Kiroku subscription member's dead letters, newest first.
listSubscriptionDeadLetters ::
    (Store :> es) => SubscriptionName -> Int32 -> Eff es (Vector DeadLetterRecord)
listSubscriptionDeadLetters (SubscriptionName name) member =
    runTransaction (Tx.statement (name, member) readDeadLettersStmt)

{- | Replay every dead letter currently listed for one subscription member.

The handler controls domain-specific decoding and decides whether its writes
were fresh or duplicates. A @Left detail@ is recorded in the corresponding
'ReplayOutcome' as 'ReplayFailed'; replay continues with later rows. Store
errors still surface through the surrounding Kiroku 'Store' interpreter.
-}
replaySubscriptionDeadLetters ::
    (Store :> es) =>
    SubscriptionName ->
    Int32 ->
    (RecordedEvent -> Eff es (Either Text ReplayResult)) ->
    Eff es [ReplayOutcome]
replaySubscriptionDeadLetters subscriptionName member handler = do
    rows <- listSubscriptionDeadLetters subscriptionName member
    sources <- findSources rows
    traverse (replayOne sources) (Vector.toList rows)
  where
    replayOne sources row = do
        let position = deadLetterPosition row
            eventId = deadLetterEventIdValue row
        result <- case Map.lookup eventId sources of
            Just event
                | event ^. #globalPosition == position -> do
                    handled <- handler event
                    pure (either ReplayFailed id handled)
            _ -> pure ReplaySourceMissing
        pure
            ReplayOutcome
                { replayGlobalPosition = position
                , replayEventId = eventId
                , replayResult = result
                }

-- Scan once for the complete replay batch. Pages are descending, and the next
-- cursor is always a position Kiroku returned; no global-position arithmetic.
findSources ::
    (Store :> es) => Vector DeadLetterRecord -> Eff es (Map EventId RecordedEvent)
findSources rows = go (GlobalPosition 0) Map.empty
  where
    wanted =
        Map.fromList
            [ (deadLetterEventIdValue row, deadLetterPosition row)
            | row <- Vector.toList rows
            ]

    go cursor found
        | Map.size found == Map.size wanted = pure found
        | otherwise = do
            page <- readAllBackward cursor replayReadPageSize
            if Vector.null page
                then pure found
                else do
                    let found' = Vector.foldl' remember found page
                        nextCursor = Vector.last page ^. #globalPosition
                    go nextCursor found'

    remember found event =
        case Map.lookup (event ^. #eventId) wanted of
            Just expectedPosition
                | event ^. #globalPosition == expectedPosition ->
                    Map.insert (event ^. #eventId) event found
            _ -> found

replayReadPageSize :: Int32
replayReadPageSize = 256

deadLetterPosition :: DeadLetterRecord -> GlobalPosition
deadLetterPosition row = GlobalPosition (row ^. #deadLetterGlobalPosition)

deadLetterEventIdValue :: DeadLetterRecord -> EventId
deadLetterEventIdValue row = EventId (row ^. #deadLetterEventId)