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. Versioned source-event coordinates
--   derive both IDs deterministically across producer and publication 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,
    module Keiro.Outbox.Identity,
    deriveProducerIdentity,
    recordProducerEnqueueOutcome,

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

    -- * Inline escape hatch
    freshOutboxId,
    enqueueIntegrationEventTx,

    -- * Canonical producer-subscription helper
    IntegrationProducer (..),
    IntegrationProducerConfigError (..),
    IntegrationEventDraft (..),
    mkIntegrationProducer,
    mintIntegrationEvent,
    freshIntegrationEvent,
    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 Data.Word (Word32)
import Effectful (Eff, IOE, (:>))
import Effectful.Exception (displayException, trySync)
import Keiro.Integration.Event
  ( IntegrationContentType,
    IntegrationEvent (..),
    SchemaReference,
    TraceContext,
  )
import Keiro.Outbox.Identity
import Keiro.Outbox.Kafka (outboxRowToKafkaRecord)
import Keiro.Outbox.Schema
import Keiro.Outbox.Types
import Keiro.Prelude
import Keiro.Telemetry
  ( KeiroMetrics,
    recordOutboxBacklog,
    recordOutboxDeadlettered,
    recordOutboxIdentityConflict,
    recordOutboxPublished,
    recordOutboxReclaimed,
    recordOutboxRejected,
    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. Source-event coordinates
-- derive stable IDs across enqueue and publication 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' — non-empty namespace, validated using TypeID prefix
--   syntax. The resulting deterministic message ID is opaque text, not a TypeID.
--   Prefer 'mkIntegrationProducer'; direct record construction bypasses validation.
-- * '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
  | Text.null (producer ^. #messageIdPrefix) = Left (InvalidMessageIdPrefix "" "namespace must not be empty")
  | otherwise = 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 by 'enqueueProducerEventTx' using deterministic identity.
--
-- @sourceEventId@ and @sourceGlobalPosition@ default to the values on the
-- underlying 'RecordedEvent' (see 'enqueueProducerEventTx'); 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 = freshIntegrationEvent
{-# DEPRECATED mintIntegrationEvent "Use enqueueProducerEventTx for replay-safe producer identity, or freshIntegrationEvent for explicitly fresh envelopes." #-}

-- | Generate an explicitly fresh envelope. Persist it before retrying; this
-- helper alone provides no producer replay identity or provenance defaulting.
freshIntegrationEvent :: (IOE :> es) => IntegrationProducer e -> IntegrationEventDraft -> Eff es IntegrationEvent
freshIntegrationEvent 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
    }

-- | Observe a completed enqueue attempt outside its transaction. Invoke once
-- after the runner returns; SQL serialization retries do not multiply metrics.
recordProducerEnqueueOutcome :: (MonadIO m) => Maybe KeiroMetrics -> ProducerEnqueueOutcome -> m ()
recordProducerEnqueueOutcome metrics = \case
  ProducerIdentityConflict {} -> recordOutboxIdentityConflict metrics 1
  _ -> pure ()

-- | Pure identity for one stable producer/source-event coordinate.
deriveProducerIdentity :: IntegrationProducer e -> ProducerEventKey -> ProducerIdentity
deriveProducerIdentity producer = deriveIdentity (producer ^. #source) (producer ^. #name) (producer ^. #messageIdPrefix)

-- | Enqueue a source event emission. Use index zero for today's single-draft
-- mapper. Missing source provenance defaults from the recorded event. On a
-- conflict, callers should condemn the surrounding checkpoint transaction and
-- report the returned field classes after the transaction completes.
-- Suppression is bounded by outbox retention; wire identity survives GC.
enqueueProducerEventTx ::
  IntegrationProducer e ->
  RecordedEvent ->
  Word32 ->
  IntegrationEventDraft ->
  Tx.Transaction ProducerEnqueueOutcome
enqueueProducerEventTx producer recorded emission draft =
  enqueueProducerOutboxTx identity event
  where
    identity = deriveProducerIdentity producer (ProducerEventKey (recorded ^. #eventId) emission)
    withProvenance =
      draft
        & #sourceEventId
        .~ ((draft ^. #sourceEventId) <|> Just (recorded ^. #eventId))
        & #sourceGlobalPosition
        .~ ((draft ^. #sourceGlobalPosition) <|> Just (recorded ^. #globalPosition))
    event = normalizeProducerEvent (draftToEvent (producer ^. #source) (identity ^. #messageId) withProvenance)

-- ---------------------------------------------------------------------------
-- 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
  | -- | The transport intentionally and permanently refused publication.
    PublishRejected !PublishRejection
  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))
  recordOutboxRejected mMetrics (fromIntegral (summary ^. #rejected))
  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, rejected = 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
        PublishRejected _ -> 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 ((_, PublishRejected _) : 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
          rejectedRows = marks ^. #rejectedRows
          failedRows = marks ^. #failedRows
          skippedRows = marks ^. #skippedRows
      committed <- runTransaction $ do
        publishedCount <- markOutboxSentBatchTx sentIds now
        rejectionMarks <- traverse (markRejected now) rejectedRows
        failureMarks <- traverse (markFailed now) failedRows
        skippedMarks <- traverse (markSkipped now) skippedRows
        let failedStatuses = [status | Just status <- failureMarks]
            deadCount = length [() | OutboxDead <- failedStatuses]
        pure
          CommittedMarks
            { published = publishedCount,
              rejected = length [() | True <- rejectionMarks],
              retried = length [() | OutboxFailed <- failedStatuses] + length [() | True <- skippedMarks],
              dead = deadCount
            }
      pure
        OutboxPublishSummary
          { claimed = length batch,
            published = committed ^. #published,
            rejected = committed ^. #rejected,
            retried = committed ^. #retried,
            dead = committed ^. #dead,
            haltedOn = halted
          }

    markRejected :: UTCTime -> (OutboxId, PublishRejection) -> Tx.Transaction Bool
    markRejected now (outboxId, rejection) =
      markOutboxRejectedTx outboxId rejection now

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

    markSkipped :: UTCTime -> OutboxRow -> Tx.Transaction Bool
    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],
    rejectedRows :: ![(OutboxId, PublishRejection)],
    failedRows :: ![(OutboxRow, Text)],
    skippedRows :: ![OutboxRow]
  }
  deriving stock (Generic)

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

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

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

data CommittedMarks = CommittedMarks
  { published :: !Int,
    rejected :: !Int,
    retried :: !Int,
    dead :: !Int
  }
  deriving stock (Generic)

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