Inbox.hs

-- | Idempotent inbox for cross-bounded-context integration events.
--
-- The inbox lives in the consuming bounded context. When a Kafka consumer
-- receives an integration event, the inbox records the event's stable
-- external identity and runs the local handler in the same Postgres
-- transaction. Duplicate redeliveries (Kafka offset retry, rebalance,
-- producer republish) become observable as duplicates instead of
-- re-running the handler.
--
-- The wrapper is a single-transaction primitive: the completed inbox row
-- and the handler's local writes commit atomically. If the handler raises
-- or condemns the transaction, the inbox row never appears and the next
-- delivery starts fresh.
--
-- Completed-row retention defines the deduplication window. After
-- 'garbageCollectCompleted' removes a row, a later delivery of the same key is
-- processed again. A concurrent GC can also delete a conflicting completed row
-- between the insert attempt and its lookup; the handler then commits without a
-- replacement deduplication row, so a later redelivery can run it again. These
-- cases preserve at-least-once delivery, not permanent exactly-once processing;
-- size retention beyond the maximum redelivery delay and keep handlers
-- idempotent.
module Keiro.Inbox
  ( -- * Re-exports
    module Keiro.Inbox.Types,

    -- * Storage primitives
    lookupInbox,
    listInbox,
    listCompletedInboxGcCandidates,
    garbageCollectCompleted,
    countInboxBacklog,
    markFailedTx,

    -- * Transactional handler wrapper
    runInboxTransaction,
    runInboxTransactionWith,
    runInboxTransactionWithKey,
    runInboxTransactionWithRetries,
    runInboxTransactionWithRetriesWith,
    runInboxTransactionWithRetriesKey,
    runInboxTransactionBatch,
    runInboxDelegated,
    runInboxDelegatedWithRetries,
    runInboxDelegatedBatch,
    sampleInboxBacklog,
  )
where

import Data.Map.Strict qualified as Map
import Data.Maybe (listToMaybe, mapMaybe)
import Data.Set qualified as Set
import Data.Text qualified as Text
import Effectful (Eff, IOE, (:>))
import Effectful.Exception (displayException, trySync)
import Keiro.Inbox.Schema
import Keiro.Inbox.Types
import Keiro.Integration.Event (IntegrationEvent)
import Keiro.Prelude
import Keiro.Telemetry
  ( KeiroMetrics,
    recordInboxBacklog,
    recordInboxDuplicates,
    recordInboxFailed,
    recordInboxPoisoned,
    recordInboxProcessed,
  )
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import "hasql-transaction" Hasql.Transaction qualified as Tx

data BatchPlan
  = BatchKeyError !InboxError
  | BatchDuplicate
  | BatchWork !Text !Text !IntegrationEvent !(Maybe KafkaDeliveryRef)

-- | Run @handler@ at most once for each @(source, dedupe_key)@.
--
-- Computes the dedupe key from @policy@ and @kafka@, then in one
-- transaction:
--
-- * Inserts the inbox row with status @completed@.
-- * If the row already exists, branches on its status: 'InboxCompleted'
--   → 'InboxDuplicate'; 'InboxProcessing' → 'InboxInProgress';
--   'InboxFailed' → 'InboxPreviouslyFailed'.
-- * Otherwise runs @handler@; the row commits only if the handler succeeds.
--
-- The handler is invoked with the decoded 'IntegrationEvent' so it does
-- not need to redecode bytes. On exception or 'Tx.condemn' the whole
-- transaction rolls back, including the inbox row insert — the next
-- delivery sees no row and can retry.
runInboxTransaction ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (Either InboxError (InboxResult a))
runInboxTransaction mMetrics policy event kafka handler =
  runInboxTransactionWith mMetrics PersistFullEnvelope policy event kafka handler

