Fsm.hs

{-# OPTIONS_GHC -Werror=incomplete-patterns #-}

-- | The explicit finite state machine driving a subscription worker.
--
-- A subscription worker is, at every instant, in exactly one named
-- 'SubscriptionState'. A single, exhaustively pattern-matched transition
-- function 'step' names every legal move: given the current state and an
-- 'Input' (a thing that just happened — a batch arrived, a fetch failed, the
-- queue overflowed, …) it returns the next state and a list of 'Effect's the
-- /driver/ ('Kiroku.Store.Subscription.Worker.runWorker') must perform.
--
-- Keeping the transition table in one pure function has two payoffs:
--
--   * It is the single, documented /extension seam/ for the sibling plan
--     @docs\/plans\/40-...@ (per-event retry \/ dead-letter). When that plan adds
--     a 'Kiroku.Store.Subscription.Types.SubscriptionResult' constructor it will
--     also add a matching 'Input' constructor; because 'step' is exhaustively
--     pattern-matched on 'Input' within each driving state and this module is
--     compiled with @-Werror=incomplete-patterns@, that addition fails to
--     compile until every relevant clause handles it, rather than falling
--     through silently.
--   * There is exactly one place to read the current state for observability
--     (the worker mirrors it into a @TVar@ — see M4).
--
-- This module is deliberately a near-leaf in the dependency graph: it depends
-- only on 'Kiroku.Store.Types' and @hasql-pool@, never on
-- 'Kiroku.Store.Subscription.Types' or 'Kiroku.Store.Observability'. That keeps
-- the graph acyclic once 'Kiroku.Store.Subscription.Types' grows a handle field
-- of type 'SubscriptionState'. The terminal-reason enumeration
-- 'SubscriptionStopReason' lives here for the same reason and is re-exported by
-- 'Kiroku.Store.Observability' so the public event API is unchanged.
module Kiroku.Store.Subscription.Fsm
  ( -- * State
    SubscriptionState (..),
    ResumeCondition (..),
    stateCursor,
    stateName,

    -- * Terminal reasons
    SubscriptionStopReason (..),

    -- * Per-event dispositions (retry / dead-letter)
    RetryDelay (..),
    retryDelayMicros,
    DeadLetterReason (..),
    deadLetterSummary,
    deadLetterReasonJson,

    -- * Driver alphabet
    Input (..),
    Effect (..),

    -- * Transition function
    step,
  )
where

import Control.Exception (SomeException)
import Data.Aeson (Value, object, (.=))
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime)
import Hasql.Pool qualified as Pool
import Kiroku.Store.Settings (DecodeFailure (..), DecodedBatch, decodedBatchLastEvent, decodedBatchLength)
import Kiroku.Store.Types (EventId (..), GlobalPosition (..), RecordedEvent (..))

-- | Why a subscription's worker thread stopped.
--
-- Defined here (rather than in 'Kiroku.Store.Observability') so the FSM's
-- 'Stopped' state can carry it without 'Kiroku.Store.Subscription.Fsm' having to
-- import the observability module — which would close a dependency cycle once
-- 'Kiroku.Store.Subscription.Types' references 'SubscriptionState'.
-- 'Kiroku.Store.Observability' re-exports this type, so the @KirokuEvent@ API is
-- unchanged.
data SubscriptionStopReason
  = -- | The handler returned 'Kiroku.Store.Subscription.Types.Stop' for some
    --       event. The normal completion path; the checkpoint is saved at that event.
    StopHandlerRequested
  | -- | The caller invoked @cancel@. No checkpoint advance is guaranteed;
    --       in-flight events replay on the next restart.
    StopCancelled
  | -- | The publisher marked the subscription overflowed under
    --       'Kiroku.Store.Subscription.Types.DropSubscription' and the worker
    --       surfaced 'Kiroku.Store.Subscription.Types.SubscriptionOverflowed'.
    StopOverflowed
  | -- | The worker thread died from an uncaught exception (typically a handler
    --       exception). The 'SomeException' carries the cause.
    StopWorkerCrashed !SomeException
  | -- | Default undecodable-event retries exhausted; checkpoint remains before it.
    StopUndecodable !DecodeFailure
  deriving stock (Show)

