ReadModel.hs

{- | Querying the read side, with explicit consistency.

A 'ReadModel' is a named, versioned SQL projection table plus the query that
reads it. Querying it does more than run SQL: 'runQuery' first verifies the
table's registered schema is current and 'Live' (rejecting a stale or
mid-rebuild model), then honours the requested 'ConsistencyMode' before
running the query in a transaction.

The consistency modes trade freshness against latency:

* 'Strong' — capture the store head position at query start and block until
  the model's subscription cursor reaches it.
* 'Eventual' — query immediately. Read-your-writes is the projection worker's
  responsibility under 'Eventual'.
* 'PositionWait' — block until the model's subscription has caught up to a
  target 'GlobalPosition' (typically the position returned by the command
  the caller just ran), giving read-your-writes against an asynchronous
  projection. 'waitFor' implements the polling loop and times out with
  'ReadModelWaitTimeout'.

Schema lifecycle (registration, status transitions) lives in
"Keiro.ReadModel.Schema", which is re-exported here.

Register each model once at projection startup with 'registerReadModel' before
serving queries. Queries fail with 'ReadModelUnregistered' when startup wiring
has not registered the model; they never create registry rows themselves.
-}
module Keiro.ReadModel (
    -- * Definition
    ReadModel (..),
    qualifiedTableName,

    -- * Consistency
    ConsistencyMode (..),
    StrongScope (..),
    PositionWaitOptions (..),
    defaultStrongWaitOptions,

    -- * Querying
    runQuery,
    runQueryWith,
    waitFor,
    readSubscriptionPosition,
    storeHeadPosition,
    categoryHeadPosition,

    -- * Errors
    ReadModelError (..),

    -- * Schema lifecycle
    module Keiro.ReadModel.Schema,
)
where

import Control.Concurrent (threadDelay)
import Data.Time.Clock (diffUTCTime)
import Data.Vector qualified as Vector
import Effectful (Eff, IOE, (:>))
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Statement (Statement, preparable)
import Keiro.Connection (qualifyTable)
import Keiro.Prelude
import Keiro.ReadModel.Schema
import Keiro.Telemetry (KeiroMetrics, recordProjectionWaitTimeouts)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Read (readAllBackward)
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (GlobalPosition (..))
import "hasql-transaction" Hasql.Transaction qualified as Tx
import Prelude qualified

{- | A queryable read-side projection over a query input @q@ and result @r@.

* 'name' — logical identity, also the key in the @keiro_read_models@
  registry.
* 'tableName' — the underlying projection table.
* 'schema' — the PostgreSQL schema the read-model /data/ table lives in. The
  application qualifies its 'query' SQL against this schema (typically via
  'Keiro.Connection.qualifyTable' or 'qualifiedTableName'); Keiro does not
  rewrite 'query'. This is the application's data schema and is entirely
  separate from Keiro's own @keiro@ schema, where the @keiro_read_models@
  registry lives. It is deliberately not persisted (see 'ensureReadModel').
* 'subscriptionName' — the cursor that tracks how far the projection worker
  has consumed the event log; consulted by 'PositionWait'.
* 'version' \/ 'shapeHash' — schema identity; a query fails with
  'ReadModelStaleSchema' if the registered values diverge, forcing a rebuild.
* 'defaultConsistency' — the 'ConsistencyMode' used by 'runQuery'.
* 'strongScope' — the event-log head a 'Strong' query waits for.
* 'query' — the SQL read, as a 'Hasql.Transaction.Transaction'.
-}
data ReadModel q r = ReadModel
    { name :: !Text
    , tableName :: !Text
    , schema :: !Text
    , subscriptionName :: !Text
    , version :: !Int
    , shapeHash :: !Text
    , defaultConsistency :: !ConsistencyMode
    , strongScope :: !StrongScope
    , query :: !(q -> Tx.Transaction r)
    }
    deriving stock (Generic)

