-- SPDX-FileCopyrightText: 2026 Alexandra de Wit
--
-- SPDX-License-Identifier: MIT

{- | Loop robustness and supervision for the worker.

The loop cannot be killed by a single bad iteration. A failed @receive@ arrives
as the queue handle's typed fault value: the step logs it and backs off (its own
fixed pacing over the typed channel). Residue -- an exception escaping a
dependency's typed contract mid-iteration -- is the supervision combinator's
concern ('Ecluse.Core.Supervision.superviseLoop' wraps the step under the
caller-supplied policy), classified per that policy: transient residue is logged
and retried with bounded exponential backoff, while a wiring fault the policy
names 'Ecluse.Core.Supervision.Permanent' fails up through the composition
root's race and takes the process down (fail-stop). Each successful poll and each
completed job advances the 'WorkerHeartbeat', so a stalled loop is visible to the
liveness probe.

Shutdown tears the loop down cleanly: the composition root runs it raced against
the server within its resource bracket, so process teardown cancels the loop
thread (the combinator never catches cancellation) and an in-flight, un-acked
message simply redelivers -- safe, because publishing is idempotent (a version
already present is success).
-}
module Ecluse.Core.Worker.Loop (
    workerLoop,
) where

import Katip (Severity (DebugS, ErrorS), logFM, ls)
import UnliftIO.Concurrent (threadDelay)

import Ecluse.Core.Queue (MirrorQueue (receive), qfDetail)
import Ecluse.Core.Supervision (SupervisionPolicy, superviseLoop)
import Ecluse.Core.Worker.Job (processBatch)
import Ecluse.Core.Worker.Types

{- | The continuous consume loop: long-poll for a batch, process it, repeat,
supervised under the given policy (the composition root names the wiring faults
that must fail up rather than retry; tests inject their own).

A failed poll arrives as the handle's typed 'Ecluse.Core.Queue.QueueFault' value:
it is logged and the step backs off and polls again, so a queue outage cannot
kill the worker thread. A successful poll advances the heartbeat (whether or not
the batch was empty), and 'processBatch' advances it again after each completed
job, so a liveness probe sees the loop is alive even while a healthy worker grinds
through a long batch of large artifacts; an idle queue is a healthy empty poll, not
a stall. The heartbeat advances only on demonstrated progress (a successful
@receive@ or a completed job), so a worker that cannot poll at all (a persistently
faulting @receive@) keeps retrying but never advances it: the heartbeat goes stale
and @\/livez@ fails, surfacing a fully-dead worker for the orchestrator to restart.
-}
workerLoop :: SupervisionPolicy -> WorkerM Void
workerLoop :: SupervisionPolicy -> WorkerM Void
workerLoop SupervisionPolicy
policy = SupervisionPolicy -> WorkerM () -> WorkerM Void
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
SupervisionPolicy -> m () -> m Void
superviseLoop SupervisionPolicy
policy WorkerM ()
pollAndProcess
  where
    pollAndProcess :: WorkerM ()
    pollAndProcess :: WorkerM ()
pollAndProcess = do
        queue <- (WorkerRuntime -> MirrorQueue) -> WorkerM MirrorQueue
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks WorkerRuntime -> MirrorQueue
wrQueue
        liftIO (receive queue) >>= \case
            Left QueueFault
fault -> do
                -- A failed poll: no heartbeat advance (the loop is retrying, not
                -- healthy-idle), log the typed fault, and back off before the next
                -- poll so a dead backend is retried at a bounded rate. This is the
                -- step's own pacing over the typed channel; the supervisor's
                -- exponential backoff paces only residue.
                Severity -> LogStr -> WorkerM ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
ErrorS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"worker receive failed, backing off: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> QueueFault -> Text
qfDetail QueueFault
fault))
                WorkerM ()
backoff
            Right [QueueMessage]
messages -> do
                case [QueueMessage]
messages of
                    [] -> WorkerM ()
forall (f :: * -> *). Applicative f => f ()
pass
                    [QueueMessage]
_ -> Severity -> LogStr -> WorkerM ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
DebugS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"worker received " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show ([QueueMessage] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [QueueMessage]
messages) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" messages" :: Text))
                -- Beat on every successful poll -- an empty long-poll is a healthy idle.
                -- 'processBatch' beats again after each job, so a long batch cannot starve it.
                WorkerM ()
recordWorkerProgress
                [QueueMessage] -> WorkerM ()
processBatch [QueueMessage]
messages

-- The fixed pause after a faulted poll, so a persistently failing queue backend
-- is retried at a bounded rate rather than hot-looping.
backoff :: WorkerM ()
backoff :: WorkerM ()
backoff = Int -> WorkerM ()
forall (m :: * -> *). MonadIO m => Int -> m ()
threadDelay Int
1_000_000