Instance.hs

{-# LANGUAGE OverloadedRecordDot #-}

-- | Durable workflow instance summaries.
--
-- The journal stream and @keiro_workflow_steps@ index remain the source of truth
-- for replay. This module maintains one @keiro_workflows@ row per logical
-- workflow instance so the resume worker can track lifecycle, attempts, and
-- leases without scanning journal history.
module Keiro.Workflow.Instance
  ( WorkflowStatus (..),
    WorkflowInstanceRow (..),
    WorkflowInstanceFilter (..),
    defaultWorkflowInstanceFilter,
    ResurrectOutcome (..),
    CancelWorkflowOutcome (..),
    ClaimOutcome (..),
    statusToText,
    statusFromText,
    upsertInstanceTx,
    markInstanceSuspendedAwaiting,
    lookupInstance,
    lookupInstanceSession,
    WorkflowHeaderKey,
    workflowHeaderKeyBytes,
    WorkflowHeaderObservation (..),
    lookupInstanceConditionalSession,
    listWorkflowInstances,
    inspectionInstanceRows,
    cancelWorkflow,
    claimInstance,
    renewInstanceLeaseTx,
    renewInstanceLease,
    releaseInstance,
    forceReleaseInstanceLease,
    recordCrashTx,
    resetInstanceAttempts,
    reviveFailedInstanceTx,
    resurrectFailedWorkflow,
  )
where

import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5)
import Data.Aeson ((.:))
import Data.Aeson qualified as Aeson
import Data.Aeson.Types (parseEither)
import Data.ByteString qualified as ByteString
import Data.ByteString.Lazy qualified as LazyByteString
import Data.Functor.Compose (Compose (..))
import Data.Functor.Contravariant ((>$<))
import Data.Int (Int32)
import Data.List (nub)
import Data.List.NonEmpty qualified as NonEmpty
import Data.Map.Strict qualified as Map
import Data.Maybe (catMaybes, mapMaybe)
import Data.Text qualified as Text
import Data.Time (NominalDiffTime, UTCTime (..), addDays, addUTCTime, fromGregorian)
import Data.UUID (UUID)
import Data.UUID qualified as UUID
import Effectful (Eff, IOE, (:>))
import Effectful.Exception (throwIO)
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Session qualified as Session
import Hasql.Statement (Statement, preparable)
import Keiro.Prelude
import Keiro.Workflow.Awakeable.Schema
  ( AwakeableStatus (..),
    lookupAwakeableStatusTx,
  )
import Keiro.Workflow.Child.Cancel (ensureChildCancelled)
import Keiro.Workflow.Child.Schema
  ( ChildStatus (..),
    lookupChild,
    reviveFailedChildTx,
  )
import Keiro.Workflow.Inspection.Query
import Keiro.Workflow.Instance.Schema
  ( WorkflowStatus (..),
    statusFromText,
    statusToText,
    upsertInstanceTx,
  )
import Keiro.Workflow.Journal
  ( JournalAppendOutcome (..),
    prepareJournalAppend,
  )
import Keiro.Workflow.Schema
  ( currentGeneration,
    deleteStepRowTx,
    loadStepIndex,
    lockWorkflowStepTx,
    lookupStepResultTx,
    workflowStepLockKey,
  )
import Keiro.Workflow.Types
  ( WorkflowError (..),
    WorkflowId (..),
    WorkflowJournalEvent (..),
    WorkflowName (..),
    awakeableStepPrefix,
    cancelledStepName,
    completedStepName,
    continuedAsNewStepName,
    failedStepName,
  )
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import "hasql-transaction" Hasql.Transaction qualified as Tx

data WorkflowInstanceRow = WorkflowInstanceRow
  { workflowId :: !Text,
    workflowName :: !Text,
    generation :: !Int32,
    status :: !WorkflowStatus,
    attempts :: !Int32,
    lastError :: !(Maybe Text),
    nextAttemptAt :: !(Maybe UTCTime),
    wakeAfter :: !(Maybe UTCTime),
    leasedBy :: !(Maybe Text),
    leaseExpiresAt :: !(Maybe UTCTime),
    createdAt :: !UTCTime,
    updatedAt :: !UTCTime,
    completedAt :: !(Maybe UTCTime)
  }
  deriving stock (Generic, Eq, Show)

-- | Filters and keyset cursor for operator-facing workflow enumeration.
--
-- Results are ordered by @(workflow_name, workflow_id)@. Pass the final row's
-- name and id as 'afterKey' to fetch the next page without the instability and
-- growing scan cost of an @OFFSET@ query. A non-positive 'pageSize' returns an
-- empty page.
data WorkflowInstanceFilter = WorkflowInstanceFilter
  { statuses :: !(Maybe (NonEmpty WorkflowStatus)),
    workflowName :: !(Maybe Text),
    afterKey :: !(Maybe (Text, Text)),
    pageSize :: !Int
  }
  deriving stock (Generic, Eq, Show)

