Outbox.hs

{- | Durable integration-event outbox.

The outbox decouples "this service has decided to publish an integration
event" from "this service has actually published it". Two surfaces use
it:

* The canonical 'IntegrationProducer' helper maps durable private events
  to public 'Keiro.Integration.Event.IntegrationEvent' values and enqueues
  one outbox row per mapped event. It mints @messageId@ as a prefixed
  UUIDv7 (TypeID) so the id is time-ordered, human-readable, and stable
  across publish retries.
* 'enqueueOutboxTx' is the inline escape hatch for sagas and process
  managers that need to emit an integration event without an intermediate
  private domain event. It runs inside the caller's
  'Hasql.Transaction.Transaction'.

The 'publishClaimedOutbox' worker is transport-neutral. It claims rows
with @FOR UPDATE SKIP LOCKED@ plus the configured 'OrderingPolicy',
hands claimed batches to a caller-supplied publish function, and marks rows
sent, retryable, or dead. The Kafka adapter lives in
'Keiro.Outbox.Kafka'.

Run 'outboxMaintenancePass' on a separate, slower schedule to reclaim rows
left in @publishing@ by crashed workers and to sample the backlog gauge.

The per-key and per-source ordering policies sort by @created_at@, which
PostgreSQL fills at transaction start. The canonical 'IntegrationProducer'
subscription serializes same-key enqueues, so its ordering is stable. Callers
using the inline 'enqueueIntegrationEventTx' escape hatch concurrently for the
same key must serialize those enqueues themselves or accept best-effort order:
two transactions can commit in the opposite order of their @created_at@ values.
-}
module Keiro.Outbox (
    -- * Re-exports
    module Keiro.Outbox.Types,

    -- * Storage primitives (transport-neutral)
    enqueueOutboxTx,
    claimOutboxBatch,
    requeueStuckOutbox,
    markOutboxSent,
    lookupOutbox,
    listOutbox,
    countOutboxBacklog,
    garbageCollectSent,

    -- * Inline escape hatch
    freshOutboxId,
    enqueueIntegrationEventTx,

    -- * Canonical producer-subscription helper
    IntegrationProducer (..),
    IntegrationProducerConfigError (..),
    IntegrationEventDraft (..),
    mkIntegrationProducer,
    mintIntegrationEvent,
    draftToEvent,
    enqueueProducerEventTx,

    -- * Publisher worker
    PublishOutcome (..),
    publishClaimedOutbox,
    outboxMaintenancePass,
    sampleOutboxBacklog,
)
where

import Data.ByteString (ByteString)
import Data.Map.Strict qualified as Map
import Data.Text qualified as Text
import Data.TypeID qualified as TypeID
import Data.UUID.V7 qualified as V7
import Effectful (Eff, IOE, (:>))
import Effectful.Exception (displayException, trySync)
import Keiro.Integration.Event (
    IntegrationContentType,
    IntegrationEvent (..),
    SchemaReference,
    TraceContext,
 )
import Keiro.Outbox.Kafka (outboxRowToKafkaRecord)
import Keiro.Outbox.Schema
import Keiro.Outbox.Types
import Keiro.Prelude
import Keiro.Telemetry (
    KeiroMetrics,
    recordOutboxBacklog,
    recordOutboxDeadlettered,
    recordOutboxPublished,
    recordOutboxReclaimed,
    recordOutboxRetried,
    withProducerSpan,
 )
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (EventId, GlobalPosition, RecordedEvent)
import OpenTelemetry.Attributes.Key (AttributeKey (..), unkey)
import OpenTelemetry.SemanticConventions (error_type)
import OpenTelemetry.Trace.Core (SpanStatus (..), addAttribute, setStatus)
import "hasql-transaction" Hasql.Transaction qualified as Tx

keiro_outbox_batch_size :: AttributeKey Int64
keiro_outbox_batch_size = AttributeKey "keiro.outbox.batch.size"

