Snapshot.hs

-- | Workflow journal snapshots: the workflow-specific 'StateCodec' and the
-- read\/write helpers the runtime uses to skip a full version-0 replay.
--
-- A durable workflow folds its journal stream into a 'WorkflowState' — a
-- @'Map' 'Text' 'Value'@ of step-name to encoded result. For a long-lived
-- workflow, re-reading the whole journal on every run and resume is ruinous.
-- This module supplies the EP-4 snapshot machinery the runtime reuses to
-- persist that folded map and, on the next run, seed from it and replay only
-- the tail.
--
-- == Why a workflow-specific codec
--
-- EP-4's 'Keiro.Snapshot.Codec.defaultStateCodec' derives its 'shapeHash' from
-- a /statically-known/ keiki register-file slot list
-- ('Keiki.Shape.regFileShapeHash'). A workflow's step names are dynamic runtime
-- strings, so there is no static slot list. The accumulated state is instead a
-- self-describing JSON object, so it round-trips through Aeson trivially and its
-- /shape/ never varies with which steps ran. Hence the fixed sentinel
-- 'workflowStateShapeHash'; per-step result-type evolution is each step's own
-- 'Aeson.ToJSON'\/'Aeson.FromJSON' concern, surfaced at the @step@ decode in
-- "Keiro.Workflow", not here.
--
-- == Advisory semantics
--
-- A snapshot is an optimization, never a source of truth. 'loadWorkflowSnapshot'
-- returns 'Nothing' — meaning "replay from version 0" — when the stream has no
-- id yet, no matching snapshot, or an undecodable snapshot. So a stale or corrupt
-- snapshot degrades performance at worst, never correctness.
--
-- Runtime snapshot writes are advisory too. After a workflow journal append has
-- committed, "Keiro.Workflow" swallows any snapshot-store failure and increments
-- @keiro.snapshot.write.failures@; the committed step, completion, or rotation
-- therefore still determines a successful run. This low-level module continues
-- to expose the store primitive, while the runtime owns that error handling.
--
-- Snapshot writes store the full accumulated step map each time the selected
-- policy fires. With an every-n policy that means rewriting the complete map at
-- each boundary; the intended way to bound that cost for forever-running
-- workflows is 'Keiro.Workflow.continueAsNew', which starts a fresh generation
-- with a small carried seed.
module Keiro.Workflow.Snapshot
  ( workflowStateCodec,
    workflowStateCodecVersion,
    workflowStateShapeHash,
    lookupWorkflowSnapshot,
    loadWorkflowSnapshot,
    writeWorkflowSnapshot,
  )
where

import Data.Aeson (Result (..))
import Data.Text qualified as Text
import Effectful (Eff, (:>))
import Keiro.EventStream (StateCodec (..))
import Keiro.Prelude
import Keiro.Snapshot (SnapshotMissReason (..), writeSnapshot)
import Keiro.Snapshot.Schema (lookupSnapshot)
import Keiro.Workflow.Types (WorkflowState)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Read (lookupStreamId)
import Kiroku.Store.Types (StreamId, StreamName, StreamVersion)

-- | The codec version of the workflow snapshot envelope. Bumped only if the
-- map /envelope/ encoding itself changes (e.g. a switch from a bare JSON
-- object to a tagged container), never for a per-step result-type change.
workflowStateCodecVersion :: Int
workflowStateCodecVersion = 1

-- | A FIXED sentinel shape hash. The accumulated state is always "a JSON
-- object of step-name strings to self-describing JSON values"; its /shape/
-- never varies with which steps ran, so the shape hash is constant. Per-step
-- result-type evolution is the step's own 'Aeson.ToJSON'\/'Aeson.FromJSON'
-- concern, surfaced at the step decode in "Keiro.Workflow", not here.
workflowStateShapeHash :: Text
workflowStateShapeHash = "keiro.workflow.stepmap.v1"

-- | The workflow-specific 'StateCodec'. Encodes the accumulated
-- 'WorkflowState' map straight to JSON and decodes it back, with a fixed
-- discriminant ('workflowStateCodecVersion', 'workflowStateShapeHash').
workflowStateCodec :: StateCodec WorkflowState
workflowStateCodec =
  StateCodec
    { stateCodecVersion = workflowStateCodecVersion,
      shapeHash = workflowStateShapeHash,
      stateShapeHash = workflowStateShapeHash,
      encode = toJSON,
      decode = \value -> case fromJSON value of
        Success m -> Right m
        Error msg -> Left (Text.pack msg)
    }

-- | Encode the accumulated map and upsert @keiro_snapshots@ for the journal
-- stream. Called from the @step@ miss-path (and the completion site) with the
-- 'Kiroku.Store.Types.AppendResult'\'s @streamId@ and post-append @streamVersion@.
-- The underlying upsert keeps only the highest version per stream, so a re-fire
-- at an already-snapshotted version is a harmless no-op.
writeWorkflowSnapshot :: (Store :> es) => StreamId -> StreamVersion -> WorkflowState -> Eff es ()
writeWorkflowSnapshot streamId version state =
  writeSnapshot streamId version workflowStateCodec state

-- | Resolve the journal stream id, look up the latest matching snapshot row,
-- and decode it to a @('WorkflowState', 'StreamVersion')@ seed. Returns
-- 'Nothing' — meaning "replay from 0" — when the stream has no id yet, no
-- matching snapshot, or an undecodable snapshot (advisory semantics). Mirrors
-- 'Keiro.Snapshot.hydrateWithSnapshot''s miss-is-benign contract.
loadWorkflowSnapshot :: (Store :> es) => StreamName -> Eff es (Maybe (WorkflowState, StreamVersion))
loadWorkflowSnapshot journalName =
  lookupWorkflowSnapshot journalName <&> either (const Nothing) Just

-- | Resolve and decode a workflow snapshot while retaining the reason a
-- usable seed was unavailable. This is the observable counterpart to
-- 'loadWorkflowSnapshot'.
lookupWorkflowSnapshot :: (Store :> es) => StreamName -> Eff es (Either SnapshotMissReason (WorkflowState, StreamVersion))
lookupWorkflowSnapshot journalName = do
  mStreamId <- lookupStreamId journalName
  case mStreamId of
    Nothing -> pure (Left SnapshotNoStream)
    Just streamId -> do
      mRow <-
        lookupSnapshot
          streamId
          workflowStateCodecVersion
          workflowStateShapeHash
          workflowStateShapeHash
      pure $ case mRow of
        Nothing -> Left SnapshotNotFound
        Just row ->
          case (workflowStateCodec ^. #decode) (row ^. #state) of
            Left message -> Left (SnapshotDecodeFailed message)
            Right state -> Right (state, row ^. #streamVersion)