-- | List every status and workflow name, starting at the first key, 100 rows
-- at a time.
defaultWorkflowInstanceFilter :: WorkflowInstanceFilter
defaultWorkflowInstanceFilter =
  WorkflowInstanceFilter
    { statuses = Nothing,
      workflowName = Nothing,
      afterKey = Nothing,
      pageSize = 100
    }

data ResurrectOutcome
  = WorkflowResurrected
  | WorkflowNotFailed
  | WorkflowNotFound
  deriving stock (Generic, Eq, Show)

-- | Honest result of an operator cancellation request.
data CancelWorkflowOutcome
  = WorkflowCancelRecorded
  | WorkflowAlreadyTerminal !WorkflowStatus
  | WorkflowCancelUnknown
  deriving stock (Generic, Eq, Show)

-- | Why 'claimInstance' did or did not acquire the advance lease.
data ClaimOutcome
  = -- | This caller now holds the lease.
    ClaimAcquired
  | -- | Another live lease exists; the instance is being advanced elsewhere.
    ClaimLeaseHeld
  | -- | The crash-backoff gate (@next_attempt_at@) is still in the future.
    ClaimPaced
  | -- | The row is terminal or gone; it will drop out of discovery by itself.
    ClaimUnavailable
  deriving stock (Generic, Eq, Show)

-- | Record that a run parked on @awaitedStep@, arbitrating against a wake
-- delivery that may be landing at the same moment.
--
-- Discovery is exact: a @suspended@ instance with no due wake hint is never
-- returned, so a suspended status written /after/ a wake has already been
-- delivered would strand the workflow forever. The window is real — a run
-- consults the step index, finds the awaited step absent, runs its arm, and only
-- then writes its status, and a wake can commit anywhere in between.
--
-- The fix reuses the lock the append path already takes. Every wake delivery
-- goes through @prepareJournalAppend@, which holds the per-step advisory lock
-- ('lockWorkflowStepTx' on 'workflowStepLockKey') while it appends and upserts
-- the instance row. Awakeable cancellation takes that lock too, but writes no
-- step row, so this transaction additionally consults the awakeable row's
-- terminal lifecycle status. Taking the same lock here totally orders the
-- writers:
--
-- * suspend wins the lock — it writes @suspended@; the wake, queued behind it,
--   then writes @running@;
-- * wake wins the lock — this transaction sees either the committed step-index
--   row or the terminal awakeable row and writes @running@ itself.
--
-- Either way no resolved wake is left behind a @suspended@ status. The
-- re-check reads the same authoritative @keiro_workflow_steps@ index the
-- @Await@ miss path consults, which is written in the same transaction as every
-- journal append. For @awk:@ steps it also reads @keiro_awakeables@, whose row
-- is the durable lifecycle authority for cancellation and completion. This
-- covers cancellation (the only wake transition without a step row) and the
-- self-repair arms whose append lands before their stale suspend write.
markInstanceSuspendedAwaiting ::
  (Store :> es) =>
  WorkflowName ->
  WorkflowId ->
  -- | The generation the run operated on.
  Int ->
  -- | The step name the run parked on.
  Text ->
  Eff es ()
markInstanceSuspendedAwaiting (WorkflowName nameText) (WorkflowId widText) gen awaitedStep =
  runTransaction $ do
    lockWorkflowStepTx (workflowStepLockKey widText nameText gen awaitedStep)
    resolved <- lookupStepResultTx widText nameText gen awaitedStep
    abandoned <- case resolved of
      Just _ -> pure False
      Nothing -> case awakeableUuidFromStep awaitedStep of
        Nothing -> pure False
        Just aid -> maybe False (/= Pending) <$> lookupAwakeableStatusTx aid
    let status = if isJust resolved || abandoned then WfRunning else WfSuspended
    upsertInstanceTx widText nameText (fromIntegral gen) status Nothing

awakeableUuidFromStep :: Text -> Maybe UUID
awakeableUuidFromStep stepName =
  UUID.fromText =<< Text.stripPrefix awakeableStepPrefix stepName

lookupInstance :: (Store :> es) => WorkflowName -> WorkflowId -> Eff es (Maybe WorkflowInstanceRow)
lookupInstance (WorkflowName name) (WorkflowId wid) =
  runTransaction (Tx.statement (wid, name) lookupInstanceStmt)

-- | Read one header in the caller's session. A single SELECT already observes
-- one database snapshot; inspection hosts need no transaction-control round
-- trips around it. This reuses the ordinary lookup's statement and decoder.
lookupInstanceSession :: WorkflowName -> WorkflowId -> Session.Session (Maybe WorkflowInstanceRow)
lookupInstanceSession (WorkflowName name) (WorkflowId wid) =
  Session.statement (wid, name) lookupInstanceStmt