-- | How long to wait before redelivering a retried event.
--
-- The handler's @Kiroku.Store.Subscription.Types.Retry@ disposition carries one of
-- these; the worker sleeps for it before the next redelivery. Mirrors Shibuya's
-- @Shibuya.Core.Ack.RetryDelay@ shape (a 'NominalDiffTime') so the
-- @shibuya-kiroku-adapter@ can translate @AckRetry@ without loss, but is a
-- Kiroku-owned type so @kiroku-store@ does not depend on @shibuya-core@.
newtype RetryDelay = RetryDelay NominalDiffTime
  deriving stock (Eq, Show)

-- | The retry delay in microseconds, clamped at @0@, suitable for
-- 'Control.Concurrent.threadDelay'. Fractional seconds are rounded.
retryDelayMicros :: RetryDelay -> Int
retryDelayMicros (RetryDelay d) = max 0 (round (realToFrac d * 1_000_000 :: Double))

-- | Why an event was dead-lettered (recorded in @kiroku.dead_letters@ while the
-- subscription advances past it). Kiroku-owned so @kiroku-store@ stays independent
-- of @shibuya-core@; the adapter converts Shibuya's @DeadLetterReason@ into one of
-- these. 'deadLetterSummary' produces the @reason_summary@ text column and
-- 'deadLetterReasonJson' the structured @reason@ JSONB column.
data DeadLetterReason
  = -- | Permanently unprocessable; the 'Text' is operator-facing detail.
    DeadLetterPoison !Text
  | -- | The event payload failed validation/parsing; the 'Text' is the detail.
    DeadLetterInvalid !Text
  | -- | The bounded retry budget was exhausted after the given number of total deliveries.
    DeadLetterMaxAttempts !Int
  | -- | A custom reason: a summary plus structured JSON detail.
    DeadLetterOther !Text !Value
  | -- | A consumer explicitly chose to skip an undecodable event.
    DeadLetterDecodeFailure !DecodeFailure
  deriving stock (Eq, Show)

-- | The short operator-facing summary stored in @dead_letters.reason_summary@.
deadLetterSummary :: DeadLetterReason -> Text
deadLetterSummary = \case
  DeadLetterPoison detail -> "poison: " <> detail
  DeadLetterInvalid detail -> "invalid payload: " <> detail
  DeadLetterMaxAttempts n -> "max retry attempts exceeded (" <> T.pack (show n) <> ")"
  DeadLetterOther summary _ -> summary
  DeadLetterDecodeFailure failure -> "decode failure: " <> decodeFailureReason failure

-- | The structured JSON detail stored in @dead_letters.reason@ (JSONB).
deadLetterReasonJson :: DeadLetterReason -> Value
deadLetterReasonJson = \case
  DeadLetterPoison detail -> object ["kind" .= ("poison" :: Text), "detail" .= detail]
  DeadLetterInvalid detail -> object ["kind" .= ("invalid_payload" :: Text), "detail" .= detail]
  DeadLetterMaxAttempts n -> object ["kind" .= ("max_attempts_exceeded" :: Text), "attempts" .= n]
  DeadLetterOther summary detail -> object ["kind" .= ("other" :: Text), "summary" .= summary, "detail" .= detail]
  DeadLetterDecodeFailure failure -> object ["kind" .= ("decode_failure" :: Text), "event_id" .= (case decodeFailureEventId failure of EventId eid -> show eid), "detail" .= decodeFailureReason failure]

-- | What unblocks a 'Paused' worker.
--
-- In M1 there is one resume condition: the bounded queue has drained enough that
-- the publisher cleared the subscriber's @Paused@ flag. The constructor is named
-- rather than a bare unit so a future backpressure source (e.g. an explicit
-- resume signal) can be added without changing the 'SubscriptionState' shape.
data ResumeCondition
  = -- | Resume once the consumer drains its bounded queue.
    ResumeOnDrain
  deriving stock (Eq, Show)

