Awakeable.hs

-- | Awakeables: durable promises an external system resolves.
--
-- == What this gives you
--
-- A workflow allocates an opaque 'AwakeableId', hands it to some external system
-- (a webhook handler, a human approver, an LLM tool call), and suspends until
-- that system /signals/ the id with a result — no polling, no bespoke
-- "wait for event X then react in a process manager" wiring.
--
-- @
-- approvalFlow :: ('Workflow' ':>' es, 'Store' ':>' es, 'IOE' ':>' es) => Eff es Text
-- approvalFlow = do
--   (aid, await) <- 'awakeableNamed' (StepName \"approval\")  -- allocate the promise
--   -- (hand @aid@ to a webhook handler / human / LLM tool here)
--   decision <- await                                       -- SUSPEND until signalled
--   'Keiro.Workflow.step' (StepName \"use\") (pure (decision <> \"!\"))
-- @
--
-- The @pending@ row is committed as part of the journaled allocation step,
-- before the id can be returned or handed to an external system. On the __first__
-- 'Keiro.Workflow.runWorkflow' this returns 'Suspended' (the run parked on
-- @await@). An external caller later runs
-- @'signalAwakeable' aid \"ok\"@, which flips the row to @completed@ /and/
-- appends a @StepRecorded \"awk:\<uuid\>\"@ to the workflow's journal. The
-- __next__ run replays past the now-resolved @await@ and 'Completed's.
--
-- == Contract recap for downstream plans (the v2 MasterPlan)
--
-- * 'AwakeableId' is journaled randomness: new allocations generate an opaque v4
--   UUID and record it under @awkid:\<label\>@ before awaiting @awk:\<uuid\>@.
--   Replay reads the journaled id, so a resumed workflow allocates the same id it
--   already handed out without making that id guessable from public coordinates.
-- * 'awakeableNamed' (caller-supplied label) is the __stable primitive__;
--   'awakeable' is an ordinal convenience whose label is positional (a fragile
--   derivation across code edits — see its Haddock).
-- * Awakeables journal their completion as ordinary 'StepRecorded' events under
--   the reserved @awk:@ prefix ('Keiro.Workflow.awakeableStepPrefix'), never a
--   new event type, so EP-38's replay loop stays uniform.
-- * 'signalAwakeable' is idempotent /and/ crash-safe: it commits the row update
--   and journal append in one transaction for new signals, a double signal
--   returns 'False' and does not change the recorded value, and a signal of an
--   already-@completed@ awakeable re-appends the journal entry from the stored
--   payload to repair historical wedges. A signal that loses a row race to
--   cancellation returns 'False' without appending.
-- * 'cancelAwakeable' abandons a still-@pending@ promise; a workflow that
--   re-enters its @await@ then throws 'WorkflowAwakeableCancelled', which the
--   author can @catch@ for compensation. If uncaught, EP-42's resume worker
--   records an attempt, backs off, and eventually appends 'WorkflowFailed' at the
--   configured failure ceiling.
-- * @countPendingAwakeables@ (in "Keiro.Workflow.Awakeable.Schema") backs EP-44's
--   @keiro.workflow.awakeables.pending@ gauge.
module Keiro.Workflow.Awakeable
  ( -- * Awakeable ids
    AwakeableId (..),
    awakeableIdToUuid,
    awakeableIdText,

    -- * Authoring surface (inside a workflow)
    awakeableNamed,
    awakeable,

    -- * External completion (outside a workflow)
    signalAwakeable,
    signalAwakeableFrom,
    cancelAwakeable,

    -- * Errors
    WorkflowAwakeableCancelled (..),
  )
where

import Control.Exception (Exception)
import Data.List.NonEmpty qualified as NonEmpty
import Data.Text qualified as Text
import Data.UUID (UUID)
import Data.UUID qualified as UUID
import Data.UUID.V4 qualified as UUID.V4
import Effectful (Eff, IOE, (:>))
import Effectful.Exception (throwIO)
import Keiro.Prelude
import Keiro.Workflow
  ( JournalAppendOutcome (..),
    StepName (..),
    Workflow,
    WorkflowError (..),
    WorkflowId (..),
    WorkflowJournalEvent (..),
    WorkflowName (..),
    appendJournalEntry,
    awaitStep,
    awakeableAllocStepPrefix,
    awakeableStepPrefix,
    currentGeneration,
    currentRunGeneration,
    currentWorkflow,
    freshOrdinal,
    prepareJournalAppend,
    step,
  )
import Keiro.Workflow.Awakeable.Internal.Identity (generation0AwakeableUuidProbes)
import Keiro.Workflow.Awakeable.Schema
  ( AwakeableRow,
    AwakeableStatus (..),
    cancelAwakeableTx,
    completeAwakeableTx,
    lookupAwakeable,
    lookupAwakeableStatusTx,
    registerAwakeableTx,
  )
import Keiro.Workflow.Instance (WorkflowStatus (..), upsertInstanceTx)
import Keiro.Workflow.Schema (lockWorkflowStepTx, workflowStepLockKey)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Transaction (runTransaction)
import "hasql-transaction" Hasql.Transaction qualified as Tx

-- ---------------------------------------------------------------------------
-- Awakeable ids
-- ---------------------------------------------------------------------------

-- | The opaque id of an awakeable. New allocations are random and journaled by
-- 'awakeableNamed'. Frozen generation-0 probes live in the explicitly named
-- "Keiro.Workflow.Awakeable.Compatibility" module. The @ToJSON@\/@FromJSON@
-- instances are over the
-- inner UUID, so the workflow journal can replay the id and webhook payloads may
-- carry it.
newtype AwakeableId = AwakeableId UUID
  deriving stock (Eq, Show, Generic)
  deriving newtype (ToJSON, FromJSON)

-- | The raw UUID inside an 'AwakeableId'.
awakeableIdToUuid :: AwakeableId -> UUID
awakeableIdToUuid (AwakeableId u) = u

-- | The 'AwakeableId' rendered as text — the suffix of the @awk:\<uuid\>@
-- journal step name an awakeable's completion is recorded under.
awakeableIdText :: AwakeableId -> Text
awakeableIdText = UUID.toText . awakeableIdToUuid

-- ---------------------------------------------------------------------------
-- Errors
-- ---------------------------------------------------------------------------

-- | Thrown out of 'Keiro.Workflow.runWorkflow' when a workflow re-enters the
-- @await@ of an awakeable that was 'cancelAwakeable'd. A cancelled awakeable will
-- never be signalled, so suspending forever would be wrong and silently
-- completing would fabricate a result; the workflow author can @catch@ this to
-- run compensation. If uncaught, the resume worker records the attempt and
-- eventually marks the workflow failed at its configured ceiling.
newtype WorkflowAwakeableCancelled = WorkflowAwakeableCancelled AwakeableId
  deriving stock (Eq, Show)

instance Exception WorkflowAwakeableCancelled

-- ---------------------------------------------------------------------------
-- Authoring surface
-- ---------------------------------------------------------------------------

-- | Allocate an awakeable under the stable, caller-supplied @label@. Returns
-- the 'AwakeableId' (hand it to the external system) and an @await@ action that
-- 'Suspended's the workflow until the awakeable is signalled, then yields the
-- decoded payload on a later run.
--
-- The @label@ is the only fully-deterministic option: it survives code edits that
-- insert or remove awakeables elsewhere in the workflow (the same robustness
-- argument EP-38 makes for named steps over positional history). Prefer this over
-- 'awakeable' for anything that may outlive a code change mid-flight.
--
-- The label is stable across code edits but __not__ across
-- 'Keiro.Workflow.continueAsNew'. The allocated id is journaled under an
-- @awkid:\<label\>@ step, and rotation opens a generation whose journal has no
-- such step, so the next run allocates a fresh id under the same label. The
-- previously handed-out id is orphaned: its row survives until workflow GC
-- collects it, 'signalAwakeable' against it still reports whether /that row/
-- transitioned, and nothing in the new generation is woken. Re-notify the holder
-- from the re-run allocation step whenever a workflow both rotates and hands
-- awakeable ids to the outside world.
awakeableNamed ::
  (Workflow :> es, Store :> es, IOE :> es, FromJSON a) =>
  StepName ->
  Eff es (AwakeableId, Eff es a)
awakeableNamed (StepName label) = do
  (name, wid) <- currentWorkflow
  gen <- currentRunGeneration
  aid <-
    step (StepName (awakeableAllocStepPrefix <> label)) $ do
      allocated <- allocateAwakeableId name wid gen label
      runTransaction $
        registerAwakeableTx
          (awakeableIdToUuid allocated)
          (unWorkflowName name)
          (unWorkflowId wid)
      pure allocated
  let stepNm = StepName (awakeableStepPrefix <> awakeableIdText aid)
      await = awaitCancellable name wid aid stepNm
  pure (aid, await)

allocateAwakeableId ::
  (Store :> es, IOE :> es) =>
  WorkflowName ->
  WorkflowId ->
  Int ->
  Text ->
  Eff es AwakeableId
allocateAwakeableId name wid gen label
  | gen <= 0 = adopt (NonEmpty.toList probes)
  | otherwise = freshAwakeableId
  where
    probes = fmap AwakeableId (generation0AwakeableUuidProbes name wid label)
    freshAwakeableId = AwakeableId <$> liftIO UUID.V4.nextRandom
    adopt [] = freshAwakeableId
    adopt (candidate : rest) = do
      row <- lookupAwakeable (awakeableIdToUuid candidate)
      case row of
        Just _ -> pure candidate
        Nothing -> adopt rest

-- | Allocate an awakeable under an ordinal label (the @N@th awakeable in a run
-- becomes @ord:N@). Convenient, but its determinism is __conditional__: adding or
-- removing an 'awakeable' call earlier in the workflow shifts every later ordinal
-- and so changes their derived ids, which corrupts an in-flight workflow exactly
-- the way EP-38 warns positional history does. Prefer 'awakeableNamed' for
-- anything that may outlive a code edit.
awakeable ::
  (Workflow :> es, Store :> es, IOE :> es, FromJSON a) =>
  Eff es (AwakeableId, Eff es a)
awakeable = do
  n <- freshOrdinal awakeableStepPrefix
  awakeableNamed (StepName ("ord:" <> Text.pack (show n)))

-- | EP-38's 'awaitStep', wrapped so that a re-entered @await@ on a
-- 'Cancelled' awakeable throws 'WorkflowAwakeableCancelled' instead of suspending
-- forever. The check lives /inside/ the arming action because 'awaitStep' runs
-- @arm@ only on the miss path (the awakeable not yet journaled) and re-runs it on
-- every resume until it resolves: on a miss we either notice the cancel and throw,
-- or (re-)register the idempotent @pending@ row and suspend. On the hit path
-- ('signalAwakeable' already journaled the result) @arm@ is never run, so a
-- signalled-then-cancelled race still returns the signalled value (signal wins — a
-- resolved promise cannot be un-resolved).
awaitCancellable ::
  (Workflow :> es, Store :> es, IOE :> es, FromJSON a) =>
  WorkflowName -> WorkflowId -> AwakeableId -> StepName -> Eff es a
awaitCancellable name wid aid stepNm =
  awaitStep stepNm $ do
    existing <- lookupAwakeable (awakeableIdToUuid aid)
    case existing of
      Just row
        | row ^. #status == Cancelled ->
            throwIO (WorkflowAwakeableCancelled aid)
        | row ^. #status == Completed,
          Just payload <- row ^. #payload -> do
            now <- liftIO getCurrentTime
            appendJournalEntry
              name
              wid
              StepRecorded
                { stepName = unStepName stepNm,
                  result = payload,
                  recordedAt = now
                }
      _ ->
        runTransaction $
          registerAwakeableTx (awakeableIdToUuid aid) (unWorkflowName name) (unWorkflowId wid)

-- ---------------------------------------------------------------------------
-- External completion
-- ---------------------------------------------------------------------------

-- | Resolve an awakeable from outside the workflow: store @result@ in the
-- @keiro_awakeables@ row /and/ append a @StepRecorded@ to the owning workflow's
-- journal so the next run replays past the @await@.
--
-- Idempotent and crash-safe:
--
-- * Returns 'True' only when /this/ call transitioned the row @pending@ ->
--   @completed@; a second signal (or a signal of a @cancelled@ row) returns
--   'False' and leaves the stored payload unchanged.
-- * For a @pending@ row, the row transition and journal append happen in one
--   transaction. For an already-@completed@ row, the journal entry is re-appended
--   from the stored payload to repair rows wedged before that atomic path existed.
--   The append path is idempotent (deterministic event id plus step-index check),
--   so a re-append collapses to a no-op once the entry is present.
-- * If a cancellation wins after this function's initial row read but before its
--   guarded completion, the transaction re-reads the status and appends nothing.
--   The signal returns 'False', so cancellation cannot both trigger compensation
--   and leak a completion value into the workflow journal.
--
-- A 'False' return therefore does not mean "nothing happened": the journal may
-- still have been repaired. Returns 'False' for an unknown id.
signalAwakeable :: (IOE :> es, Store :> es, ToJSON r) => AwakeableId -> r -> Eff es Bool
signalAwakeable aid result =
  lookupAwakeable (awakeableIdToUuid aid) >>= \case
    Nothing -> pure False
    Just row -> signalAwakeableFrom row result

-- | The transaction-decision core of 'signalAwakeable', exposed so race
-- contracts can deterministically interpose between the initial row read and the
-- guarded completion. Normal callers should use 'signalAwakeable'.
--
-- The supplied row may be stale. This function therefore trusts it only for the
-- owner coordinates and candidate payload; when a pending-to-completed UPDATE
-- loses, it re-reads status inside the same transaction and appends only if
-- another signal completed the row. A winning cancellation gets no append.
signalAwakeableFrom ::
  (IOE :> es, Store :> es, ToJSON r) =>
  AwakeableRow ->
  r ->
  Eff es Bool
signalAwakeableFrom row result
  | row ^. #status == Cancelled = pure False
  | otherwise = do
      now <- liftIO getCurrentTime
      let aid = AwakeableId (row ^. #awakeableId)
          payload =
            if row ^. #status == Completed
              then row ^. #payload
              else Just (toJSON result)
      case payload of
        Nothing -> pure False
        Just payloadValue -> do
          let ownerName = WorkflowName (row ^. #ownerWorkflowName)
              ownerId = WorkflowId (row ^. #ownerWorkflowId)
          gen <- currentGeneration ownerName ownerId
          appendTx <-
            prepareJournalAppend
              ownerName
              ownerId
              gen
              StepRecorded
                { stepName = awakeableStepPrefix <> awakeableIdText aid,
                  result = payloadValue,
                  recordedAt = now
                }
          (transitioned, appendOutcome) <-
            runTransaction $ do
              transitioned <-
                if row ^. #status == Pending
                  then completeAwakeableTx (awakeableIdToUuid aid) (toJSON result) now
                  else pure False
              if transitioned || row ^. #status == Completed
                then do
                  outcome <- appendTx
                  condemnOnAppendConflict outcome
                  pure (transitioned, Just outcome)
                else
                  lookupAwakeableStatusTx (awakeableIdToUuid aid) >>= \case
                    Just Completed -> do
                      outcome <- appendTx
                      condemnOnAppendConflict outcome
                      pure (False, Just outcome)
                    _ -> pure (False, Nothing)
          for_ appendOutcome throwOnAppendConflict
          pure transitioned

-- A refusal is deliberately not condemned. The owning workflow is terminal, so
-- the journal entry must not land — but the promise itself is still resolved
-- durably, and rolling the row transition back would leave the awakeable
-- pending forever with no one left to signal it.
condemnOnAppendConflict :: JournalAppendOutcome -> Tx.Transaction ()
condemnOnAppendConflict = \case
  JournalAppendConflict {} -> Tx.condemn
  JournalRefusedTerminal {} -> pure ()
  _ -> pure ()

throwOnAppendConflict :: JournalAppendOutcome -> Eff es ()
throwOnAppendConflict = \case
  JournalAppendConflict err -> throwIO (WorkflowJournalAppendError (Text.pack (show err)))
  -- Not an error: signalling a terminal workflow's promise is a no-op delivery.
  JournalRefusedTerminal {} -> pure ()
  _ -> pure ()

-- | Abandon a still-@pending@ awakeable: flips its row to @cancelled@ and
-- writes __no__ journal entry (there is no result value to record). Returns 'True'
-- when it transitioned a @pending@ row, 'False' otherwise (already completed,
-- already cancelled, or unknown). A workflow that later re-enters the awakeable's
-- @await@ then throws 'WorkflowAwakeableCancelled'.
--
-- Cancellation and a concurrent stale suspend write are serialized under the
-- same generation-scoped per-step advisory lock that protects wake delivery.
-- If cancellation commits first, the suspend re-check observes the terminal
-- awakeable row; if suspension commits first, cancellation's instance upsert
-- restores @running@. Either order leaves the workflow discoverable.
--
-- Because there is no journal append, the same transaction flips the /owning
-- workflow's/ @keiro_workflows@ row to @running@ — the status a wake delivery
-- would have written. Cancellation is a wake-source lifecycle transition like
-- any other, and the instance row is what the resume worker discovers: without
-- the flip, a workflow parked on a cancelled promise would be invisible to
-- discovery and could never reach its @await@ arm to throw
-- 'WorkflowAwakeableCancelled'. @running@ here means "this workflow has progress
-- to make" — here, observing the cancellation. The upsert takes @GREATEST@ of the
-- stored and supplied generation, so passing 0 preserves whatever generation the
-- instance is on, and it never revives a terminal instance.
cancelAwakeable :: (Store :> es) => AwakeableId -> Eff es Bool
cancelAwakeable aid =
  lookupAwakeable (awakeableIdToUuid aid) >>= \case
    Nothing -> pure False
    Just row -> do
      let ownerName = WorkflowName (row ^. #ownerWorkflowName)
          ownerId = WorkflowId (row ^. #ownerWorkflowId)
      gen <- currentGeneration ownerName ownerId
      runTransaction $ do
        lockWorkflowStepTx
          ( workflowStepLockKey
              (unWorkflowId ownerId)
              (unWorkflowName ownerName)
              gen
              (awakeableStepPrefix <> awakeableIdText aid)
          )
        cancelAwakeableTx (awakeableIdToUuid aid) >>= \case
          Nothing -> pure False
          Just (ownerNameText, ownerIdText) -> do
            upsertInstanceTx ownerIdText ownerNameText 0 WfRunning Nothing
            pure True