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,
    deterministicAwakeableId,

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

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

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

import Control.Exception (Exception)
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 Data.UUID.V5 qualified as UUID.V5
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.Schema
  ( AwakeableRow,
    AwakeableStatus (..),
    cancelAwakeableTx,
    completeAwakeableTx,
    lookupAwakeable,
    lookupAwakeableStatusTx,
    registerAwakeableTx,
  )
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'; 'deterministicAwakeableId' is retained only as a legacy
-- generation-0 adoption helper. 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

-- | The legacy deterministic 'AwakeableId' for a @(workflow name, workflow id,
-- label)@: a v5 UUID over @(\"keiro\":\"awakeable\":name:id:label)@.
--
-- This is predictable from public coordinates, so new code must not hand-derive
-- ids with it. It remains exported for operators and for generation-0 adoption:
-- if a pre-change workflow already registered a row under this id, the first
-- post-change allocation adopts that row so the in-flight promise keeps working.
deterministicAwakeableId :: WorkflowName -> WorkflowId -> Text -> AwakeableId
deterministicAwakeableId (WorkflowName name) (WorkflowId wid) label =
  AwakeableId $
    UUID.V5.generateNamed UUID.V5.namespaceURL $
      fmap (fromIntegral . fromEnum) $
        Text.unpack $
          Text.intercalate ":" ["keiro", "awakeable", name, wid, label]

-- ---------------------------------------------------------------------------
-- 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.
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 = do
      let legacy = deterministicAwakeableId name wid label
      existing <- lookupAwakeable (awakeableIdToUuid legacy)
      case existing of
        Just _ -> pure legacy
        Nothing -> AwakeableId <$> liftIO UUID.V4.nextRandom
  | otherwise = AwakeableId <$> liftIO UUID.V4.nextRandom

-- | 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

condemnOnAppendConflict :: JournalAppendOutcome -> Tx.Transaction ()
condemnOnAppendConflict = \case
  JournalAppendConflict {} -> Tx.condemn
  _ -> pure ()

throwOnAppendConflict :: JournalAppendOutcome -> Eff es ()
throwOnAppendConflict = \case
  JournalAppendConflict err -> throwIO (WorkflowJournalAppendError (Text.pack (show err)))
  _ -> 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'.
cancelAwakeable :: (Store :> es) => AwakeableId -> Eff es Bool
cancelAwakeable aid =
  runTransaction $ cancelAwakeableTx (awakeableIdToUuid aid)