Child.hs

-- | Child workflows: spawn, wait on, and cancel a workflow from inside another.
--
-- == What this gives you
--
-- Fan-out\/fan-in composition of durable workflows. A /parent/ workflow spawns a
-- /child/ (a second workflow with its own journal stream), waits for the child's
-- result, and may cancel a child it no longer needs — with the whole relationship
-- surviving a crash, because the spawn is recorded in the parent's journal and the
-- link is stored in @keiro_workflow_children@.
--
-- @
-- parent :: ('Workflow' ':>' es, 'Store' ':>' es, 'IOE' ':>' es) => Eff es Text
-- parent = do
--   h      <- 'spawnChild' (WorkflowName \"ship-order\") (WorkflowId \"ord-7\") shipWorkflow
--   result <- 'awaitChild' h                  -- SUSPEND until the child completes
--   _      <- 'Keiro.Workflow.step' (StepName \"notify\") (pure (\"shipped: \" <> result))
--   pure result
-- @
--
-- == How it works (contract recap for downstream plans)
--
-- * __'spawnChild' is a journaled step.__ It records a @StepRecorded
--   \"child:\<childId\>\"@ in the /parent/ journal (so a replay short-circuits the
--   spawn and never re-spawns) and inserts a @running@ row in
--   @keiro_workflow_children@ linking the child back to the parent plus the
--   parent-journal step the parent awaits (@child:\<childId\>:result@). It does
--   __not__ run the child inline — the EP-42 resume worker drives the child to
--   completion from the application's @WorkflowRegistry@, so __a child's
--   'WorkflowName' must be registered there__, exactly like any resumable
--   workflow. (The @childDef@ argument is taken for authoring ergonomics; the
--   registry, keyed by 'WorkflowName', is the source of truth for the child's
--   body.)
-- * __'awaitChild' reuses EP-38's suspension primitive__ — it is
--   @'awaitStep' (StepName \"child:\<childId\>:result\") arm@. On the miss path it
--   re-delivers a completed child's stored result onto the current parent
--   generation (attach semantics), throws 'WorkflowChildCancelled' if the child
--   was cancelled meanwhile, throws 'WorkflowChildFailed' from the durable child
--   row if it failed, and otherwise re-asserts nothing new (the spawn already
--   registered the link). On the hit path
--   it decodes a tagged parent-journal envelope: @{"ok": result}@ returns the
--   child result, @{"cancelled": true}@ throws 'WorkflowChildCancelled',
--   @{"failed": reason}@ throws 'WorkflowChildFailed', and legacy raw values are
--   still decoded as pre-envelope successful results.
-- * __'runChildWorkflow'__ is what the resume worker selects (instead of bare
--   'Keiro.Workflow.runWorkflowWith') for any workflow that is some parent's child:
--   on 'Completed' it runs 'childCompletionHook', which flips the child row to
--   @completed@ and appends @{"ok": result}@ to the parent's
--   @child:\<childId\>:result@ step. If it finds a historically
--   @cancelled@-but-unmarked row, it ensures the child cancellation marker and
--   parent sentinel before returning 'Cancelled'.
-- * __'cancelChild'__ flips the child row to @cancelled@ and, in the same
--   transaction, writes a 'Keiro.Workflow.WorkflowCancelled' marker to the
--   /child/ journal plus a @{"cancelled": true}@ sentinel as the parent's
--   await-step result. Retrying after a historical row-only cancel repairs both
--   markers even though the return value remains 'False' ("this call did not
--   transition the row").
-- * @'Keiro.Workflow.Child.Schema.countActiveChildren'@ backs a potential
--   @keiro.workflow.children.active@ gauge for EP-44.
module Keiro.Workflow.Child
  ( -- * Child handles and reserved step names
    ChildHandle (..),
    childSpawnStepName,
    childResultStepName,

    -- * Authoring surface (inside a parent workflow)
    spawnChild,
    awaitChild,

    -- * External control
    cancelChild,

    -- * Driving a child to completion (used by the resume worker)
    runChildWorkflow,
    childCompletionHook,

    -- * Errors
    WorkflowChildCancelled (..),
    WorkflowChildFailed (..),
  )
where

import Control.Exception (Exception)
import Data.Aeson qualified as Aeson
import Data.Aeson.KeyMap qualified as KeyMap
import Data.Text qualified as Text
import Effectful (Eff, IOE, (:>))
import Effectful.Error.Static (Error)
import Effectful.Exception (throwIO)
import Keiro.Prelude
import Keiro.Workflow
  ( JournalAppendOutcome (..),
    StepName (..),
    Workflow,
    WorkflowError (..),
    WorkflowId (..),
    WorkflowJournalEvent (..),
    WorkflowName (..),
    WorkflowOutcome (..),
    WorkflowRunOptions,
    appendJournalEntry,
    awaitStep,
    childStepPrefix,
    currentGeneration,
    currentWorkflow,
    prepareJournalAppend,
    runWorkflowWith,
    step,
  )
import Keiro.Workflow.Child.Schema
  ( ChildRow,
    ChildStatus (..),
    lookupChild,
    markChildCancelledTx,
    markChildResultTx,
    registerChildTx,
  )
import Keiro.Workflow.Instance (WorkflowStatus (..), lookupInstance, upsertInstanceTx)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Error (StoreError)
import Kiroku.Store.Transaction (runTransaction)
import "hasql-transaction" Hasql.Transaction qualified as Tx

-- ---------------------------------------------------------------------------
-- Handles and reserved step names
-- ---------------------------------------------------------------------------

-- | An in-memory handle to a spawned child, returned by 'spawnChild' and taken
-- by 'awaitChild' \/ 'cancelChild'. The phantom type parameter @a@ records the
-- child's result type so 'awaitChild' decodes without an extra annotation; the
-- handle itself carries only the child's name and id (the child journal is
-- @wf:\<childName\>-\<childWfId\>@).
data ChildHandle a = ChildHandle
  { childName :: !WorkflowName,
    childWfId :: !WorkflowId
  }
  deriving stock (Eq, Show)

-- | The reserved /spawn/ step name a child is recorded under in its parent's
-- journal: @child:\<childId\>@ (uses EP-38's 'childStepPrefix').
childSpawnStepName :: WorkflowId -> Text
childSpawnStepName (WorkflowId wid) = childStepPrefix <> wid

-- | The reserved /result/ step name the parent awaits and the child's
-- completion is propagated under: @child:\<childId\>:result@.
childResultStepName :: WorkflowId -> Text
childResultStepName (WorkflowId wid) = childStepPrefix <> wid <> ":result"

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

-- | Thrown out of 'awaitChild' when the awaited child was 'cancelChild'led. A
-- cancelled child never produces a result, so suspending forever would be wrong
-- and fabricating a result would be wrong; the parent author can @catch@ this to
-- run compensation. If uncaught, the resume worker records the parent attempt,
-- backs it off, and eventually marks the parent failed at its configured ceiling.
-- Mirrors EP-40's 'Keiro.Workflow.Awakeable.WorkflowAwakeableCancelled'.
data WorkflowChildCancelled = WorkflowChildCancelled WorkflowName WorkflowId
  deriving stock (Eq, Show)

instance Exception WorkflowChildCancelled

-- | Thrown out of 'awaitChild' when the child was terminally failed by the
-- resume worker. Carries the child's identity plus the persisted failure reason.
data WorkflowChildFailed = WorkflowChildFailed WorkflowName WorkflowId Text
  deriving stock (Eq, Show)

instance Exception WorkflowChildFailed

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

-- | Spawn a child workflow. Records a @StepRecorded \"child:\<childId\>\"@ in
-- the /parent/ journal (so a replay short-circuits and never re-spawns) and
-- inserts an idempotent @running@ link row, then returns a 'ChildHandle'. The
-- child is driven to completion by the resume worker from the registry, so
-- @childNm@ __must be registered there__; @childDef@ is accepted for authoring
-- ergonomics but is not run here.
--
-- A child id names one execution globally. Spawning an id whose child row already
-- completed attaches to that execution: 'awaitChild' re-delivers the stored
-- result onto the parent's current generation. To run a fresh child after
-- 'Keiro.Workflow.continueAsNew', derive a fresh child id from the carried seed.
spawnChild ::
  (Workflow :> es, Store :> es) =>
  -- | The child's name (must be in the resume worker's registry).
  WorkflowName ->
  -- | The child's id (names the child journal).
  WorkflowId ->
  -- | The child workflow definition (for authoring; the registry actually runs it).
  Eff (Workflow : es) a ->
  Eff es (ChildHandle a)
spawnChild childNm childWid _childDef = do
  (parentNm, parentWid) <- currentWorkflow
  let spawnStep = StepName (childSpawnStepName childWid)
      resultStep = childResultStepName childWid
  -- Journaled as a step in the PARENT journal: a parent replay short-circuits
  -- this body, and the ON CONFLICT DO NOTHING register collapses on re-run.
  _ <-
    step spawnStep $ do
      runTransaction $
        registerChildTx
          (unWorkflowId childWid)
          (unWorkflowName childNm)
          (unWorkflowId parentWid)
          (unWorkflowName parentNm)
          resultStep
          *> upsertInstanceTx
            (unWorkflowId childWid)
            (unWorkflowName childNm)
            0
            WfRunning
            Nothing
      pure ()
  pure (ChildHandle childNm childWid)

-- | Suspend the parent until the child completes, then return the child's
-- result. This is EP-38's 'awaitStep' on the @child:\<childId\>:result@ step:
-- 'childCompletionHook' journals that step into the parent when the child
-- finishes, so the next parent run replays past the wait and decodes the result.
--
-- If the child was cancelled or failed, throws 'WorkflowChildCancelled' or
-- 'WorkflowChildFailed'. Parent-journal values are tagged envelopes
-- (@{"ok": ...}@, @{"cancelled": true}@, @{"failed": reason}@) with a legacy raw
-- success fallback; a decode mismatch throws 'WorkflowStepDecodeError'.
awaitChild ::
  (Workflow :> es, Store :> es, IOE :> es, FromJSON a) =>
  ChildHandle a ->
  Eff es a
awaitChild (ChildHandle childNm childWid) = do
  let resultStep = StepName (childResultStepName childWid)
      arm = do
        mrow <- lookupChild (unWorkflowId childWid) (unWorkflowName childNm)
        case mrow of
          Just row
            | (row ^. #status) == ChildCancelled ->
                throwIO (WorkflowChildCancelled childNm childWid)
            | (row ^. #status) == ChildFailed -> do
                mInstance <- lookupInstance childNm childWid
                let reason =
                      case row ^. #failureReason of
                        Just recorded -> recorded
                        Nothing ->
                          fromMaybe
                            "child workflow failed (reason not recorded)"
                            (mInstance >>= (^. #lastError))
                throwIO (WorkflowChildFailed childNm childWid reason)
            | (row ^. #status) == ChildCompleted,
              Just resultValue <- row ^. #result ->
                appendJournalEntry
                  (WorkflowName (row ^. #parentName))
                  (WorkflowId (row ^. #parentId))
                  StepRecorded
                    { stepName = row ^. #awaitStep,
                      result = childOkEnvelope resultValue,
                      recordedAt = fromMaybe (row ^. #updatedAt) (row ^. #completedAt)
                    }
          -- The spawn already registered the link; nothing more to (re-)arm.
          _ -> pure ()
  raw <- awaitStep resultStep arm
  decodeChildResult childNm childWid (unStepName resultStep) raw

-- ---------------------------------------------------------------------------
-- External control
-- ---------------------------------------------------------------------------

-- | Cancel a child. For a @running@ child, flips its
-- @keiro_workflow_children@ row to @cancelled@ and, in the same transaction,
-- writes both the child's 'WorkflowCancelled' marker and the parent's
-- @{"cancelled": true}@ await-step sentinel. Returns 'True' only when this call
-- performed the row transition. For an already-@cancelled@ row, returns 'False'
-- but still ensures both markers, repairing historical crashes that committed the
-- row flip before either journal append. Already completed/failed/unknown
-- children return 'False' and no marker is fabricated.
cancelChild ::
  (IOE :> es, Store :> es) =>
  ChildHandle a ->
  Eff es Bool
cancelChild (ChildHandle childNm childWid) = do
  mrow <- lookupChild (unWorkflowId childWid) (unWorkflowName childNm)
  case mrow of
    Nothing -> pure False
    Just row -> do
      (transitioned, childOutcome, parentOutcome) <- ensureChildCancelled row
      throwOnAppendConflict childOutcome
      throwOnAppendConflict parentOutcome
      pure transitioned

-- ---------------------------------------------------------------------------
-- Completion propagation
-- ---------------------------------------------------------------------------

-- | Drive a child workflow to completion, propagating its result to the parent.
-- This is 'runWorkflowWith' followed by 'childCompletionHook' on 'Completed': the
-- resume worker selects this (instead of bare 'runWorkflowWith') for any workflow
-- that is some parent's child, so a finished child wakes its waiting parent. A
-- 'Suspended' child propagates nothing. A child row already marked 'ChildCancelled'
-- is repaired by ensuring the child cancellation marker and parent sentinel, then
-- returns 'Cancelled'; a 'ChildFailed' row returns 'Failed'.
runChildWorkflow ::
  (IOE :> es, Store :> es, Error StoreError :> es, ToJSON a) =>
  WorkflowRunOptions ->
  WorkflowName ->
  WorkflowId ->
  Eff (Workflow : es) a ->
  Eff es (WorkflowOutcome a)
runChildWorkflow opts childNm childWid action = do
  mrow <- lookupChild (unWorkflowId childWid) (unWorkflowName childNm)
  case fmap (^. #status) mrow of
    Just ChildCancelled -> do
      for_ mrow $ \row -> do
        (_, childOutcome, parentOutcome) <- ensureChildCancelled row
        throwOnAppendConflict childOutcome
        throwOnAppendConflict parentOutcome
      pure Cancelled
    Just ChildFailed -> pure Failed
    _ -> do
      outcome <- runWorkflowWith opts childNm childWid action
      case outcome of
        Completed result -> do
          childCompletionHook childNm childWid (toJSON result)
          pure (Completed result)
        other -> pure other

-- | Propagate a finished child's result to its parent: flip the child row to
-- @completed@ (storing the raw result in the child row) and append an
-- @{"ok": result}@ @child:\<childId\>:result@ 'StepRecorded' to the /parent/
-- journal — exactly the wake source the parent's 'awaitChild' resolves on. The
-- running-row transition and parent append happen in one transaction. An
-- already-@completed@ row re-appends from the stored result, repairing historical
-- wedges; cancelled/failed children do not fabricate a success. Normally invoked
-- via 'runChildWorkflow'.
childCompletionHook ::
  (IOE :> es, Store :> es) =>
  WorkflowName ->
  WorkflowId ->
  Aeson.Value ->
  Eff es ()
childCompletionHook childNm childWid resultValue = do
  mrow <- lookupChild (unWorkflowId childWid) (unWorkflowName childNm)
  for_ mrow $ \row -> case row ^. #status of
    Running -> do
      now <- liftIO getCurrentTime
      let parentName = WorkflowName (row ^. #parentName)
          parentId = WorkflowId (row ^. #parentId)
      gen <- currentGeneration parentName parentId
      appendTx <-
        prepareJournalAppend
          parentName
          parentId
          gen
          StepRecorded
            { stepName = row ^. #awaitStep,
              result = childOkEnvelope resultValue,
              recordedAt = now
            }
      appendOutcome <-
        runTransaction $ do
          transitioned <- markChildResultTx (unWorkflowId childWid) (unWorkflowName childNm) resultValue now
          if transitioned
            then do
              appendOutcome <- appendTx
              condemnOnAppendConflict appendOutcome
              pure appendOutcome
            else pure (JournalAlreadyPresent resultValue)
      throwOnAppendConflict appendOutcome
    ChildCompleted ->
      for_ (row ^. #result) $ \stored ->
        appendJournalEntry
          (WorkflowName (row ^. #parentName))
          (WorkflowId (row ^. #parentId))
          StepRecorded
            { stepName = row ^. #awaitStep,
              result = childOkEnvelope stored,
              recordedAt = fromMaybe (row ^. #updatedAt) (row ^. #completedAt)
            }
    ChildCancelled -> pure ()
    ChildFailed -> pure ()

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

ensureChildCancelled ::
  (IOE :> es, Store :> es) =>
  ChildRow ->
  Eff es (Bool, JournalAppendOutcome, JournalAppendOutcome)
ensureChildCancelled row = do
  now <- liftIO getCurrentTime
  let childNm = WorkflowName (row ^. #childName)
      childWid = WorkflowId (row ^. #childId)
      parentNm = WorkflowName (row ^. #parentName)
      parentWid = WorkflowId (row ^. #parentId)
  childGen <- currentGeneration childNm childWid
  parentGen <- currentGeneration parentNm parentWid
  childAppendTx <- prepareJournalAppend childNm childWid childGen WorkflowCancelled {recordedAt = now}
  parentAppendTx <-
    prepareJournalAppend
      parentNm
      parentWid
      parentGen
      StepRecorded
        { stepName = row ^. #awaitStep,
          result = cancelledSentinel,
          recordedAt = now
        }
  runTransaction $ do
    transitioned <-
      if row ^. #status == Running
        then markChildCancelledTx (row ^. #childId) (row ^. #childName)
        else pure False
    if transitioned || row ^. #status == ChildCancelled
      then do
        childOutcome <- childAppendTx
        parentOutcome <- parentAppendTx
        condemnOnAppendConflict childOutcome
        condemnOnAppendConflict parentOutcome
        pure (transitioned, childOutcome, parentOutcome)
      else pure (False, JournalAlreadyPresent Aeson.Null, JournalAlreadyPresent Aeson.Null)

-- ---------------------------------------------------------------------------
-- The cancellation sentinel
-- ---------------------------------------------------------------------------

-- | The @{"cancelled": true}@ value 'cancelChild' writes as a cancelled child's
-- await-step result so 'awaitChild' can detect it and throw.
cancelledSentinel :: Aeson.Value
cancelledSentinel = Aeson.object ["cancelled" Aeson..= True]

childOkEnvelope :: Aeson.Value -> Aeson.Value
childOkEnvelope value = Aeson.object ["ok" Aeson..= value]

decodeChildResult ::
  (FromJSON a) =>
  WorkflowName ->
  WorkflowId ->
  Text ->
  Aeson.Value ->
  Eff es a
decodeChildResult childNm childWid key raw =
  case raw of
    Aeson.Object obj
      | Just okValue <- KeyMap.lookup "ok" obj ->
          decodeOrThrow okValue
      | Just (Aeson.Bool True) <- KeyMap.lookup "cancelled" obj ->
          throwIO (WorkflowChildCancelled childNm childWid)
      | Just (Aeson.String reason) <- KeyMap.lookup "failed" obj ->
          throwIO (WorkflowChildFailed childNm childWid reason)
    _ -> decodeOrThrow raw
  where
    decodeOrThrow value =
      case Aeson.fromJSON value of
        Aeson.Success a -> pure a
        Aeson.Error e -> throwIO (WorkflowStepDecodeError key (Text.pack e))