Types.hs

{- | Shared types for the idempotent integration-event inbox.

The inbox lives in the consuming bounded context. When a Kafka consumer
receives an integration event, the inbox records a stable external
identity for that message and runs the local handler in the same
transaction. Duplicate redeliveries (Kafka offset retry, rebalance,
producer republish) become observable as duplicates instead of
re-running the handler.
-}
module Keiro.Inbox.Types (
    InboxDedupePolicy (..),
    InboxPersistence (..),
    InboxStatus (..),
    InboxResult (..),
    InboxError (..),
    InboxRow (..),
    KafkaDeliveryRef (..),
    inboxStatusText,
    parseInboxStatus,
    dedupeKeyFor,
)
where

import Data.Text qualified as Text
import Data.UUID qualified as UUID
import Keiro.Integration.Event (IntegrationEvent)
import Keiro.Prelude
import Kiroku.Store.Types (EventId (..), GlobalPosition (..))

{- | Which identity is used as the inbox primary key for an
'IntegrationEvent'.

* 'PreferIntegrationMessageId' (default) — use the application-level
  @messageId@ minted at the producer's outbox enqueue. EP-19 / EP-20
  keep this id stable across publish retries, so it is the natural
  primary dedupe key for Kafka-delivered events.
* 'PreferSourceEventIdentity' — use the @sourceEventId@ of the private
  event that produced this integration event. Useful when a producer
  may emit the same logical fact under different @messageId@s (e.g.
  schema-upgrade republish), and the consumer wants those republishes
  collapsed to a single handler run.
* 'KafkaDeliveryIdentity' — use the Kafka topic-partition-offset triple
  as the dedupe key. Fallback only when neither @messageId@ nor source
  identity is available. This identifies one broker delivery, not one
  logical producer message: if a producer republishes the same logical
  message, Kafka assigns a new offset and this policy will not collapse
  the republish.
* 'CustomDedupeKey' — caller supplies the key. Use only when the other
  policies cannot represent the identity scheme; the consuming service
  owns key collision resistance.
-}
data InboxDedupePolicy
    = PreferIntegrationMessageId
    | PreferSourceEventIdentity
    | KafkaDeliveryIdentity
    | CustomDedupeKey !Text
    deriving stock (Generic, Eq, Show)

{- | How much of the integration-event envelope the inbox persists on the
success path.

The failure path always persists the full envelope because a failed
inbox row is the operator's dead-letter record.
-}
data InboxPersistence
    = PersistFullEnvelope
    | PersistDedupeOnly
    deriving stock (Generic, Eq, Show)

{- | Lifecycle state of an inbox row.

* 'InboxProcessing' — legacy on-disk state from older wrappers and
  reserved for future async paths. Current single-transaction intake
  inserts fresh successful rows directly as 'InboxCompleted'.
* 'InboxCompleted' — handler ran to completion; terminal.
* 'InboxFailed' — handler signaled a permanent failure; terminal. The
  caller is responsible for operator action (dead-letter, manual
  retry).
-}
data InboxStatus
    = InboxProcessing
    | InboxCompleted
    | InboxFailed
    deriving stock (Generic, Eq, Show)

{- | The classified outcome of 'Keiro.Inbox.runInboxTransaction'.

* 'InboxProcessed a' — first delivery; handler ran and returned @a@.
* 'InboxDuplicate' — a previous delivery already completed; handler not
  run.
* 'InboxInProgress' — a previous attempt is currently in-flight, or a
  legacy @processing@ row was read. Current single-transaction intake
  does not commit @processing@ rows. Treat as transient.
* 'InboxPreviouslyFailed' — a previous attempt recorded a permanent
  failure. Operator should review before reprocessing.
-}
data InboxResult a
    = InboxProcessed !a
    | InboxDuplicate
    | InboxInProgress
    | InboxPreviouslyFailed !(Maybe Text)
    | InboxHandlerFailed !Text !Int
    deriving stock (Generic, Eq, Show)

{- | Errors surfaced by the inbox wrapper that originate from the inbox
itself rather than from the supplied handler.
-}
data InboxError
    = -- | The integration event lacked the field required by the chosen policy.
      DedupePolicyUnsatisfied !InboxDedupePolicy
    deriving stock (Generic, Eq, Show)

{- | Optional Kafka-delivery metadata recorded alongside an inbox row.

Used by 'KafkaDeliveryIdentity' to compute the dedupe key, and stored on
the row regardless of policy so operators can correlate the inbox
record with Kafka logs. Not part of the EP-19 envelope.
-}
data KafkaDeliveryRef = KafkaDeliveryRef
    { topic :: !Text
    , partition :: !Int64
    , offset :: !Int64
    }
    deriving stock (Generic, Eq, Show)

{- | One row read back from @keiro_inbox@.

Rows written with 'PersistDedupeOnly' decode with an empty
'IntegrationEvent' @payloadBytes@, no attributes, no trace context, and
no schema reference. Identity, routing, source-event ids, occurrence
time, and Kafka delivery metadata are still preserved.
-}
data InboxRow = InboxRow
    { source :: !Text
    , dedupeKey :: !Text
    , event :: !IntegrationEvent
    , kafka :: !(Maybe KafkaDeliveryRef)
    , status :: !InboxStatus
    , attemptCount :: !Int
    , receivedAt :: !UTCTime
    , completedAt :: !(Maybe UTCTime)
    , failedAt :: !(Maybe UTCTime)
    , lastError :: !(Maybe Text)
    }
    deriving stock (Generic, Eq, Show)

inboxStatusText :: InboxStatus -> Text
inboxStatusText = \case
    InboxProcessing -> "processing"
    InboxCompleted -> "completed"
    InboxFailed -> "failed"

parseInboxStatus :: Text -> Either Text InboxStatus
parseInboxStatus = \case
    "processing" -> Right InboxProcessing
    "completed" -> Right InboxCompleted
    "failed" -> Right InboxFailed
    other -> Left ("unknown keiro_inbox.status: " <> other)

{- | Compute the inbox dedupe key for an integration event under the
given policy plus optional Kafka delivery context. Returns 'Left' when
the policy demands a field the envelope does not carry (for example,
'PreferSourceEventIdentity' on an envelope with no
@sourceEventId@).
-}
dedupeKeyFor ::
    InboxDedupePolicy ->
    IntegrationEvent ->
    Maybe KafkaDeliveryRef ->
    Either InboxError Text
dedupeKeyFor policy event kafka = case policy of
    PreferIntegrationMessageId ->
        let mid = event ^. #messageId
         in if Text.null mid
                then Left (DedupePolicyUnsatisfied policy)
                else Right mid
    PreferSourceEventIdentity ->
        case event ^. #sourceEventId of
            Just (EventId u) -> Right (UUID.toText u)
            Nothing ->
                case event ^. #sourceGlobalPosition of
                    Just (GlobalPosition p) -> Right (Text.pack (show p))
                    Nothing -> Left (DedupePolicyUnsatisfied policy)
    KafkaDeliveryIdentity ->
        case kafka of
            Just ref ->
                Right
                    ( (ref ^. #topic)
                        <> ":"
                        <> Text.pack (show (ref ^. #partition))
                        <> ":"
                        <> Text.pack (show (ref ^. #offset))
                    )
            Nothing -> Left (DedupePolicyUnsatisfied policy)
    CustomDedupeKey k ->
        if Text.null k
            then Left (DedupePolicyUnsatisfied policy)
            else Right k