-- | The named state a subscription worker is in.
--
--   * 'CatchingUp' — reading history directly from PostgreSQL in batches from
--     @cursor@ until it reaches the publisher's last-published position.
--     @attempt@ counts consecutive failed catch-up fetches and drives the
--     capped exponential backoff; it resets to 0 on any successful fetch.
--   * 'Live' — caught up; receiving new events (from the publisher's bounded
--     queue for AllStreams, or by re-querying for Category / consumer-group
--     members).
--   * 'Paused' — recoverable backpressure: a slow consumer filled its bounded
--     queue, so the worker stopped pulling and waits for 'resumeWhen' (added in
--     M2; in M1 this state is defined but never entered).
--   * 'Reconnecting' — recovering from a lost database connection while live;
--     backs off and re-enters 'CatchingUp' from @cursor@ (added in M3; in M1
--     defined but never entered).
--   * 'Retrying' — a handler returned
--     'Kiroku.Store.Subscription.Types.Retry' for the event at @cursor@; the
--     worker is redelivering it (@attempt@ counts redeliveries so far) and has not
--     advanced the checkpoint past it. Added by the sibling retry\/dead-letter plan
--     (@docs\/plans\/40-...@). It is a /surfaced observability state/: the worker's
--     delivery primitive writes it into the observable state @TVar@ while a
--     redelivery is pending and restores the prior driving state afterward, so it is
--     visible through @currentState@ but is never itself a driving state fed back to
--     'step' (the @Retrying@ clause in 'step' is therefore defensive).
--   * 'Stopped' — terminal, carrying the 'SubscriptionStopReason'.
data SubscriptionState
  = CatchingUp {cursor :: !GlobalPosition, attempt :: !Int}
  | Live {cursor :: !GlobalPosition}
  | Paused {cursor :: !GlobalPosition, resumeWhen :: !ResumeCondition}
  | Reconnecting {cursor :: !GlobalPosition, attempt :: !Int}
  | Retrying {cursor :: !GlobalPosition, attempt :: !Int}
  | Stopped {reason :: !SubscriptionStopReason}
  deriving stock (Show)

-- | The cursor of a non-terminal state, or 'GlobalPosition' @0@ for 'Stopped'.
-- Used by the driver to attach the current position to emitted events.
stateCursor :: SubscriptionState -> GlobalPosition
stateCursor = \case
  CatchingUp c _ -> c
  Live c -> c
  Paused c _ -> c
  Reconnecting c _ -> c
  Retrying c _ -> c
  Stopped _ -> GlobalPosition 0

-- | A stable, low-cardinality label for the state's /name/, independent of its
-- payload (cursor, attempt, reason). Suitable as a metric label value or an admin
-- column. The strings are fixed identifiers, not the derived 'Show' output, so
-- they will not drift if a constructor's fields change.
stateName :: SubscriptionState -> Text
stateName = \case
  CatchingUp {} -> "catching_up"
  Live {} -> "live"
  Paused {} -> "paused"
  Reconnecting {} -> "reconnecting"
  Retrying {} -> "retrying"
  Stopped {} -> "stopped"

-- | A thing that just happened, fed to 'step' to compute the next state.
--
-- The set is closed in M1; the sibling retry\/dead-letter plan extends it. Adding
-- a constructor here forces a compile error in every driving-state clause of
-- 'step' that does not yet handle it (this module is @-Werror=incomplete-patterns@).
data Input
  = -- | A non-empty history\/live batch arrived.
    BatchFetched !DecodedBatch
  | -- | A scanned namespace window; progress remains speculative until delivery.
    NamespaceWindowFetched !DecodedBatch !GlobalPosition !Bool
  | -- | Every applicable disposition in the window has completed.
    NamespaceWindowCompleted !GlobalPosition !Bool
  | -- | A fetch returned no rows (catch-up is complete).
    FetchEmpty
  | -- | A fetch hit a database error.
    FetchFailed !Pool.UsageError
  | -- | The cursor reached the publisher position; switch to live.
    CaughtUp
  | -- | The handler returned 'Stop' at this position.
    HandlerStopped !GlobalPosition
  | -- | The bounded queue filled under the fail-fast 'DropSubscription' policy
    --       (terminal).
    QueueOverflowed
  | -- | The bounded queue filled under the recoverable 'PauseAndResume' policy:
    --       pause rather than terminate.
    QueueBackpressured
  | -- | The worker drained the stale queue and is ready to recover (re-catch-up).
    QueueDrained
  | -- | The worker lost its database pool while live.
    ConnectionLost !GlobalPosition !Pool.UsageError
  | -- | The caller cancelled the worker.
    Cancelled
  deriving stock (Show)