-- | An exact image of all header fields, produced only by the owning query.
-- It is a representation key, never a timestamp, hash or transaction id.
data WorkflowHeaderKey = WorkflowHeaderKey !ByteString.ByteString !Text !Text
  deriving stock (Eq)

instance Show WorkflowHeaderKey where
  show _ = "<workflow header key>"

workflowHeaderKeyBytes :: WorkflowHeaderKey -> Int
workflowHeaderKeyBytes (WorkflowHeaderKey bytes _ _) = ByteString.length bytes

data WorkflowHeaderObservation
  = HeaderUnchanged
  | HeaderChanged !WorkflowHeaderKey !WorkflowInstanceRow
  | HeaderImageTooLarge
  deriving stock (Eq, Show)

-- | Recheck the complete current row in one SELECT. An equal image permits
-- reuse of its pure rendering; every request still observes the database.
-- The byte limit bounds the returned image before client decoding. Missing
-- rows, changed rows and oversized images remain distinct.
lookupInstanceConditionalSession :: Int -> Maybe WorkflowHeaderKey -> WorkflowName -> WorkflowId -> Session.Session (Maybe WorkflowHeaderObservation)
lookupInstanceConditionalSession limit known (WorkflowName name) (WorkflowId wid) =
  Session.statement (image, fromIntegral limit) lookupInstanceConditionalStmt
  where
    -- A complete prior image already contains the exact lookup identity. Reuse
    -- its encoded bytes, avoiding separate serialization of both key strings.
    -- A foreign identity starts with an incomplete image and cannot compare
    -- equal to the complete database header.
    image = case known of
      Just (WorkflowHeaderKey bytes previousName previousId)
        | name == previousName && wid == previousId -> bytes
      _ -> LazyByteString.toStrict (Aeson.encode (Aeson.object ["workflow_name" Aeson..= name, "workflow_id" Aeson..= wid]))

-- | Enumerate workflow instance summaries using stable keyset pagination.
--
-- Status and name filters are exact. Rows inserted before the supplied cursor
-- are intentionally not revisited; rows deleted or updated concurrently never
-- cause later keys to be skipped as an @OFFSET@ query could.
listWorkflowInstances ::
  (Store :> es) =>
  WorkflowInstanceFilter ->
  Eff es [WorkflowInstanceRow]
