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,
    garbageCollectCompleted,
    countInboxBacklog,
    markFailedTx,

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

import Data.Map.Strict qualified as Map
import Data.Maybe (listToMaybe, mapMaybe)
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

-- | 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)