Types.hs

-- | Shared types for the durable integration-event outbox.
--
-- The outbox is the durable handoff between "this service has decided to
-- publish this integration event" and "this service has actually published
-- it". The producer subscription writes one outbox row per mapped private
-- event; the publisher worker drains rows into Kafka and marks each one
-- sent, retryable, or dead.
module Keiro.Outbox.Types
  ( OutboxId (..),
    OutboxStatus (..),
    OrderingPolicy (..),
    BackoffSchedule (..),
    ExponentialBackoffOptions (..),
    OutboxMessage (..),
    OutboxRow (..),
    OutboxPublishOptions (..),
    OutboxPublishConfigError (..),
    OutboxPublishSummary (..),
    OutboxMaintenanceOptions (..),
    OutboxMaintenanceSummary (..),
    defaultPublishOptions,
    defaultMaintenanceOptions,
    mkOutboxPublishOptions,
    statusText,
    parseStatus,
    nextDelay,
  )
where

import Data.Time.Clock (NominalDiffTime)
import Data.UUID (UUID)
import Data.UUID qualified as UUID
import Keiro.Integration.Event (IntegrationEvent)
import Keiro.Prelude
import OpenTelemetry.Trace.Core (Tracer)

-- | Primary key of a 'keiro_outbox' row. Stable across publish retries.
newtype OutboxId = OutboxId {unOutboxId :: UUID}
  deriving stock (Generic, Eq, Ord, Show)

instance ToJSON OutboxId where
  toJSON = toJSON . UUID.toText . unOutboxId

instance FromJSON OutboxId where
  parseJSON v = do
    text <- parseJSON v
    case UUID.fromText text of
      Nothing -> fail ("OutboxId: not a UUID: " <> show text)
      Just uuid -> pure (OutboxId uuid)

-- | Lifecycle state of an outbox row.
--
-- * 'OutboxPending' — never attempted.
-- * 'OutboxPublishing' — currently held by a publisher worker (between
--   claim and the call to mark sent/failed/dead). Rows left in this state
--   after a worker crash are reclaimed by 'Keiro.Outbox.outboxMaintenancePass'
--   after 'publishingTimeout'.
-- * 'OutboxSent' — Kafka acknowledged the publish; terminal.
-- * 'OutboxFailed' — last attempt failed; will be retried after
--   'next_attempt_at'.
-- * 'OutboxDead' — terminal failure after 'maxAttempts' consecutive
--   failures. Stays in the table for operator inspection.
data OutboxStatus
  = OutboxPending
  | OutboxPublishing
  | OutboxSent
  | OutboxFailed
  | OutboxDead
  deriving stock (Generic, Eq, Show)

-- | Ordering policy enforced by the publisher worker's claim query.
--
-- * 'PerKeyHeadOfLine' (default) — within a @source@, a non-terminal row
--   with key @k@ blocks every later row with the same key. Rows with
--   'Nothing' key bypass the block (Kafka does not promise cross-key order
--   for null-keyed records). One stuck aggregate cannot stall traffic on
--   other aggregates. Ordering is based on @created_at@, which PostgreSQL
--   sets to transaction-start time; callers that concurrently enqueue the
--   same key through escape hatches must serialize those enqueues themselves
--   if commit order matters.
-- * 'PerSourceStream' — within a @source@, any non-terminal row blocks
--   every later row. Use when ordering matters across keys (rare). This has
--   the same @created_at@ concurrency caveat as 'PerKeyHeadOfLine'.
-- * 'StopTheLine' — any failure halts the worker until operator
--   intervention. Use when correctness requires manual review on every
--   failure.
-- * 'BestEffort' — failed rows do not block; explicit opt-in only. Safe
--   only when published events have no per-key/causal relationship.
data OrderingPolicy
  = PerKeyHeadOfLine
  | PerSourceStream
  | StopTheLine
  | BestEffort
  deriving stock (Generic, Eq, Show)

