Rebuild.hs

{- | The supported offline read-model rebuild lifecycle.

Use this checklist rather than composing the low-level status transitions:

1. Call 'Keiro.ReadModel.Schema.registerReadModel' once at projection startup.
   Explicit registration makes misspelled or never-populated models fail with
   'ReadModelUnregistered' instead of appearing healthy.
2. Call 'startRebuild' with every feeding async projection name and the replay
   position. It atomically fences live writers, takes queries offline, truncates
   the data table, clears the named dedup keys, and resets the subscription
   checkpoint, preventing live/replay interleaving and all-deduplicated rebuilds.
3. Replay through 'Keiro.Projection.applyAsyncProjectionUnfenced'. This is the
   only apply path allowed to bypass the live-writer fence, while retaining
   deduplication inside the designated rebuild.
4. After replay catches up and application-specific verification succeeds, call
   'finishRebuild' with the same projection names and replay position. Its
   promotion guard refuses to serve a non-empty-log rebuild that applied no
   events.
5. If replay or verification fails, call 'abandonRebuild'. Queries remain
   unavailable instead of exposing partial data; repair or restore the table
   before beginning another rebuild.

Normal workers continue to call 'Keiro.Projection.applyAsyncProjection'. Its
registry lock fences them automatically while the model is rebuilding, but they
must not checkpoint an 'Keiro.Projection.AsyncFenced' event. Keiro does not yet
provide a shadow-table or online cutover mechanism; applications that need
zero-downtime rebuilds must build that orchestration above this lifecycle API.
-}
module Keiro.ReadModel.Rebuild (
    RebuildError (..),
    startRebuild,
    finishRebuild,
    rebuild,
    promote,
    abandonRebuild,
)
where

import Contravariant.Extras (contrazip2)
import Data.Text.Encoding qualified as TE
import Effectful (Eff, (:>))
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Statement (Statement, preparable)
import Keiro.Prelude
import Keiro.ReadModel
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import Kiroku.Store.Types (GlobalPosition (..))
import "hasql-transaction" Hasql.Transaction qualified as Tx

-- | Why a rebuild could not be promoted.
data RebuildError
    = {- | The model name and current store head when a rebuild started before
      existing events but applied none of its feeding projections.
      -}
      RebuildProducedNoApplies !Text !GlobalPosition
    deriving stock (Generic, Eq, Show)

{- | Atomically take a model offline, truncate its data table, clear the dedup
keys for its feeding async projections, and reset every member of its
subscription to the supplied replay position.

The registry transition runs first and holds the row lock that
'Keiro.Projection.applyAsyncProjection' uses as its writer fence. PostgreSQL
keeps the table truncate transactional. Kiroku's public checkpoint save is
monotonic, so this helper deliberately resets @subscriptions.last_seen@
directly inside the same fenced transaction.
-}
startRebuild ::
    (Store :> es) =>
    ReadModel q r ->
    [Text] ->
    GlobalPosition ->
    Eff es ReadModelMetadata
startRebuild readModel projectionNames replayFrom =
    runTransaction $ do
        metadata <- transitionReadModelTxFor readModel Rebuilding
        Tx.sql (TE.encodeUtf8 ("TRUNCATE TABLE " <> qualifiedTableName readModel))
        unless (null projectionNames) $
            Tx.statement projectionNames deleteProjectionDedupStmt
        Tx.statement
            (readModel ^. #subscriptionName, globalPositionToInt replayFrom)
            resetSubscriptionCheckpointStmt
        pure metadata

{- | Promote a completed rebuild in the same transaction as its safety check.

When async projection names are supplied, a store head beyond @replayFrom@
requires at least one new dedup row. 'startRebuild' deleted those rows, so their
count is the model-independent number of projection applications during this
rebuild. An empty projection-name list denotes an inline-only model and skips
the guard because no async dedup rows can exist for it.
-}
finishRebuild ::
    (Store :> es) =>
    ReadModel q r ->
    [Text] ->
    GlobalPosition ->
    Eff es (Either RebuildError ReadModelMetadata)
finishRebuild readModel projectionNames replayFrom =
    runTransaction $ do
        headPosition <- Tx.statement () storeHeadPositionStmt
        applyCount <-
            if null projectionNames
                then pure 0
                else Tx.statement projectionNames countProjectionDedupStmt
        if null projectionNames || applyCount > 0 || headPosition <= replayFrom
            then Right <$> transitionReadModelTxFor readModel Live
            else pure (Left (RebuildProducedNoApplies (readModel ^. #name) headPosition))

{- | Low-level status transition only. It does not truncate data, reset dedup
keys or checkpoints, or establish the complete rebuild workflow. Use
'startRebuild' for supported rebuilds.
-}
rebuild :: (Store :> es) => ReadModel q r -> Eff es ReadModelMetadata
rebuild readModel =
    markRebuilding
        (readModel ^. #name)
        (readModel ^. #version)
        (readModel ^. #shapeHash)

{- | Low-level status transition only. It bypasses the empty-rebuild guard. Use
'finishRebuild' to promote a supported rebuild.
-}
promote :: (Store :> es) => ReadModel q r -> Eff es ReadModelMetadata
promote readModel =
    markLive
        (readModel ^. #name)
        (readModel ^. #version)
        (readModel ^. #shapeHash)

-- | Mark a model 'Abandoned', backing out of an in-progress rebuild.
abandonRebuild :: (Store :> es) => ReadModel q r -> Eff es ReadModelMetadata
abandonRebuild readModel =
    markAbandoned
        (readModel ^. #name)
        (readModel ^. #version)
        (readModel ^. #shapeHash)

transitionReadModelTxFor :: ReadModel q r -> ReadModelStatus -> Tx.Transaction ReadModelMetadata
transitionReadModelTxFor readModel status =
    transitionReadModelTx
        (readModel ^. #name)
        (readModel ^. #version)
        (readModel ^. #shapeHash)
        status

deleteProjectionDedupStmt :: Statement [Text] ()
deleteProjectionDedupStmt =
    preparable
        """
        DELETE FROM keiro.keiro_projection_dedup
        WHERE projection_name = ANY($1)
        """
        (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.text))))
        D.noResult

countProjectionDedupStmt :: Statement [Text] Int64
countProjectionDedupStmt =
    preparable
        """
        SELECT count(*)
        FROM keiro.keiro_projection_dedup
        WHERE projection_name = ANY($1)
        """
        (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.text))))
        (D.singleRow (D.column (D.nonNullable D.int8)))

resetSubscriptionCheckpointStmt :: Statement (Text, Int64) ()
resetSubscriptionCheckpointStmt =
    preparable
        """
        UPDATE subscriptions
        SET last_seen = $2, updated_at = now()
        WHERE subscription_name = $1
        """
        ( contrazip2
            (E.param (E.nonNullable E.text))
            (E.param (E.nonNullable E.int8))
        )
        D.noResult

storeHeadPositionStmt :: Statement () GlobalPosition
storeHeadPositionStmt =
    preparable
        """
        SELECT COALESCE(max(stream_version), 0)
        FROM stream_events
        WHERE stream_id = 0
        """
        E.noParams
        (D.singleRow (GlobalPosition <$> D.column (D.nonNullable D.int8)))

globalPositionToInt :: GlobalPosition -> Int64
globalPositionToInt (GlobalPosition position) = position