-- | Mint a fresh time-ordered UUIDv7 for use as an 'OutboxId'.
freshOutboxId :: (IOE :> es) => Eff es OutboxId
freshOutboxId = fmap OutboxId (liftIO V7.genUUID)

{- | Enqueue an 'IntegrationEvent' from a saga or process manager that is
already running inside a 'runCommandWithSqlEvents' transaction. The
caller supplies a stable 'OutboxId' so retried command attempts coalesce
on the @(source, message_id)@ unique constraint.

Ordering caveat: under 'PerKeyHeadOfLine' and 'PerSourceStream', the publisher
orders rows by @created_at@, which PostgreSQL sets to transaction-start time.
If two concurrent transactions enqueue the same key/source and commit in the
opposite order, a publisher can observe that order. Serialize same-key enqueues
when strict order matters.
-}
enqueueIntegrationEventTx ::
    OutboxId ->
    IntegrationEvent ->
    Tx.Transaction ()
enqueueIntegrationEventTx outboxId event =
    enqueueOutboxTx (OutboxMessage{outboxId, event})

-- ---------------------------------------------------------------------------
-- Producer-subscription helper
-- ---------------------------------------------------------------------------

{- | Configuration for the canonical producer subscription.

A service running 'IntegrationProducer' reads its private event stream,
decodes each event with a 'Keiro.Codec.Codec', calls 'mapEvent', and for
each 'Just' result writes one 'keiro_outbox' row. The helper mints
@messageId@ on each insert so the id is stable across publish retries.

* 'name' — subscription name used to checkpoint the producer's cursor
  in the @subscriptions@ table.
* 'source' — value written into @keiro_outbox.source@; identifies the
  producing bounded context.
* 'messageIdPrefix' — TypeID prefix used when minting @messageId@.
  Must be 1-63 lowercase Latin letters (e.g. @\"msg\"@, @\"order\"@).
  Prefer constructing producers with 'mkIntegrationProducer'; an invalid
  prefix passed directly to 'IntegrationProducer' raises when the first
  message id is minted.
* 'mapEvent' — pure mapper from a private 'RecordedEvent' and its
  decoded payload to an 'IntegrationEventDraft'. Returning 'Nothing'
  skips the event without enqueuing a row.
-}
data IntegrationProducer e = IntegrationProducer
    { name :: !Text
    , source :: !Text
    , messageIdPrefix :: !Text
    , mapEvent :: !(RecordedEvent -> e -> Maybe IntegrationEventDraft)
    }
    deriving stock (Generic)

data IntegrationProducerConfigError
    = InvalidMessageIdPrefix !Text !Text
    deriving stock (Generic, Eq, Show)

-- | Validate an integration producer before starting its subscription.
mkIntegrationProducer :: IntegrationProducer e -> Either IntegrationProducerConfigError (IntegrationProducer e)
mkIntegrationProducer producer =
    case TypeID.checkPrefix (producer ^. #messageIdPrefix) of
        Nothing -> Right producer
        Just err ->
            Left
                ( InvalidMessageIdPrefix
                    (producer ^. #messageIdPrefix)
                    (Text.pack (show err))
                )

{- | Everything in 'IntegrationEvent' except 'messageId' and 'source' —
those are filled in by 'mintIntegrationEvent' from the producer
configuration and the freshly minted TypeID.

@sourceEventId@ and @sourceGlobalPosition@ default to the values on the
underlying 'RecordedEvent' (see 'mintIntegrationEvent'); a mapper that
needs to override them can replace the draft fields directly.
-}
data IntegrationEventDraft = IntegrationEventDraft
    { destination :: !Text
    , key :: !(Maybe Text)
    , eventType :: !Text
    , schemaVersion :: !Int
    , contentType :: !IntegrationContentType
    , schemaReference :: !(Maybe SchemaReference)
    , sourceEventId :: !(Maybe EventId)
    , sourceGlobalPosition :: !(Maybe GlobalPosition)
    , payloadBytes :: !ByteString
    , occurredAt :: !UTCTime
    , causationId :: !(Maybe EventId)
    , correlationId :: !(Maybe EventId)
    , traceContext :: !(Maybe TraceContext)
    , attributes :: !(Maybe Value)
    }
    deriving stock (Generic, Eq, Show)

{- | Mint a fresh @messageId@ (TypeID with the producer's prefix) and build
the full 'IntegrationEvent' from the draft. Lives in 'IO' because TypeID
generation reads the global UUIDv7 sequence counter.
-}
mintIntegrationEvent ::
    (IOE :> es) =>
    IntegrationProducer e ->
    IntegrationEventDraft ->
    Eff es IntegrationEvent
mintIntegrationEvent producer draft = do
    typeId <- liftIO (TypeID.genTypeID (producer ^. #messageIdPrefix))
    pure (draftToEvent (producer ^. #source) (TypeID.toText typeId) draft)

-- | Build an 'IntegrationEvent' from a source, a minted message id, and a draft.
draftToEvent :: Text -> Text -> IntegrationEventDraft -> IntegrationEvent
draftToEvent source minted draft =
    IntegrationEvent
        { messageId = minted
        , source
        , destination = draft ^. #destination
        , key = draft ^. #key
        , eventType = draft ^. #eventType
        , schemaVersion = draft ^. #schemaVersion
        , contentType = draft ^. #contentType
        , schemaReference = draft ^. #schemaReference
        , sourceEventId = draft ^. #sourceEventId
        , sourceGlobalPosition = draft ^. #sourceGlobalPosition
        , payloadBytes = draft ^. #payloadBytes
        , occurredAt = draft ^. #occurredAt
        , causationId = draft ^. #causationId
        , correlationId = draft ^. #correlationId
        , traceContext = draft ^. #traceContext
        , attributes = draft ^. #attributes
        }

{- | Enqueue one drafted producer event inside an existing transaction.

This is the primitive a subscription worker calls per event. It mints a
fresh @messageId@ (TypeID), constructs the full envelope, and inserts
the row. The caller supplies the 'OutboxId' so retries from a known
subscription cursor coalesce on @(source, message_id)@.

The TypeID is minted before the insert; if the transaction rolls back
the message id is discarded (no observable effect) and the next attempt
mints a different id. Idempotency at the row level relies on a stable
'OutboxId', not the minted message id.

Ordering caveat: @created_at@ records transaction-start time. Under
'PerKeyHeadOfLine' or 'PerSourceStream', concurrent transactions for the same
key/source can commit in the opposite order and are therefore best-effort
unless the caller serializes them. The canonical producer subscription does
serialize same-key enqueues.
-}
enqueueProducerEventTx ::
    forall e es.
    (IOE :> es) =>
    IntegrationProducer e ->
    OutboxId ->
    IntegrationEventDraft ->
    Eff es (Tx.Transaction ())
enqueueProducerEventTx producer outboxId draft = do
    event <- mintIntegrationEvent producer draft
    pure (enqueueOutboxTx (OutboxMessage{outboxId, event}))

-- ---------------------------------------------------------------------------
-- Publisher worker
-- ---------------------------------------------------------------------------

-- | Result of one publish attempt as reported by the transport-specific publisher.
data PublishOutcome
    = -- | Kafka acknowledged the publish.
      PublishSucceeded
    | -- | Publish failed; will be retried after the configured backoff.
      PublishFailed !Text
    deriving stock (Generic, Eq, Show)

{- | Drain claimed outbox rows by handing the claimed batch to @publish@ and
reflecting the outcomes back into row statuses.

Claims rows in batches of @batchSize@ under the active 'OrderingPolicy',
calls @publish@ with the claimed rows in claim order, and marks every row
sent or — using 'markOutboxFailedTx' — failed/dead. The publish result must
contain one outcome per input row; a missing outcome is treated as
@PublishFailed "publisher returned no outcome"@. If the publisher throws,
every row in that call is treated as failed with the exception text.

For ordered policies, if a row fails then later rows in the same ordered group
are skipped and returned to @failed@ without consuming an attempt; these
skipped rows count as 'OutboxPublishSummary.retried'. A real Kafka transport
must not successfully deliver a later same-key record after reporting an
earlier same-key failure from the same call. On 'StopTheLine', the worker calls
@publish@ with singleton batches and halts after the first failed row, recording
the offending 'OutboxId' in 'haltedOn'.

Returns when one of:

* No rows are claimable.
* The active policy is 'StopTheLine' and a publish failed.

The worker does not loop indefinitely; the application is expected to
schedule it repeatedly (e.g. once per process-compose tick).
-}
publishClaimedOutbox ::
    forall es.
    (IOE :> es, Store :> es) =>
    ([OutboxRow] -> Eff es [(OutboxId, PublishOutcome)]) ->
    OutboxPublishOptions ->
    Maybe KeiroMetrics ->
    Eff es OutboxPublishSummary
publishClaimedOutbox publish options mMetrics = do
    now <- liftIO getCurrentTime
    rows <- claimOutboxBatch (options ^. #orderingPolicy) (options ^. #batchSize) now
    summary <- publishBatch rows
    -- Counters from the aggregated pass summary (each a no-op under 'Nothing';
    -- a zero delta is harmless).
    recordOutboxPublished mMetrics (fromIntegral (summary ^. #published))
    recordOutboxRetried mMetrics (fromIntegral (summary ^. #retried))
    recordOutboxDeadlettered mMetrics (fromIntegral (summary ^. #dead))
    pure summary
  where
    publishBatch :: [OutboxRow] -> Eff es OutboxPublishSummary
    publishBatch [] =
        pure OutboxPublishSummary{claimed = 0, published = 0, retried = 0, dead = 0, haltedOn = Nothing}
    publishBatch batch =
        case options ^. #orderingPolicy of
            StopTheLine -> publishStopTheLine batch batch [] Nothing
            policy -> do
                outcomes <- publishRows batch
                markProcessedOutcomes policy batch outcomes Nothing

    publishStopTheLine ::
        [OutboxRow] ->
        [OutboxRow] ->
        [(OutboxId, PublishOutcome)] ->
        Maybe OutboxId ->
        Eff es OutboxPublishSummary
    publishStopTheLine original [] outcomes halted =
        markProcessedOutcomes StopTheLine original (Map.fromList outcomes) halted
    publishStopTheLine original (row : rest) outcomes _ = do
        result <- publishRows [row]
        let outcome = outcomeFor result row
            outcomes' = outcomes <> [(row ^. #outboxId, outcome)]
        case outcome of
            PublishSucceeded -> publishStopTheLine original rest outcomes' Nothing
            PublishFailed _ -> markProcessedOutcomes StopTheLine original (Map.fromList outcomes') (Just (row ^. #outboxId))

    publishRows :: [OutboxRow] -> Eff es (Map.Map OutboxId PublishOutcome)
    publishRows [] = pure Map.empty
    publishRows batch@(firstRow : _) =
        Map.fromList
            <$> withBatchSpan
                batch
                firstRow
                ( do
                    attempted <- trySync (publish batch)
                    let normalized =
                            case attempted of
                                Left err ->
                                    let errMsg = Text.pack (displayException err)
                                     in [(row ^. #outboxId, PublishFailed errMsg) | row <- batch]
                                Right reported ->
                                    normalizeOutcomes batch reported
                    pure normalized
                )

    withBatchSpan ::
        [OutboxRow] ->
        OutboxRow ->
        Eff es [(OutboxId, PublishOutcome)] ->
        Eff es [(OutboxId, PublishOutcome)]
    withBatchSpan batch firstRow action =
        withProducerSpan
            (options ^. #tracer)
            (firstRow ^. #event)
            (outboxRowToKafkaRecord firstRow)
            $ \mSpan -> do
                for_ mSpan $ \sp ->
                    addAttribute sp (unkey keiro_outbox_batch_size) (fromIntegral (length batch) :: Int64)
                outcomes <- action
                case (mSpan, firstFailure outcomes) of
                    (Just sp, Just errMsg) -> do
                        addAttribute sp (unkey error_type) ("publish_failed" :: Text)
                        setStatus sp (Error errMsg)
                    _ -> pure ()
                pure outcomes

    normalizeOutcomes :: [OutboxRow] -> [(OutboxId, PublishOutcome)] -> [(OutboxId, PublishOutcome)]
    normalizeOutcomes batch reported =
        let reportedMap = Map.fromList reported
         in [ (row ^. #outboxId, outcomeFor reportedMap row)
            | row <- batch
            ]

    outcomeFor :: Map.Map OutboxId PublishOutcome -> OutboxRow -> PublishOutcome
    outcomeFor outcomes row =
        fromMaybe (PublishFailed "publisher returned no outcome") $
            Map.lookup (row ^. #outboxId) outcomes

    firstFailure :: [(OutboxId, PublishOutcome)] -> Maybe Text
    firstFailure [] = Nothing
    firstFailure ((_, PublishSucceeded) : rest) = firstFailure rest
    firstFailure ((_, PublishFailed errMsg) : _) = Just errMsg

    markProcessedOutcomes ::
        OrderingPolicy ->
        [OutboxRow] ->
        Map.Map OutboxId PublishOutcome ->
        Maybe OutboxId ->
        Eff es OutboxPublishSummary
    markProcessedOutcomes policy batch outcomes halted = do
        now <- liftIO getCurrentTime
        let marks = foldMap (groupMarks outcomes) (outcomeGroups policy batch)
            sentIds = marks ^. #sentIds
            failedRows = marks ^. #failedRows
            skippedRows = marks ^. #skippedRows
        failedStatuses <-
            if null failedRows && null skippedRows
                then pure []
                else runTransaction $ do
                    statuses <- traverse (markFailed now) failedRows
                    traverse_ (markSkipped now) skippedRows
                    pure statuses
        _ <- markOutboxSentBatch sentIds now
        let deadCount = length [() | OutboxDead <- failedStatuses]
            retriedFailures = length failedStatuses - deadCount
        pure
            OutboxPublishSummary
                { claimed = length batch
                , published = length sentIds
                , retried = retriedFailures + length skippedRows
                , dead = deadCount
                , haltedOn = halted
                }

    markFailed :: UTCTime -> (OutboxRow, Text) -> Tx.Transaction OutboxStatus
    markFailed now (row, errMsg) =
        markOutboxFailedTx
            (row ^. #outboxId)
            errMsg
            (options ^. #maxAttempts)
            (nextDelay (options ^. #backoff) (row ^. #attemptCount))
            now

    markSkipped :: UTCTime -> OutboxRow -> Tx.Transaction ()
    markSkipped now row =
        markOutboxSkippedTx
            (row ^. #outboxId)
            "skipped: earlier record for the same key failed"
            now

{- | Reclaim crashed publisher rows and record the outbox backlog gauge.

Schedule this pass independently from 'publishClaimedOutbox', typically on a
slower timer. It is the only library worker path that reclaims rows stranded in
@publishing@.
-}
outboxMaintenancePass ::
    (IOE :> es, Store :> es) =>
    OutboxMaintenanceOptions ->
    Maybe KeiroMetrics ->
    Eff es OutboxMaintenanceSummary
outboxMaintenancePass options mMetrics = do
    now <- liftIO getCurrentTime
    (requeued, deadLettered) <-
        requeueStuckOutbox
            (options ^. #maxAttempts)
            (options ^. #publishingTimeout)
            now
    recordOutboxReclaimed mMetrics (fromIntegral requeued)
    recordOutboxDeadlettered mMetrics (fromIntegral deadLettered)
    backlog <- countOutboxBacklog
    recordOutboxBacklog mMetrics (fromIntegral backlog)
    pure OutboxMaintenanceSummary{requeued, deadLettered, backlog}

-- | Count publishable rows and record the outbox backlog gauge when metrics are enabled.
sampleOutboxBacklog :: (IOE :> es, Store :> es) => Maybe KeiroMetrics -> Eff es ()
sampleOutboxBacklog Nothing = pure ()
sampleOutboxBacklog (Just metrics) = do
    backlog <- countOutboxBacklog
    recordOutboxBacklog (Just metrics) (fromIntegral backlog)

data OutcomeMarks = OutcomeMarks
    { sentIds :: ![OutboxId]
    , failedRows :: ![(OutboxRow, Text)]
    , skippedRows :: ![OutboxRow]
    }
    deriving stock (Generic)

instance Semigroup OutcomeMarks where
    left <> right =
        OutcomeMarks
            { sentIds = (left ^. #sentIds) <> (right ^. #sentIds)
            , failedRows = (left ^. #failedRows) <> (right ^. #failedRows)
            , skippedRows = (left ^. #skippedRows) <> (right ^. #skippedRows)
            }

instance Monoid OutcomeMarks where
    mempty = OutcomeMarks{sentIds = [], failedRows = [], skippedRows = []}

groupMarks :: Map.Map OutboxId PublishOutcome -> [OutboxRow] -> OutcomeMarks
groupMarks outcomes = go []
  where
    go sent [] = mempty{sentIds = sent}
    go sent (row : rest) =
        case fromMaybe (PublishFailed "publisher returned no outcome") (Map.lookup (row ^. #outboxId) outcomes) of
            PublishSucceeded -> go (sent <> [row ^. #outboxId]) rest
            PublishFailed errMsg ->
                OutcomeMarks
                    { sentIds = sent
                    , failedRows = [(row, errMsg)]
                    , skippedRows = rest
                    }

data OutcomeGroupKey
    = BatchGroup
    | SourceGroup !Text
    | RowGroup !OutboxId
    | KeyGroup !Text !Text
    deriving stock (Generic, Eq)

data OutcomeGroup = OutcomeGroup
    { groupKey :: !OutcomeGroupKey
    , groupRows :: ![OutboxRow]
    }
    deriving stock (Generic)

outcomeGroups :: OrderingPolicy -> [OutboxRow] -> [[OutboxRow]]
outcomeGroups policy =
    fmap (^. #groupRows) . foldl' addGroup []
  where
    addGroup groups row =
        appendGroup (keyFor row) row groups
    keyFor row =
        case policy of
            -- A claimed batch can hold runs from several independent sources;
            -- a failure in one source's run must not skip another source's rows.
            PerSourceStream -> SourceGroup (row ^. #event . #source)
            -- Deliberately one group: any failure halts the worker, and the
            -- entire remaining batch is skipped without consuming attempts.
            StopTheLine -> BatchGroup
            BestEffort -> RowGroup (row ^. #outboxId)
            _ ->
                case row ^. #event . #key of
                    Nothing -> RowGroup (row ^. #outboxId)
                    Just key -> KeyGroup (row ^. #event . #source) key

appendGroup :: OutcomeGroupKey -> OutboxRow -> [OutcomeGroup] -> [OutcomeGroup]
appendGroup key row [] = [OutcomeGroup{groupKey = key, groupRows = [row]}]
appendGroup key row (group : rest)
    | group ^. #groupKey == key =
        (group & #groupRows %~ (<> [row])) : rest
    | otherwise =
        group : appendGroup key row rest