-- | Variant of 'runInboxTransaction' that controls success-path
-- envelope persistence.
--
-- 'PersistDedupeOnly' keeps enough columns for dedupe and operator
-- correlation but stores an empty payload and omits schema, trace, and
-- attribute columns for successfully processed rows.
runInboxTransactionWith ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  InboxPersistence ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (Either InboxError (InboxResult a))
runInboxTransactionWith mMetrics persistence policy event kafka handler =
  case dedupeKeyFor policy event kafka of
    Left err -> pure (Left err)
    Right dedupe ->
      Right
        <$> runInboxTransactionWithKeyPersist
          mMetrics
          persistence
          (event ^. #source)
          dedupe
          event
          kafka
          handler

-- | Lower-level variant that takes the dedupe key directly.
--
-- Use when the policy is not enough to express the identity scheme — for
-- example, when the consumer joins fields from multiple headers or
-- derives the key from the payload itself.
runInboxTransactionWithKey ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Text ->
  Text ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (InboxResult a)
runInboxTransactionWithKey mMetrics src dedupe event kafka handler =
  runInboxTransactionWithKeyPersist mMetrics PersistFullEnvelope src dedupe event kafka handler

runInboxTransactionWithKeyPersist ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  InboxPersistence ->
  Text ->
  Text ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (InboxResult a)
runInboxTransactionWithKeyPersist mMetrics persistence src dedupe event kafka handler = do
  now <- liftIO getCurrentTime
  result <-
    runTransaction $
      attemptOneTx persistence Nothing src dedupe event kafka now handler
  -- Record the classification counter outside the handler transaction.
  -- Backlog gauge sampling is intentionally scheduled separately via
  -- 'sampleInboxBacklog'.
  recordInboxResult mMetrics Nothing result
  pure result

-- | Run @handler@ with opt-in poison-message accounting.
--
-- This wrapper behaves like 'runInboxTransaction' for fresh messages,
-- duplicates, and in-flight rows, but changes the behavior for handler
-- exceptions and previously failed rows:
--
-- * A synchronous exception from @handler@ rolls back the handler
--   transaction, then records a failed attempt in a second transaction and
--   returns 'InboxHandlerFailed' with the new attempt count.
-- * A previously failed row with @attempt_count < ceiling@ is retried.
-- * A previously failed row with @attempt_count >= ceiling@ returns
--   'InboxPreviouslyFailed' without running the handler. The consumer can
--   commit its offset and move on; the failed inbox row is the dead-letter
--   record for operator review.
--
-- 'Tx.condemn' is not treated as a handler failure by this wrapper. It
-- keeps the original rollback semantics from 'runInboxTransaction'.
runInboxTransactionWithRetries ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Int ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (Either InboxError (InboxResult a))
runInboxTransactionWithRetries mMetrics attemptCeiling policy event kafka handler =
  runInboxTransactionWithRetriesWith mMetrics attemptCeiling PersistFullEnvelope policy event kafka handler

-- | Variant of 'runInboxTransactionWithRetries' that controls success-path persistence.
runInboxTransactionWithRetriesWith ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Int ->
  InboxPersistence ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (Either InboxError (InboxResult a))
runInboxTransactionWithRetriesWith mMetrics attemptCeiling persistence policy event kafka handler =
  case dedupeKeyFor policy event kafka of
    Left err -> pure (Left err)
    Right dedupe ->
      Right
        <$> runInboxTransactionWithRetriesKeyPersist
          mMetrics
          attemptCeiling
          persistence
          (event ^. #source)
          dedupe
          event
          kafka
          handler

-- | Lower-level retrying variant that takes the dedupe key directly.
runInboxTransactionWithRetriesKey ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Int ->
  Text ->
  Text ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (InboxResult a)
runInboxTransactionWithRetriesKey mMetrics attemptCeiling src dedupe event kafka handler =
  runInboxTransactionWithRetriesKeyPersist mMetrics attemptCeiling PersistFullEnvelope src dedupe event kafka handler

runInboxTransactionWithRetriesKeyPersist ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Int ->
  InboxPersistence ->
  Text ->
  Text ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es (InboxResult a)
runInboxTransactionWithRetriesKeyPersist mMetrics attemptCeiling persistence src dedupe event kafka handler = do
  now <- liftIO getCurrentTime
  attempted <-
    trySync $
      runTransaction $
        attemptOneTx persistence (Just attemptCeiling) src dedupe event kafka now handler
  result <- case attempted of
    Right ok -> pure ok
    Left err -> do
      failedAt <- liftIO getCurrentTime
      let errMsg = Text.pack (displayException err)
      attempts <-
        runTransaction $
          recordFailedAttemptTx src dedupe event kafka errMsg failedAt
      pure (InboxHandlerFailed errMsg attempts)
  recordInboxResult mMetrics (Just attemptCeiling) result
  pure result

-- | Process a batch of inbox deliveries with a single transactional fast path.
--
-- The fast path computes each @(source, dedupe_key)@, suppresses repeated
-- keys within the batch as duplicates, then runs all remaining deliveries
-- in one Postgres transaction. If any handler throws or condemns that
-- transaction, the whole batch rolls back and every original delivery is
-- retried through 'runInboxTransactionWithRetries'. That fallback preserves
-- per-message failure accounting and prevents one poison message from
-- discarding unrelated batch mates.
--
-- 'Tx.condemn' rolls the transaction back at commit but returns normally,
-- so it cannot be observed from the transaction's return value alone.
-- The batch detects it by re-reading one row it should have committed:
-- every write in the fast path belongs to a delivery classified
-- 'InboxProcessed' (fresh insert as @completed@ or retry promotion to
-- @completed@), so if the first such row is not @completed@ after the
-- transaction returns, the whole batch was condemned and the per-message
-- fallback runs. A batch with no 'InboxProcessed' rows performed no writes,
-- so a condemned transaction loses nothing.
runInboxTransactionBatch ::
  forall a es.
  (IOE :> es, Store :> es) =>
  Maybe KeiroMetrics ->
  Int ->
  InboxDedupePolicy ->
  InboxPersistence ->
  [(IntegrationEvent, Maybe KafkaDeliveryRef)] ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Eff es [Either InboxError (InboxResult a)]
runInboxTransactionBatch mMetrics attemptCeiling policy persistence deliveries handler = do
  now <- liftIO getCurrentTime
  let plan = planInboxBatch policy deliveries
  attempted <-
    trySync $
      runTransaction $
        traverse
          ( \case
              BatchKeyError err -> pure (Left err)
              BatchDuplicate -> pure (Right InboxDuplicate)
              BatchWork src dedupe event kafka ->
                Right <$> attemptOneTx persistence (Just attemptCeiling) src dedupe event kafka now handler
          )
          plan
  case attempted of
    Right results -> do
      committed <- verifyBatchCommitted plan results
      if committed
        then do
          for_ results $ \case
            Right result -> recordInboxResult mMetrics (Just attemptCeiling) result
            Left _ -> pure ()
          pure results
        else perMessageFallback
    Left _ -> perMessageFallback
  where
    perMessageFallback :: Eff es [Either InboxError (InboxResult a)]
    perMessageFallback =
      traverse
        ( \(event, kafka) ->
            runInboxTransactionWithRetriesWith mMetrics attemptCeiling persistence policy event kafka handler
        )
        deliveries

    -- A condemned transaction returns its results normally but commits
    -- nothing. Re-read the first row the batch claims to have completed;
    -- if it is not @completed@, the transaction rolled back at commit.
    verifyBatchCommitted ::
      [BatchPlan] ->
      [Either InboxError (InboxResult a)] ->
      Eff es Bool
    verifyBatchCommitted plan results =
      case listToMaybe (mapMaybe processedKey (zip plan results)) of
        Nothing -> pure True
        Just (src, dedupe) -> do
          row <- lookupInbox src dedupe
          pure (fmap (^. #status) row == Just InboxCompleted)

    processedKey :: (BatchPlan, Either InboxError (InboxResult a)) -> Maybe (Text, Text)
    processedKey = \case
      (BatchWork src dedupe _ _, Right (InboxProcessed _)) -> Just (src, dedupe)
      _ -> Nothing

attemptOneTx ::
  InboxPersistence ->
  Maybe Int ->
  Text ->
  Text ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  UTCTime ->
  (IntegrationEvent -> Tx.Transaction a) ->
  Tx.Transaction (InboxResult a)
attemptOneTx persistence attemptCeiling src dedupe event kafka now handler = do
  inserted <- tryInsertCompletedTx persistence src dedupe event kafka now
  case inserted of
    Right () -> do
      handled <- handler event
      pure (InboxProcessed handled)
    Left row -> case row ^. #status of
      InboxCompleted -> pure InboxDuplicate
      InboxProcessing -> pure InboxInProgress
      InboxFailed -> case attemptCeiling of
        Nothing -> pure (InboxPreviouslyFailed (row ^. #lastError))
        Just attemptLimit
          | row ^. #attemptCount >= attemptLimit ->
              pure (InboxPreviouslyFailed (row ^. #lastError))
          | otherwise -> do
              handled <- handler event
              markCompletedTx src dedupe now
              pure (InboxProcessed handled)

recordInboxResult :: (IOE :> es) => Maybe KeiroMetrics -> Maybe Int -> InboxResult a -> Eff es ()
recordInboxResult mMetrics attemptCeiling = \case
  InboxProcessed _ -> recordInboxProcessed mMetrics 1
  InboxDuplicate -> recordInboxDuplicates mMetrics 1
  InboxPreviouslyFailed _ -> recordInboxFailed mMetrics 1
  InboxHandlerFailed _ attempts -> do
    recordInboxFailed mMetrics 1
    case attemptCeiling of
      Just attemptLimit | attempts >= attemptLimit -> recordInboxPoisoned mMetrics 1
      _ -> pure ()
  InboxInProgress -> pure ()

planInboxBatch ::
  InboxDedupePolicy ->
  [(IntegrationEvent, Maybe KafkaDeliveryRef)] ->
  [BatchPlan]
planInboxBatch policy = go Map.empty
  where
    go _ [] = []
    go seen ((event, kafka) : rest) =
      case dedupeKeyFor policy event kafka of
        Left err -> BatchKeyError err : go seen rest
        Right dedupe ->
          let key = (event ^. #source, dedupe)
           in if Map.member key seen
                then BatchDuplicate : go seen rest
                else BatchWork (event ^. #source) dedupe event kafka : go (Map.insert key () seen) rest

-- | Run an integration handler whose downstream operation owns the durable
-- deduplication receipt.
--
-- This wrapper computes the same policy key as the table-backed inbox, but it
-- performs no inbox reads or writes and requires no 'Store' effect. The handler
-- receives the computed key and must cover every protected effect with that
-- identity. A returned 'DelegatedOutcome' is the handler's assertion about the
-- downstream result; it is not independently verified by this wrapper.
--
-- Synchronous and asynchronous exceptions both propagate. Use
-- 'runInboxDelegatedWithRetries' when synchronous failures should be classified
-- for a caller-owned retry ladder.
runInboxDelegated ::
  forall a es.
  (IOE :> es) =>
  Maybe KeiroMetrics ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (Text -> IntegrationEvent -> Eff es (DelegatedOutcome a)) ->
  Eff es (Either InboxError (InboxResult a))
runInboxDelegated mMetrics policy event kafka handler =
  case dedupeKeyFor policy event kafka of
    Left err -> pure (Left err)
    Right dedupe -> do
      result <- delegatedResult <$> handler dedupe event
      recordInboxResult mMetrics Nothing result
      pure (Right result)

-- | Run delegated intake with an explicit, caller-owned retry position.
--
-- At an attempt above the configured ceiling, the handler is not invoked and
-- the result is 'InboxPreviouslyFailed'. At or below the ceiling, synchronous
-- exceptions become 'InboxHandlerFailed' with the current attempt number;
-- asynchronous cancellation still propagates. Typed errors in the handler's
-- effect stack are not exceptions and must be handled by the caller.
runInboxDelegatedWithRetries ::
  forall a es.
  (IOE :> es) =>
  Maybe KeiroMetrics ->
  DelegatedRetryContext ->
  InboxDedupePolicy ->
  IntegrationEvent ->
  Maybe KafkaDeliveryRef ->
  (Text -> IntegrationEvent -> Eff es (DelegatedOutcome a)) ->
  Eff es (Either InboxError (InboxResult a))
runInboxDelegatedWithRetries mMetrics retryContext policy event kafka handler =
  case dedupeKeyFor policy event kafka of
    Left err -> pure (Left err)
    Right dedupe -> do
      let attemptLimit = delegatedRetryCeiling retryContext
          attempt = delegatedRetryAttempt retryContext
      result <-
        if attempt > attemptLimit
          then pure (InboxPreviouslyFailed Nothing)
          else do
            attempted <- trySync (handler dedupe event)
            pure $ case attempted of
              Right outcome -> delegatedResult outcome
              Left err -> InboxHandlerFailed (Text.pack (displayException err)) attempt
      recordInboxResult mMetrics (Just attemptLimit) result
      pure (Right result)

-- | Process a bounded chunk of delegated deliveries sequentially.
--
-- Successful identities are remembered only for this call, keyed by source and
-- dedupe key. A later occurrence of a successful identity is classified as a
-- duplicate without invoking the handler. Policy errors and synchronous
-- handler exceptions are returned per item and do not suppress a later retry of
-- the same identity. Async cancellation propagates immediately. This function
-- creates no threads or transactions and retains O(n) results and keys for an
-- input chunk of size n.
runInboxDelegatedBatch ::
  forall a es.
  (IOE :> es) =>
  Maybe KeiroMetrics ->
  InboxDedupePolicy ->
  [(IntegrationEvent, Maybe KafkaDeliveryRef)] ->
  (Text -> IntegrationEvent -> Eff es (DelegatedOutcome a)) ->
  Eff es [Either InboxError (InboxResult a)]
runInboxDelegatedBatch mMetrics policy deliveries handler =
  go Set.empty [] deliveries
  where
    go _ results [] = pure (reverse results)
    go seen results ((event, kafka) : rest) =
      case dedupeKeyFor policy event kafka of
        Left err -> go seen (Left err : results) rest
        Right dedupe -> do
          let identity = (event ^. #source, dedupe)
          if Set.member identity seen
            then do
              recordInboxResult mMetrics Nothing InboxDuplicate
              go seen (Right InboxDuplicate : results) rest
            else do
              attempted <- trySync (handler dedupe event)
              case attempted of
                Left err -> do
                  let result = InboxHandlerFailed (Text.pack (displayException err)) 1
                  recordInboxResult mMetrics Nothing result
                  go seen (Right result : results) rest
                Right outcome -> do
                  let result = delegatedResult outcome
                      seen' = Set.insert identity seen
                  recordInboxResult mMetrics Nothing result
                  seen' `seq` go seen' (Right result : results) rest

delegatedResult :: DelegatedOutcome a -> InboxResult a
delegatedResult = \case
  DelegatedFresh value -> InboxProcessed value
  DelegatedDuplicate -> InboxDuplicate

-- | Count the inbox backlog and record the gauge when metrics are enabled.
--
-- The backlog is non-terminal rows: legacy @processing@ rows plus failed rows.
-- Schedule this on its own interval; it is intentionally not part of the
-- per-message intake path.
sampleInboxBacklog :: (IOE :> es, Store :> es) => Maybe KeiroMetrics -> Eff es ()
sampleInboxBacklog Nothing = pure ()
sampleInboxBacklog (Just metrics) = do
  backlog <- countInboxBacklog
  recordInboxBacklog (Just metrics) (fromIntegral backlog)