listWorkflowInstances filters
  | filters ^. #pageSize <= 0 = pure []
  | otherwise =
      runTransaction $
        Tx.statement
          ( NonEmpty.toList . fmap statusToText <$> filters ^. #statuses,
            filters ^. #workflowName,
            fst <$> filters ^. #afterKey,
            snd <$> filters ^. #afterKey,
            fromIntegral
              ( min
                  (filters ^. #pageSize)
                  (fromIntegral (maxBound :: Int32))
              ) ::
              Int32
          )
          listWorkflowInstancesStmt

-- | Stop a top-level or linked-child workflow at its next durable boundary.
--
-- Cancellation is an append-only journal marker. Linked children delegate to
-- the same transaction as 'Keiro.Workflow.Child.cancelChild' so their parent is
-- woken with the typed cancellation sentinel. Children are not cascaded: an
-- operator must cancel descendants explicitly. A step action already in flight
-- may finish and journal idempotently; no later boundary may start.
cancelWorkflow ::
  (IOE :> es, Store :> es) =>
  WorkflowName ->
  WorkflowId ->
  Eff es CancelWorkflowOutcome
cancelWorkflow name@(WorkflowName nameText) wid@(WorkflowId widText) =
  lookupInstance name wid >>= \case
    Just row
      | Just terminal <- terminalStatus (row ^. #status) ->
          pure (WorkflowAlreadyTerminal terminal)
    mrow -> do
      exists <- case mrow of
        Just _ -> pure True
        Nothing -> do
          gen <- currentGeneration name wid
          not . Map.null <$> loadStepIndex name wid gen
      if not exists
        then pure WorkflowCancelUnknown
        else
          lookupChild widText nameText >>= \case
            Just child -> cancelLinkedChild child
            Nothing -> cancelTopLevel
  where
    terminalStatus = \case
      WfCompleted -> Just WfCompleted
      WfCancelled -> Just WfCancelled
      WfFailed -> Just WfFailed
      _ -> Nothing

    cancelLinkedChild child = do
      (transitioned, childOutcome, parentOutcome) <- ensureChildCancelled child
      traverse_ throwOnJournalConflict [childOutcome, parentOutcome]
      if transitioned
        then pure WorkflowCancelRecorded
        else case child ^. #status of
          ChildCancelled -> pure (WorkflowAlreadyTerminal WfCancelled)
          ChildCompleted -> pure (WorkflowAlreadyTerminal WfCompleted)
          ChildFailed -> pure (WorkflowAlreadyTerminal WfFailed)
          -- The guarded child-row transition lost a race. Re-read both the
          -- instance and child rows before deciding which terminal state won.
          Running -> cancelWorkflow name wid

    cancelTopLevel = do
      gen <- currentGeneration name wid
      now <- liftIO getCurrentTime
      appendTx <-
        prepareJournalAppend
          name
          wid
          gen
          WorkflowCancelled {recordedAt = now}
      runTransaction appendTx >>= \case
        JournalAppended {} -> pure WorkflowCancelRecorded
        JournalAlreadyPresent {} ->
          pure (WorkflowAlreadyTerminal WfCancelled)
        JournalRefusedTerminal marker
          | marker == continuedAsNewStepName -> cancelWorkflow name wid
          | marker == completedStepName -> pure (WorkflowAlreadyTerminal WfCompleted)
          | marker == failedStepName -> pure (WorkflowAlreadyTerminal WfFailed)
          | marker == cancelledStepName -> pure (WorkflowAlreadyTerminal WfCancelled)
          | otherwise -> cancelWorkflow name wid
        conflict@JournalAppendConflict {} ->
          throwOnJournalConflict conflict *> pure WorkflowCancelUnknown

    throwOnJournalConflict = \case
      JournalAppendConflict err ->
        throwIO (WorkflowJournalAppendError (Text.pack (show err)))
      _ -> pure ()

-- | Claim one runnable workflow instance and classify any refusal.
--
-- A live lease takes precedence over crash pacing because it means another
-- worker is actively responsible for the instance. The guarded update and the
-- refusal classification share one transaction and one clock reading.
claimInstance :: (IOE :> es, Store :> es) => Text -> NominalDiffTime -> WorkflowName -> WorkflowId -> Eff es ClaimOutcome
claimInstance owner ttl (WorkflowName nameText) (WorkflowId widText) = do
  now <- liftIO getCurrentTime
  runTransaction $ do
    -- Generation 0, not the resolved current generation: 'ensureInstanceStmt' is
    -- an ON CONFLICT DO NOTHING insert, so the value is used only when no row
    -- exists at all — and every discovered workflow has one (migration 0011
    -- backfilled the pre-existing instances, every append upserts, and spawnChild
    -- writes the child's row inside the spawn step's transaction). Where the
    -- insert does fire, 0 is a floor that the next truthful writer raises:
    -- 'upsertInstanceTx' takes GREATEST(stored, supplied) on conflict. Resolving
    -- MAX(generation) here would cost a query per claim to learn a number the
    -- insert almost never uses.
    Tx.statement (widText, nameText, 0 :: Int32) ensureInstanceStmt
    claimed <-
      Tx.statement
        (widText, nameText, owner, now, addUTCTime ttl now)
        claimInstanceStmt
    case claimed of
      Just True -> pure ClaimAcquired
      _ -> do
        state <- Tx.statement (widText, nameText) classifyClaimRefusalStmt
        pure $ case state of
          Nothing -> ClaimUnavailable
          Just (status, leaseExpiry, nextAttempt)
            | status `notElem` [WfRunning, WfSuspended] -> ClaimUnavailable
            | maybe False (>= now) leaseExpiry -> ClaimLeaseHeld
            | maybe False (> now) nextAttempt -> ClaimPaced
            | otherwise -> ClaimUnavailable

-- | Extend an instance lease only when @owner@ still holds it.
--
-- The caller supplies one clock reading so @updated_at@ and the new expiry share
-- the same boundary. Returns 'False' after ownership is lost or the row vanishes.
renewInstanceLeaseTx ::
  Text ->
  NominalDiffTime ->
  UTCTime ->
  Text ->
  Text ->
  Tx.Transaction Bool
renewInstanceLeaseTx owner ttl now wid name =
  Tx.statement
    (wid, name, owner, now, addUTCTime ttl now)
    renewInstanceLeaseStmt

-- | Effect-level wrapper around 'renewInstanceLeaseTx' using the current time.
renewInstanceLease ::
  (IOE :> es, Store :> es) =>
  Text ->
  NominalDiffTime ->
  WorkflowName ->
  WorkflowId ->
  Eff es Bool
renewInstanceLease owner ttl (WorkflowName name) (WorkflowId wid) = do
  now <- liftIO getCurrentTime
  runTransaction (renewInstanceLeaseTx owner ttl now wid name)

releaseInstance :: (Store :> es) => Text -> Bool -> WorkflowName -> WorkflowId -> Eff es ()
releaseInstance owner progressed (WorkflowName name) (WorkflowId wid) =
  runTransaction $
    Tx.statement (wid, name, owner, progressed) releaseInstanceStmt

-- | Clear any current instance lease, returning 'True' only when a lease was
-- present.
--
-- A previous live owner is not interrupted inside an action already in flight.
-- Its next owner-guarded 'renewInstanceLease' matches no row and raises
-- 'Keiro.Workflow.WorkflowLeaseLost' before a later workflow boundary, while a
-- replacement owner can claim immediately instead of waiting for the old TTL.
forceReleaseInstanceLease ::
  (Store :> es) =>
  WorkflowName ->
  WorkflowId ->
  Eff es Bool
forceReleaseInstanceLease (WorkflowName name) (WorkflowId wid) =
  runTransaction (Tx.statement (wid, name) forceReleaseInstanceLeaseStmt)

-- | Record a crashed advance against the instance row: bump @attempts@, store
-- the rendered error, and push @next_attempt_at@ out along the backoff ladder.
-- Returns the new attempt count, or 'Nothing' when the row matched nothing
-- because the workflow reached a terminal status between the crash and this
-- update (a parent's @cancelChild@, an operator cancellation, a concurrent
-- failure marker). That race is ordinary, not exceptional: there is no live
-- instance left to pace, so nothing is recorded and the caller skips it.
recordCrashTx :: Text -> Text -> Text -> Tx.Transaction (Maybe Int32)
recordCrashTx wid name err =
  Tx.statement (wid, name, err) recordCrashStmt

resetInstanceAttempts :: (Store :> es) => WorkflowName -> WorkflowId -> Eff es ()
resetInstanceAttempts (WorkflowName name) (WorkflowId wid) =
  runTransaction (Tx.statement (wid, name) resetInstanceAttemptsStmt)

reviveFailedInstanceTx :: Text -> Text -> Tx.Transaction Bool
reviveFailedInstanceTx wid name =
  Tx.statement (wid, name) reviveFailedInstanceStmt

-- | Return a terminally failed workflow to the runnable pool.
--
-- The operation removes only the current generation's derived failed-marker index
-- row; the append-only 'Keiro.Workflow.WorkflowFailed' journal event remains as
-- history. A failed child link is revived in the same transaction. Parent failure
-- sentinels already delivered to another journal are not retracted.
resurrectFailedWorkflow ::
  (Store :> es) =>
  WorkflowName ->
  WorkflowId ->
  Eff es ResurrectOutcome
resurrectFailedWorkflow name@(WorkflowName nameText) wid@(WorkflowId widText) =
  lookupInstance name wid >>= \case
    Nothing -> pure WorkflowNotFound
    Just row
      | row ^. #status /= WfFailed -> pure WorkflowNotFailed
      | otherwise -> do
          gen <- currentGeneration name wid
          revived <-
            runTransaction $ do
              instanceRevived <- reviveFailedInstanceTx widText nameText
              when instanceRevived $ do
                deleteStepRowTx widText nameText gen failedStepName
                void (reviveFailedChildTx widText nameText)
              pure instanceRevived
          pure $
            if revived
              then WorkflowResurrected
              else WorkflowNotFailed

lookupInstanceStmt :: Statement (Text, Text) (Maybe WorkflowInstanceRow)
lookupInstanceStmt =
  preparable
    """
    SELECT workflow_id, workflow_name, generation, status, attempts,
           last_error, next_attempt_at, wake_after, leased_by, lease_expires_at,
           created_at, updated_at, completed_at
    FROM keiro.keiro_workflows
    WHERE workflow_id = $1 AND workflow_name = $2
    """
    ( contrazip2
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    (D.rowMaybe instanceRowDecoder)

lookupInstanceConditionalStmt :: Statement (ByteString.ByteString, Int64) (Maybe WorkflowHeaderObservation)
{-# NOINLINE lookupInstanceConditionalStmt #-}
lookupInstanceConditionalStmt =
  preparable
    """
    WITH header AS MATERIALIZED (
      SELECT workflow_id, workflow_name, generation, status, attempts,
             last_error,
             CASE WHEN next_attempt_at IS NOT NULL THEN CASE WHEN isfinite(next_attempt_at) THEN jsonb_build_array(
               (next_attempt_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (next_attempt_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(next_attempt_at::text) END END AS next_attempt_at,
             CASE WHEN wake_after IS NOT NULL THEN CASE WHEN isfinite(wake_after) THEN jsonb_build_array(
               (wake_after AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (wake_after AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(wake_after::text) END END AS wake_after,
             leased_by,
             CASE WHEN lease_expires_at IS NOT NULL THEN CASE WHEN isfinite(lease_expires_at) THEN jsonb_build_array(
               (lease_expires_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (lease_expires_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(lease_expires_at::text) END END AS lease_expires_at,
             CASE WHEN isfinite(created_at) THEN jsonb_build_array((created_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (created_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(created_at::text) END AS created_at,
             CASE WHEN isfinite(updated_at) THEN jsonb_build_array((updated_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (updated_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(updated_at::text) END AS updated_at,
             CASE WHEN completed_at IS NOT NULL THEN CASE WHEN isfinite(completed_at) THEN jsonb_build_array(
               (completed_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',
               extract(epoch FROM (completed_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(completed_at::text) END END AS completed_at
      FROM keiro.keiro_workflows
      WHERE workflow_id = ($1::jsonb ->> 'workflow_id')
        AND workflow_name = ($1::jsonb ->> 'workflow_name')
    ), image AS MATERIALIZED (
      SELECT CASE
        WHEN $2 <= 0 OR octet_length(workflow_id)::bigint
             + octet_length(workflow_name)::bigint
             + coalesce(octet_length(last_error)::bigint,0)
             + coalesce(octet_length(leased_by)::bigint,0) > $2
          THEN NULL
        ELSE to_jsonb(header)
      END AS value FROM header
    ), checked AS MATERIALIZED (
      SELECT value, value IS NULL OR octet_length(value::text) > $2 AS oversized
      FROM image
    )
    SELECT oversized,
           CASE WHEN oversized OR value = $1::jsonb THEN NULL ELSE value END
    FROM checked
    """
    ( contrazip2
        (E.param (E.nonNullable E.jsonbBytes))
        (E.param (E.nonNullable E.int8))
    )
    (D.rowMaybe (observation <$> D.column (D.nonNullable D.bool) <*> D.column (D.nullable (D.jsonbBytes decodeImage))))
  where
    observation True _ = HeaderImageTooLarge
    observation False Nothing = HeaderUnchanged
    observation False (Just (key, row)) = HeaderChanged key row
    decodeImage bytes = do
      value <- either (Left . Text.pack) Right (Aeson.eitherDecodeStrict' bytes)
      row <- either (Left . Text.pack) Right (parseEither parseHeader value)
      pure (WorkflowHeaderKey bytes (row ^. #workflowName) (row ^. #workflowId), row)
    parseHeader = Aeson.withObject "workflow header image" $ \row ->
      WorkflowInstanceRow
        <$> row .: "workflow_id"
        <*> row .: "workflow_name"
        <*> row .: "generation"
        <*> (statusFromText <$> row .: "status")
        <*> row .: "attempts"
        <*> row .: "last_error"
        <*> (row .: "next_attempt_at" >>= traverse timestampImage)
        <*> (row .: "wake_after" >>= traverse timestampImage)
        <*> row .: "leased_by"
        <*> (row .: "lease_expires_at" >>= traverse timestampImage)
        <*> (row .: "created_at" >>= timestampImage)
        <*> (row .: "updated_at" >>= timestampImage)
        <*> (row .: "completed_at" >>= traverse timestampImage)
    -- Separate days from seconds so epoch division cannot round a distant
    -- timestamp's microseconds. BC dates also avoid textual JSON date parsing.
    -- Match the existing binary decoder's Int64 microsecond interpretation
    -- of PostgreSQL's infinity sentinels, rather than changing legacy JSON.
    timestampImage (Aeson.String "infinity") = pure (sentinelTime (maxBound :: Int64))
    timestampImage (Aeson.String "-infinity") = pure (sentinelTime (minBound :: Int64))
    timestampImage value = do
      (days, seconds) <- Aeson.parseJSON value
      Aeson.withScientific
        "timestamp seconds"
        (\s -> pure (UTCTime (addDays days (fromGregorian 2000 1 1)) (fromRational (toRational s))))
        seconds
    sentinelTime microseconds =
      let (days, remainder) = divMod (toInteger microseconds) (86400 * 1000000)
       in UTCTime (addDays days (fromGregorian 2000 1 1)) (fromRational (toRational remainder / 1000000))

listWorkflowInstancesStmt :: Statement (Maybe [Text], Maybe Text, Maybe Text, Maybe Text, Int32) [WorkflowInstanceRow]
listWorkflowInstancesStmt =
  preparable
    """
    SELECT workflow_id, workflow_name, generation, status, attempts,
           last_error, next_attempt_at, wake_after, leased_by, lease_expires_at,
           created_at, updated_at, completed_at
    FROM keiro.keiro_workflows
    WHERE ($1::text[] IS NULL OR status = ANY($1))
      AND ($2::text IS NULL OR workflow_name = $2)
      AND (
        $3::text IS NULL
        OR (workflow_name, workflow_id) > ($3, $4)
      )
    ORDER BY workflow_name, workflow_id
    LIMIT $5
    """
    ( contrazip5
        (E.param (E.nullable (E.foldableArray (E.nonNullable E.text))))
        (E.param (E.nullable E.text))
        (E.param (E.nullable E.text))
        (E.param (E.nullable E.text))
        (E.param (E.nonNullable E.int4))
    )
    (D.rowList instanceRowDecoder)

ensureInstanceStmt :: Statement (Text, Text, Int32) ()
ensureInstanceStmt =
  preparable
    """
    INSERT INTO keiro.keiro_workflows
      (workflow_id, workflow_name, generation, status)
    VALUES ($1, $2, $3, 'running')
    ON CONFLICT (workflow_id, workflow_name) DO NOTHING
    """
    ( contrazip3
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.int4))
    )
    D.noResult

claimInstanceStmt :: Statement (Text, Text, Text, UTCTime, UTCTime) (Maybe Bool)
claimInstanceStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET leased_by = $3,
        lease_expires_at = $5,
        updated_at = $4
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND status IN ('running', 'suspended')
      AND (lease_expires_at IS NULL OR lease_expires_at < $4)
      AND (next_attempt_at IS NULL OR next_attempt_at <= $4)
    RETURNING TRUE
    """
    ( contrazip5
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.timestamptz))
        (E.param (E.nonNullable E.timestamptz))
    )
    (D.rowMaybe (D.column (D.nonNullable D.bool)))

classifyClaimRefusalStmt :: Statement (Text, Text) (Maybe (WorkflowStatus, Maybe UTCTime, Maybe UTCTime))
classifyClaimRefusalStmt =
  preparable
    """
    SELECT status, lease_expires_at, next_attempt_at
    FROM keiro.keiro_workflows
    WHERE workflow_id = $1 AND workflow_name = $2
    """
    ( contrazip2
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    ( D.rowMaybe
        ( (,,)
            <$> (statusFromText <$> D.column (D.nonNullable D.text))
            <*> D.column (D.nullable D.timestamptz)
            <*> D.column (D.nullable D.timestamptz)
        )
    )

renewInstanceLeaseStmt :: Statement (Text, Text, Text, UTCTime, UTCTime) Bool
renewInstanceLeaseStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET lease_expires_at = $5,
        updated_at = $4
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND leased_by = $3
    """
    ( contrazip5
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.timestamptz))
        (E.param (E.nonNullable E.timestamptz))
    )
    ((> 0) <$> D.rowsAffected)

releaseInstanceStmt :: Statement (Text, Text, Text, Bool) ()
releaseInstanceStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET leased_by = NULL,
        lease_expires_at = NULL,
        attempts = CASE WHEN $4 THEN 0 ELSE attempts END,
        last_error = CASE WHEN $4 THEN NULL ELSE last_error END,
        next_attempt_at = CASE WHEN $4 THEN NULL ELSE next_attempt_at END,
        updated_at = now()
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND leased_by = $3
    """
    ( contrazip4
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.bool))
    )
    D.noResult

forceReleaseInstanceLeaseStmt :: Statement (Text, Text) Bool
forceReleaseInstanceLeaseStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET leased_by = NULL,
        lease_expires_at = NULL,
        updated_at = now()
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND leased_by IS NOT NULL
    """
    ( contrazip2
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    ((> 0) <$> D.rowsAffected)

recordCrashStmt :: Statement (Text, Text, Text) (Maybe Int32)
recordCrashStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET attempts = attempts + 1,
        last_error = $3,
        next_attempt_at = now() + (LEAST(power(2, attempts + 1), 64) * interval '1 second'),
        updated_at = now()
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND status NOT IN ('completed', 'cancelled', 'failed')
    RETURNING attempts
    """
    ( contrazip3
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    -- 'rowMaybe', not 'singleRow': the WHERE clause deliberately matches
    -- nothing once the workflow is terminal, and a workflow can go terminal
    -- between its crash and this update.
    (D.rowMaybe (D.column (D.nonNullable D.int4)))

resetInstanceAttemptsStmt :: Statement (Text, Text) ()
resetInstanceAttemptsStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET attempts = 0,
        last_error = NULL,
        next_attempt_at = NULL,
        updated_at = now()
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND status NOT IN ('completed', 'cancelled', 'failed')
    """
    ( contrazip2
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    D.noResult

reviveFailedInstanceStmt :: Statement (Text, Text) Bool
reviveFailedInstanceStmt =
  preparable
    """
    UPDATE keiro.keiro_workflows
    SET status = 'running',
        attempts = 0,
        last_error = NULL,
        next_attempt_at = NULL,
        leased_by = NULL,
        lease_expires_at = NULL,
        completed_at = NULL,
        updated_at = now()
    WHERE workflow_id = $1
      AND workflow_name = $2
      AND status = 'failed'
    """
    ( contrazip2
        (E.param (E.nonNullable E.text))
        (E.param (E.nonNullable E.text))
    )
    ((> 0) <$> D.rowsAffected)

instanceRowDecoder :: D.Row WorkflowInstanceRow
instanceRowDecoder = instanceFields (D.column . D.nonNullable) (D.column . D.nullable)

-- Shared by ordinary reads and the nullable row of an inspection outer join.
-- The ordinary decoder retains the exact same columns and nullability.
instanceFields :: (Applicative f) => (forall a. D.Value a -> f a) -> (forall a. D.Value a -> f (Maybe a)) -> f WorkflowInstanceRow
instanceFields required optional =
  WorkflowInstanceRow
    <$> required D.text
    <*> required D.text
    <*> required D.int4
    <*> (statusFromText <$> required D.text)
    <*> required D.int4
    <*> optional D.text
    <*> optional D.timestamptz
    <*> optional D.timestamptz
    <*> optional D.text
    <*> optional D.timestamptz
    <*> required D.timestamptz
    <*> required D.timestamptz
    <*> optional D.timestamptz

-- | Inspection-only parameterized statement shapes. Absent filters are omitted
-- from SQL rather than hidden behind nullable ORs that defeat generic plans.
-- The facade validates the size and cursor before calling this primitive.
-- The outer join returns the database observation clock even for an empty page.
inspectionInstanceRows :: (Store :> es) => WorkflowListQuery -> Maybe WorkflowListPosition -> Int -> Eff es (UTCTime, [WorkflowInstanceRow])
inspectionInstanceRows query position count = do
  rows <- runTransaction (Tx.statement () (inspectionInstanceStatement query position count))
  case rows of
    (clock, _) : _ -> pure (clock, mapMaybe snd rows)
    [] -> error "inspection clock outer join returned no row"

inspectionInstanceStatement :: WorkflowListQuery -> Maybe WorkflowListPosition -> Int -> Statement () [(UTCTime, Maybe WorkflowInstanceRow)]
inspectionInstanceStatement query position count = preparable sql params decoder
  where
    number = Text.pack . show
    parameter :: Text -> E.Value a -> a -> (Int -> Text, Int, E.Params ())
    parameter template value argument = (\n -> Text.replace "?" ("$" <> number n) template, 1, const argument >$< E.param (E.nonNullable value))
    optional template value = fmap (parameter template value)
    selectedClasses = nub (maybe [] NonEmpty.toList query.classes)
    filters =
      [(const ("(" <> Text.intercalate " OR " ["(" <> classPredicate status <> ")" | status <- selectedClasses] <> ")"), 0, mempty) | query.classes /= Nothing]
        <> catMaybes
          [ optional "workflow_name = ?" E.text query.workflowName,
            optional "created_at >= ?" E.timestamptz query.createdSince,
            optional "created_at < ?" E.timestamptz query.createdBefore,
            optional "updated_at >= ?" E.timestamptz query.updatedSince,
            optional "updated_at < ?" E.timestamptz query.updatedBefore
          ]
        <> cursorParts
    cursorParts = case position of
      Nothing -> []
      Just (AfterWorkflowKey wid name) ->
        [ ( \n -> "(workflow_id, workflow_name) > ($" <> number n <> ",$" <> number (n + 1) <> ")",
            2,
            (const wid >$< E.param (E.nonNullable E.text)) <> (const name >$< E.param (E.nonNullable E.text))
          )
        ]
      Just (AfterWorkflowCreated created name wid) ->
        [ ( \n -> "(created_at, workflow_name, workflow_id) < ($" <> number n <> ",$" <> number (n + 1) <> ",$" <> number (n + 2) <> ")",
            3,
            (const created >$< E.param (E.nonNullable E.timestamptz)) <> (const name >$< E.param (E.nonNullable E.text)) <> (const wid >$< E.param (E.nonNullable E.text))
          )
        ]
    (next, predicates, filterParams) = foldl' (\(index, clauses, encoders) (render, width, encoder) -> (index + width, clauses <> [render index], encoders <> encoder)) (1, [], mempty) filters
    order = case query.ordering of ByKey -> "workflow_id, workflow_name"; ByCreatedDesc -> "created_at DESC, workflow_name DESC, workflow_id DESC"
    columns = "workflow_id, workflow_name, generation, status, attempts, last_error, next_attempt_at, wake_after, leased_by, lease_expires_at, created_at, updated_at, completed_at"
    -- Classification is inspection-only. Reuse existing status/key access;
    -- no generated column or inspection index adds work to ordinary writes.
    classPredicate = \case
      ClassRunning -> "status = 'running' AND attempts = 0"
      ClassRetrying -> "status = 'running' AND attempts <> 0"
      ClassSleeping -> "status = 'suspended' AND wake_after IS NOT NULL"
      ClassAwaiting -> "status = 'suspended' AND wake_after IS NULL"
      ClassCompleted -> "status = 'completed'"
      ClassCancelled -> "status = 'cancelled'"
      ClassFailed -> "status = 'failed'"
    sql =
      "/* keiro-inspection-page-v2 */ WITH inspection_page AS MATERIALIZED (SELECT "
        <> columns
        <> " FROM keiro.keiro_workflows WHERE "
        <> Text.intercalate " AND " ("true" : predicates)
        <> " ORDER BY "
        <> order
        <> " LIMIT $"
        <> number next
        <> ") SELECT now(), inspection_page.* FROM (VALUES (1)) AS clock(x) LEFT JOIN inspection_page ON true ORDER BY "
        <> order
    params = filterParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4))
    decoder = D.rowList ((,) <$> D.column (D.nonNullable D.timestamptz) <*> getCompose (instanceFields (Compose . D.column . D.nullable) (Compose . fmap Just . D.column . D.nullable)))