-- | A side effect the driver must perform after a transition.
--
-- Some constructors (e.g. 'FetchHistory', 'RunLive') describe work the driver
-- already performs by inspecting the next state; they exist so the effect list is
-- a complete description of the transition for future drivers and for readers.
data Effect
  = -- | Read a catch-up batch from the given cursor.
    FetchHistory !GlobalPosition
  | -- | Obtain the next live batch via the active live strategy.
    RunLive
  | -- | Call the handler per event, checkpointing at the batch tail.
    DeliverBatch !DecodedBatch
  | -- | Deliver before publishing the window frontier as completed state.
    DeliverNamespaceWindow !DecodedBatch !GlobalPosition !Bool
  | -- | Persist the checkpoint at this position. (Named 'Checkpoint' rather
    --       than @SaveCheckpoint@ to avoid clashing with the
    --       'Kiroku.Store.Observability.SubscriptionDbPhase' constructor of that name.)
    Checkpoint !GlobalPosition
  | -- | Block until the consumer drains the queue.
    WaitForDrain
  | -- | Sleep for the backoff schedule of this attempt number.
    Backoff !Int
  | -- | Emit @KirokuEventSubscriptionCaughtUp@.
    EmitCaughtUp
  | -- | Emit @KirokuEventSubscriptionPaused@ (M4).
    EmitPaused
  | -- | Emit @KirokuEventSubscriptionResumed@ (M4).
    EmitResumed
  | -- | Emit @KirokuEventSubscriptionReconnecting@ with this attempt (M4).
    EmitReconnecting !Int
  | -- | Terminal: stop the driver with this reason (return or rethrow).
    Halt !SubscriptionStopReason
  deriving stock (Show)

-- The last event's position in a batch, falling back to the given cursor when
-- the batch is empty (the driver never feeds 'step' an empty 'BatchFetched',
-- so the fallback is defensive).
lastPos :: GlobalPosition -> DecodedBatch -> GlobalPosition
lastPos fallback evs
  | decodedBatchLength evs == 0 = fallback
  | otherwise = globalPosition (decodedBatchLastEvent evs)