{- | The read model's fully-qualified, double-quoted table reference
@"schema"."table"@, for interpolation into the application's projection SQL.
Equal to @'Keiro.Connection.qualifyTable' ('schema' rm) ('tableName' rm)@.
-}
qualifiedTableName :: ReadModel q r -> Text
qualifiedTableName readModel =
    qualifyTable (readModel ^. #schema) (readModel ^. #tableName)

{- | How fresh a read must be before the query runs.

'Strong' waits for the model's subscription to reach the store head captured
at query start according to the model's 'strongScope'. It is intended for
asynchronous read models with a worker advancing that subscription cursor;
inline-only models should use 'Eventual' because they have no subscription
worker to advance while waiting.
'PositionWait' blocks until the projection has caught up to a caller-supplied
target log position (or times out). 'Eventual' queries immediately.
-}
data ConsistencyMode
    = Strong
    | Eventual
    | PositionWait !PositionWaitOptions
    deriving stock (Generic, Eq, Show)

{- | Which log head a 'Strong' read must reach.

'EntireLog' preserves the original behavior and is live only when the model's
subscription observes every event. A category subscription should use
'CategoryHead' with its Kiroku category, so unrelated categories cannot hold
the read behind forever. A model fed by multiple categories should use
'PositionWait' for an explicit write position or 'EntireLog' with a matching
all-stream subscription.

Kiroku currently does not advance category checkpoints on empty fetches. If it
does so in a future release, category-scoped targets may become unnecessary,
but the explicit model contract remains valid.
-}
data StrongScope
    = EntireLog
    | CategoryHead !Text
    deriving stock (Generic, Eq, Show)

{- | Parameters for a 'PositionWait' query.

* 'target' — the 'GlobalPosition' the projection must reach; 'Nothing'
  skips waiting entirely.
* 'timeoutMicros' — give up after this long with 'ReadModelWaitTimeout'.
* 'pollMicros' — delay between subscription-position checks.
-}
data PositionWaitOptions = PositionWaitOptions
    { target :: !(Maybe GlobalPosition)
    , timeoutMicros :: !Int
    , pollMicros :: !Int
    }
    deriving stock (Generic, Eq, Show)

{- | Default wait settings used by 'Strong': wait up to five seconds, polling
every 10ms, for the store head captured at query start.
-}
defaultStrongWaitOptions :: PositionWaitOptions
defaultStrongWaitOptions =
    PositionWaitOptions
        { target = Nothing
        , timeoutMicros = 5000000
        , pollMicros = 10000
        }

-- | Why a read-model query could not run.
data ReadModelError
    = {- | No registry row exists for the model. Register it once at projection
      startup with 'registerReadModel' before serving queries.
      -}
      ReadModelUnregistered !Text
    | {- | The registered schema (version or shape hash) differs from the
      model's current definition: name, expected vs. found version, then
      expected vs. found shape hash. The model must be rebuilt.
      -}
      ReadModelStaleSchema !Text !Int !Int !Text !Text
    | {- | A 'PositionWait' query timed out: model name, target position, and
      the last observed subscription position.
      -}
      ReadModelWaitTimeout !Text !GlobalPosition !GlobalPosition
    | {- | The model is registered but not 'Live' (e.g. rebuilding or
      abandoned): name and current status.
      -}
      ReadModelNotLive !Text !ReadModelStatus
    deriving stock (Generic, Eq, Show)

{- | Query a read model using its 'defaultConsistency'. Validates schema and
liveness first, waits if the mode requires it, then runs the query.
-}
runQuery ::
    (IOE :> es, Store :> es) =>
    Maybe KeiroMetrics ->
    ReadModel q r ->
    q ->
    Eff es (Either ReadModelError r)
runQuery metrics readModel =
    runQueryWith metrics (readModel ^. #defaultConsistency) readModel

{- | Query a read model with an explicit 'ConsistencyMode', overriding its
default. Validates the model's schema and liveness, honours the wait mode,
then runs the query in a transaction.
-}
runQueryWith ::
    (IOE :> es, Store :> es) =>
    Maybe KeiroMetrics ->
    ConsistencyMode ->
    ReadModel q r ->
    q ->
    Eff es (Either ReadModelError r)
runQueryWith metrics consistency readModel input = do
    schemaCheck <- ensureReadModel readModel
    case schemaCheck of
        Left err -> pure (Left err)
        Right () -> do
            waitResult <- waitIfNeeded metrics consistency readModel
            case waitResult of
                Left err -> pure (Left err)
                Right () -> Right <$> runTransaction ((readModel ^. #query) input)

{- | Block until the model's subscription has advanced to @targetPosition@,
polling at 'pollMicros' intervals. Returns @Right ()@ once caught up, or
'ReadModelWaitTimeout' if 'timeoutMicros' elapses first.
-}
waitFor ::
    (IOE :> es, Store :> es) =>
    Maybe KeiroMetrics ->
    PositionWaitOptions ->
    ReadModel q r ->
    GlobalPosition ->
    Eff es (Either ReadModelError ())
waitFor metrics options readModel targetPosition = do
    started <- liftIO getCurrentTime
    poll started (GlobalPosition 0)
  where
    poll started observed = do
        current <- readSubscriptionPosition (readModel ^. #subscriptionName)
        let observed' = fromMaybe observed current
        if observed' >= targetPosition
            then pure (Right ())
            else do
                now <- liftIO getCurrentTime
                let elapsedMicros =
                        Prelude.floor
                            (diffUTCTime now started Prelude.* 1000000)
                if elapsedMicros >= options ^. #timeoutMicros
                    then do
                        -- A genuine give-up: bump keiro.projection.wait.timeouts (no-op
                        -- under a 'Nothing' handle) before surfacing the timeout.
                        recordProjectionWaitTimeouts metrics 1
                        pure
                            (Left (ReadModelWaitTimeout (readModel ^. #name) targetPosition observed'))
                    else do
                        liftIO (threadDelay (options ^. #pollMicros))
                        poll started observed'

-- Note: the read model's 'schema' field is deliberately NOT persisted here. The
-- registry keys on name/version/shapeHash/status (the model's schema identity);
-- where the application's data table physically lives is a deployment/wiring
-- concern, not part of that identity. See EP-4's Decision Log.
ensureReadModel ::
    (Store :> es) =>
    ReadModel q r ->
    Eff es (Either ReadModelError ())
ensureReadModel readModel = do
    found <- lookupReadModel (readModel ^. #name)
    pure $ case found of
        Just metadata -> validateMetadata readModel metadata
        Nothing -> Left (ReadModelUnregistered (readModel ^. #name))

validateMetadata :: ReadModel q r -> ReadModelMetadata -> Either ReadModelError ()
validateMetadata readModel metadata
    | metadata ^. #version /= readModel ^. #version =
        stale
    | metadata ^. #shapeHash /= readModel ^. #shapeHash =
        stale
    | metadata ^. #status /= Live =
        Left (ReadModelNotLive (readModel ^. #name) (metadata ^. #status))
    | otherwise =
        Right ()
  where
    stale =
        Left
            ( ReadModelStaleSchema
                (readModel ^. #name)
                (readModel ^. #version)
                (metadata ^. #version)
                (readModel ^. #shapeHash)
                (metadata ^. #shapeHash)
            )

waitIfNeeded ::
    (IOE :> es, Store :> es) =>
    Maybe KeiroMetrics ->
    ConsistencyMode ->
    ReadModel q r ->
    Eff es (Either ReadModelError ())
waitIfNeeded metrics Strong readModel = do
    target <- case readModel ^. #strongScope of
        EntireLog -> storeHeadPosition
        CategoryHead category -> categoryHeadPosition category
    waitFor metrics (defaultStrongWaitOptions & #target ?~ target) readModel target
waitIfNeeded _ Eventual _ = pure (Right ())
waitIfNeeded metrics (PositionWait options) readModel =
    case options ^. #target of
        Nothing -> pure (Right ())
        Just targetPosition -> waitFor metrics options readModel targetPosition

readSubscriptionPosition ::
    (Store :> es) =>
    Text ->
    Eff es (Maybe GlobalPosition)
readSubscriptionPosition subscriptionName =
    runTransaction
        $ Tx.statement subscriptionName lookupSubscriptionPositionStmt

lookupSubscriptionPositionStmt :: Statement Text (Maybe GlobalPosition)
lookupSubscriptionPositionStmt =
    preparable
        """
        SELECT min(last_seen)
        FROM subscriptions
        WHERE subscription_name = $1
        """
        (E.param (E.nonNullable E.text))
        (D.singleRow (fmap GlobalPosition <$> D.column (D.nullable D.int8)))

{- | The global position of the most recent event in the @$all@ log, or
@GlobalPosition 0@ when the log is empty. 'readAllBackward' treats
@GlobalPosition 0@ as "after everything", so a limit of 1 returns the head.
-}
storeHeadPosition :: (Store :> es) => Eff es GlobalPosition
storeHeadPosition = do
    recent <- readAllBackward (GlobalPosition 0) 1
    pure $ case Vector.toList recent of
        (event : _) -> event ^. #globalPosition
        [] -> GlobalPosition 0

{- | The latest global position originating in a Kiroku category, or
@GlobalPosition 0@ when that category has no events. This deliberately reads
Kiroku's indexed @streams@ and @$all@ membership tables because Kiroku 0.3 does
not export a category-head query.
-}
categoryHeadPosition :: (Store :> es) => Text -> Eff es GlobalPosition
categoryHeadPosition category =
    runTransaction
        $ Tx.statement category categoryHeadPositionStmt

categoryHeadPositionStmt :: Statement Text GlobalPosition
categoryHeadPositionStmt =
    preparable
        """
        SELECT COALESCE(max(se.stream_version), 0)
        FROM streams s
        JOIN stream_events se
          ON se.original_stream_id = s.stream_id
         AND se.stream_id = 0
        WHERE s.category = $1
        """
        (E.param (E.nonNullable E.text))
        (D.singleRow (GlobalPosition <$> D.column (D.nonNullable D.int8)))