-- | Knobs for 'ExponentialBackoff'. @delay = min maxDelay (initial * multiplier ^ (attempt - 1))@.
data ExponentialBackoffOptions = ExponentialBackoffOptions
  { initial :: !NominalDiffTime,
    maxDelay :: !NominalDiffTime,
    multiplier :: !Double
  }
  deriving stock (Generic, Eq, Show)

-- | Backoff curve used to compute 'next_attempt_at' after a failure.
--
-- * 'ConstantBackoff' — fixed delay between retries.
-- * 'ExponentialBackoff' — exponential growth capped at @maxDelay@.
data BackoffSchedule
  = ConstantBackoff !NominalDiffTime
  | ExponentialBackoff !ExponentialBackoffOptions
  deriving stock (Generic, Eq, Show)

-- | Compute the retry delay for an attempt number (1-based: 1 = first
-- failure, 2 = second failure, …). Used by 'Keiro.Outbox.Schema.markOutboxFailedTx'
-- to derive @next_attempt_at@.
nextDelay :: BackoffSchedule -> Int -> NominalDiffTime
nextDelay (ConstantBackoff delay) _ = delay
nextDelay (ExponentialBackoff opts) attempt =
  let raw = (opts ^. #initial) * realToFrac ((opts ^. #multiplier) ** fromIntegral (max 0 (attempt - 1)))
   in min (opts ^. #maxDelay) raw

-- | A request to enqueue one integration event into the outbox. Callers
-- generate 'outboxId' (use a random UUID for ad-hoc enqueues, a
-- deterministic UUID for idempotent retries from a saga/process manager).
data OutboxMessage = OutboxMessage
  { outboxId :: !OutboxId,
    event :: !IntegrationEvent
  }
  deriving stock (Generic, Eq, Show)

-- | A row read back from @keiro_outbox@. Worker code consumes these to
-- publish to Kafka; tests use them to assert state transitions.
data OutboxRow = OutboxRow
  { outboxId :: !OutboxId,
    event :: !IntegrationEvent,
    status :: !OutboxStatus,
    attemptCount :: !Int,
    nextAttemptAt :: !UTCTime,
    lastError :: !(Maybe Text),
    publishedAt :: !(Maybe UTCTime),
    createdAt :: !UTCTime,
    updatedAt :: !UTCTime
  }
  deriving stock (Generic, Eq, Show)

-- | Knobs that govern one invocation of 'Keiro.Outbox.publishClaimedOutbox'.
--
-- The optional 'tracer' field opts the publisher into OpenTelemetry
-- instrumentation: when present, the publisher opens a @Producer@-kind
-- span around each publish call, attributing the first row's
-- 'IntegrationEvent.destination' (topic), 'IntegrationEvent.messageId',
-- and Kafka key per the messaging semantic conventions. When 'tracer' is
-- 'Nothing' (the default) the publisher emits no spans.
data OutboxPublishOptions = OutboxPublishOptions
  { batchSize :: !Int,
    maxAttempts :: !Int,
    backoff :: !BackoffSchedule,
    orderingPolicy :: !OrderingPolicy,
    publishingTimeout :: !NominalDiffTime,
    tracer :: !(Maybe Tracer)
  }
  deriving stock (Generic)

data OutboxPublishConfigError
  = InvalidOutboxBatchSize !Int
  | InvalidOutboxMaxAttempts !Int
  | InvalidOutboxPublishingTimeout !NominalDiffTime
  | InvalidConstantBackoff !NominalDiffTime
  | InvalidExponentialBackoffInitial !NominalDiffTime
  | InvalidExponentialBackoffMultiplier !Double
  | InvalidExponentialBackoffMaxDelay !NominalDiffTime !NominalDiffTime
  deriving stock (Generic, Eq, Show)

-- | Aggregate result of one publisher pass.
--
-- @published + retried + dead@ equals the number of rows claimed. 'retried'
-- includes rows that were skipped because an earlier row in the same ordered
-- publish group failed; those rows are returned to @failed@ without consuming
-- an attempt. 'haltedOn' is populated only by 'StopTheLine' policy and names
-- the failed pivot row, which is already counted in 'retried' or 'dead'.
data OutboxPublishSummary = OutboxPublishSummary
  { claimed :: !Int,
    published :: !Int,
    retried :: !Int,
    dead :: !Int,
    haltedOn :: !(Maybe OutboxId)
  }
  deriving stock (Generic, Eq, Show)

-- | Knobs for 'Keiro.Outbox.outboxMaintenancePass'.
--
-- Schedule maintenance less frequently than publish passes; it owns crash
-- reclamation and backlog gauge sampling.
data OutboxMaintenanceOptions = OutboxMaintenanceOptions
  { maxAttempts :: !Int,
    publishingTimeout :: !NominalDiffTime
  }
  deriving stock (Generic, Eq, Show)

-- | Result of one outbox maintenance pass.
data OutboxMaintenanceSummary = OutboxMaintenanceSummary
  { requeued :: !Int,
    deadLettered :: !Int,
    backlog :: !Int
  }
  deriving stock (Generic, Eq, Show)

-- | Sensible defaults: batch of 32, ten retry attempts, two-second
-- constant backoff, per-key head-of-line ordering.
defaultPublishOptions :: OutboxPublishOptions
defaultPublishOptions =
  OutboxPublishOptions
    { batchSize = 32,
      maxAttempts = 10,
      backoff = ConstantBackoff 2,
      orderingPolicy = PerKeyHeadOfLine,
      publishingTimeout = 300,
      tracer = Nothing
    }

-- | Maintenance defaults match the publisher's retry ceiling and stale-row timeout.
defaultMaintenanceOptions :: OutboxMaintenanceOptions
defaultMaintenanceOptions =
  OutboxMaintenanceOptions
    { maxAttempts = defaultPublishOptions ^. #maxAttempts,
      publishingTimeout = defaultPublishOptions ^. #publishingTimeout
    }

-- | Validate outbox publisher options before starting a worker.
mkOutboxPublishOptions :: OutboxPublishOptions -> Either OutboxPublishConfigError OutboxPublishOptions
mkOutboxPublishOptions opts
  | opts ^. #batchSize < 1 = Left (InvalidOutboxBatchSize (opts ^. #batchSize))
  | opts ^. #maxAttempts < 1 = Left (InvalidOutboxMaxAttempts (opts ^. #maxAttempts))
  | opts ^. #publishingTimeout <= 0 = Left (InvalidOutboxPublishingTimeout (opts ^. #publishingTimeout))
  | otherwise = opts <$ validateBackoff (opts ^. #backoff)

validateBackoff :: BackoffSchedule -> Either OutboxPublishConfigError ()
validateBackoff = \case
  ConstantBackoff delay
    | delay < 0 -> Left (InvalidConstantBackoff delay)
    | otherwise -> Right ()
  ExponentialBackoff backoff
    | backoff ^. #initial <= 0 -> Left (InvalidExponentialBackoffInitial (backoff ^. #initial))
    | backoff ^. #multiplier < 1 -> Left (InvalidExponentialBackoffMultiplier (backoff ^. #multiplier))
    | backoff ^. #maxDelay < backoff ^. #initial ->
        Left (InvalidExponentialBackoffMaxDelay (backoff ^. #initial) (backoff ^. #maxDelay))
    | otherwise -> Right ()

-- | Wire representation of 'OutboxStatus' used in the @status@ column.
statusText :: OutboxStatus -> Text
statusText = \case
  OutboxPending -> "pending"
  OutboxPublishing -> "publishing"
  OutboxSent -> "sent"
  OutboxFailed -> "failed"
  OutboxDead -> "dead"

-- | Inverse of 'statusText'. Unknown database values are decode failures.
parseStatus :: Text -> Either Text OutboxStatus
parseStatus = \case
  "pending" -> Right OutboxPending
  "publishing" -> Right OutboxPublishing
  "sent" -> Right OutboxSent
  "failed" -> Right OutboxFailed
  "dead" -> Right OutboxDead
  bad -> Left ("unknown keiro_outbox.status: " <> bad)