-- | The single transition function. Given the current state and an input,
-- return the next state and the effects to perform.
--
-- Exhaustively pattern-matched on 'Input' within the two /driving/ states
-- ('CatchingUp', 'Live') so that extending 'Input' (the retry\/dead-letter
-- sibling plan) forces a compile error here. The recoverable states 'Paused' and
-- 'Reconnecting' carry their M2\/M3 transitions plus defensive self-loops; the
-- terminal 'Stopped' state ignores further input.
step :: SubscriptionState -> Input -> (SubscriptionState, [Effect])
step st input = case st of
  CatchingUp c n -> case input of
    NamespaceWindowFetched evs frontier done -> (CatchingUp c n, [DeliverNamespaceWindow evs frontier done])
    NamespaceWindowCompleted frontier done -> if done then (Live frontier, [EmitCaughtUp]) else (CatchingUp frontier 0, [])
    BatchFetched evs -> (CatchingUp (lastPos c evs) 0, [DeliverBatch evs])
    FetchEmpty -> (Live c, [EmitCaughtUp])
    FetchFailed _ -> (CatchingUp c (n + 1), [Backoff n])
    CaughtUp -> (Live c, [EmitCaughtUp])
    HandlerStopped _ -> (Stopped StopHandlerRequested, [Halt StopHandlerRequested])
    QueueOverflowed -> (Stopped StopOverflowed, [Halt StopOverflowed])
    QueueBackpressured -> (CatchingUp c n, []) -- defensive: catch-up reads the DB, not the queue
    QueueDrained -> (CatchingUp c n, [])
    ConnectionLost observed _ -> (Reconnecting (max c observed) 1, [EmitReconnecting 1, Backoff 1])
    Cancelled -> (Stopped StopCancelled, [Halt StopCancelled])
  Live c -> case input of
    NamespaceWindowFetched evs frontier done -> (Live c, [DeliverNamespaceWindow evs frontier done])
    NamespaceWindowCompleted frontier _ -> (Live frontier, [])
    BatchFetched evs -> (Live (lastPos c evs), [DeliverBatch evs])
    FetchEmpty -> (Live c, [RunLive])
    FetchFailed _ -> (Reconnecting c 1, [EmitReconnecting 1, Backoff 1])
    CaughtUp -> (Live c, [])
    HandlerStopped _ -> (Stopped StopHandlerRequested, [Halt StopHandlerRequested])
    QueueOverflowed -> (Stopped StopOverflowed, [Halt StopOverflowed])
    QueueBackpressured -> (Paused c ResumeOnDrain, [EmitPaused])
    QueueDrained -> (Live c, [RunLive])
    ConnectionLost observed _ -> (Reconnecting (max c observed) 1, [EmitReconnecting 1, Backoff 1])
    Cancelled -> (Stopped StopCancelled, [Halt StopCancelled])
  Paused c rc -> case input of
    -- The worker drained the stale queue and cleared the pause flag; recover
    -- by re-catching-up from the checkpoint so any events the publisher
    -- skipped while full are re-read from the database. No loss; the live
    -- queue's stale filter drops the now-superseded queued entries on return.
    QueueDrained -> (CatchingUp c 0, [EmitResumed])
    Cancelled -> (Stopped StopCancelled, [Halt StopCancelled])
    HandlerStopped _ -> (Stopped StopHandlerRequested, [Halt StopHandlerRequested])
    _ -> (Paused c rc, [])
  Reconnecting c n -> case input of
    NamespaceWindowFetched evs frontier done -> (CatchingUp c 0, [DeliverNamespaceWindow evs frontier done])
    NamespaceWindowCompleted frontier done -> if done then (Live frontier, [EmitCaughtUp]) else (CatchingUp frontier 0, [])
    BatchFetched evs -> (CatchingUp (lastPos c evs) 0, [DeliverBatch evs])
    FetchEmpty -> (Live c, [])
    FetchFailed _ -> (Reconnecting c (n + 1), [EmitReconnecting (n + 1), Backoff (n + 1)])
    ConnectionLost observed _ -> (Reconnecting (max c observed) (n + 1), [EmitReconnecting (n + 1), Backoff (n + 1)])
    CaughtUp -> (CatchingUp c 0, [])
    HandlerStopped _ -> (Stopped StopHandlerRequested, [Halt StopHandlerRequested])
    Cancelled -> (Stopped StopCancelled, [Halt StopCancelled])
    QueueOverflowed -> (Reconnecting c n, [])
    QueueBackpressured -> (Reconnecting c n, [])
    QueueDrained -> (Reconnecting c n, [])
  -- 'Retrying' is a surfaced observability state managed inside the worker's
  -- delivery primitive (it sets the observable @TVar@ to 'Retrying' while a
  -- redelivery is pending and restores the prior driving state afterward); it
  -- is never a driving state fed back to 'step'. This clause is therefore
  -- defensive — only cancellation / a handler stop could plausibly race in, and
  -- both terminate; anything else returns to 'Live' at the same cursor.
  Retrying c _ -> case input of
    Cancelled -> (Stopped StopCancelled, [Halt StopCancelled])
    HandlerStopped _ -> (Stopped StopHandlerRequested, [Halt StopHandlerRequested])
    _ -> (Live c, [])
  Stopped r -> (Stopped r, [])