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

{- | The mirror-queue handle: the durable hand-off from the request path to the
mirror worker.

Mirroring is __demand-driven__: when a client fetches an artifact whose version
passes the rules, the proxy 'enqueue's a 'MirrorJob' and serves the artifact
immediately, never blocking on the mirror. A separate worker 'receive's jobs,
fetches and verifies the artifact, publishes it to the mirror target, and 'ack's
the job (see @docs\/architecture\/cloud-backends.md@ → "Mirror Queue").

The queue is the one cloud surface with materially different APIs per provider
(AWS SQS @SendMessage@\/@ReceiveMessage@+visibility-timeout\/@DeleteMessage@; GCP
Pub\/Sub @Publish@\/@Pull@+ack-deadline\/@Acknowledge@), so it is its own handle --
a __record of functions__ (the Handle pattern). Both providers fit the same
receive → process → ack shape; their differences (visibility timeout vs ack
deadline, batch limits, dead-letter wiring) stay behind the handle, and
'ReceiptHandle' is opaque so neither leaks.

Like the other handles, the effectful fields return __'IO', not @App@__, so an
adapter stays decoupled from the proxy's @Env@\/@App@ (see
@docs\/architecture\/technology-stack.md@ → "Key Decisions").

== Conventions

The two cloud backends both give __at-least-once delivery__, which is safe here
because publishing is idempotent (a registry treats versions as immutable). The
handle's contract reflects that:

* __'enqueue' is best-effort.__ It runs on the request hot path (enqueue, then
  serve immediately), so a failure must be logged\/metered and __never fail the
  client response__ -- the artifact is already served, and a later pull
  re-enqueues.
* __Retry is "don't 'ack'".__ A job that fails processing __transiently__ (a flaky
  fetch, a registry blip) is simply not acked; the visibility timeout \/ ack deadline
  redelivers it, and it may succeed next time. There is deliberately __no @nack@__ for
  the transient case.
* __'deadLetter' is the terminal terminus.__ A job that can __never__ succeed (an
  artifact past the plan-sized byte cap) is a terminal verdict, and each backend
  realises it its own way: the in-memory backend drops the delivery (its only
  terminus), the SQS backend returns the message with a backoff visibility timeout
  __without deleting it__, so it rides the operator's redrive policy to the
  dead-letter queue for forensic retention rather than being silently discarded. This
  is not a @nack@ (a retry) and not an 'ack' (a clean retire); it is the third,
  terminal outcome.
* __'extendVisibility'__ lets the worker hold a long publish (a large artifact)
  past the visibility window. It is an /optimization/, not correctness-critical,
  since idempotency already makes redelivery harmless.

This module provides the handle, its payload types, and the building blocks a
backend implementation reaches for; the STM-backed bounded, best-effort
__production backend__ mirroring rolls over to when no @ECLUSE_QUEUE__URL@ is set
lives in "Ecluse.Core.Queue.Memory".

It also provides 'newEnqueueBuffer', a __bounded producer-side hand-off buffer__
wrapped in front of any backend so the serve path's 'enqueue' completes in
microseconds while a composition-root drain loop delivers to the (possibly slow)
backend off the request path.
-}
module Ecluse.Core.Queue (
    -- * Queue handle
    MirrorQueue (..),
    noMirrorQueue,

    -- * Faults
    QueueFault (..),
    queueTransportFault,

    -- * Payloads
    MirrorJob (..),
    RemoteSpanContext (..),
    QueueMessage (..),

    -- * Opaque receipt
    ReceiptHandle,
    mkReceiptHandle,
    unReceiptHandle,

    -- * Durations
    Seconds (..),

    -- * Backend building blocks
    writeOrDrop,
    reportWorthy,

    -- * Buffered producer hand-off
    newEnqueueBuffer,
) where

import Control.Concurrent.STM.TBQueue (TBQueue, isFullTBQueue, newTBQueueIO, readTBQueue, writeTBQueue)
import UnliftIO.Concurrent (threadDelay)
import UnliftIO.Exception (tryAny)

import Ecluse.Core.Fault (TransportCause (TransportProtocol), TransportFault (TransportFault), transportFault)
import Ecluse.Core.Package (PackageName)
import Ecluse.Core.Security.Egress (RegistryUrl)
import Ecluse.Core.Supervision (BackoffSchedule (BackoffSchedule, bsBaseMicros, bsCapMicros), backoffMicros)
import Ecluse.Core.Version (Version)

{- | A mirror job: everything the worker needs to back-fill one artifact into the
mirror target. The version was gated by the rules at serve time (when the job was
enqueued); the worker __re-evaluates current policy__ through the same shared
admission oracle before mirroring (see "Ecluse.Core.Worker.Job"), then fetches the
bytes, verifies them against the digests of the artifact that re-evaluation
re-admitted, and publishes.

The queue payload is a trust boundary, so it carries __selection keys, never
authority__: the filename ('jobArtifactFilename') names the artifact the worker's
ingest re-evaluation selects and gates under current policy, and the payload
carries no digest or size at all -- the descriptor the tamper gate and the publish
document consume ('Ecluse.Core.Registry.MirrorArtifact') is derived entirely from
the artifact that re-evaluation re-admits.
-}
data MirrorJob = MirrorJob
    { MirrorJob -> PackageName
jobPackage :: PackageName
    -- ^ The package whose artifact is being mirrored.
    , MirrorJob -> Version
jobVersion :: Version
    -- ^ The specific version to mirror.
    , MirrorJob -> RegistryUrl
jobArtifactUrl :: RegistryUrl
    {- ^ Where to fetch the artifact bytes from (the public upstream), carried as
    the validated https egress witness rather than bare text; the SQS wire decode
    re-forms it, since the queue payload is a trust boundary.
    -}
    , MirrorJob -> Text
jobArtifactFilename :: Text
    {- ^ The serve-time-admitted artifact's filename: the selection key the
    worker's ingest re-evaluation gates by, cross-checked against current metadata
    by the shared admission gate rather than trusted.
    -}
    , MirrorJob -> Maybe RemoteSpanContext
jobTraceContext :: Maybe RemoteSpanContext
    {- ^ The trace context of the serve-time span that enqueued the job, captured
    at enqueue time so the worker's per-job span can __link__ back to the request
    that produced the work across the asynchronous hop. 'Nothing' when tracing was
    off at enqueue time (or for a job from a producer that carried none). The queue
    treats it as opaque transport; only the tracing port reads it.
    -}
    }
    deriving stock (MirrorJob -> MirrorJob -> Bool
(MirrorJob -> MirrorJob -> Bool)
-> (MirrorJob -> MirrorJob -> Bool) -> Eq MirrorJob
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: MirrorJob -> MirrorJob -> Bool
== :: MirrorJob -> MirrorJob -> Bool
$c/= :: MirrorJob -> MirrorJob -> Bool
/= :: MirrorJob -> MirrorJob -> Bool
Eq, Int -> MirrorJob -> ShowS
[MirrorJob] -> ShowS
MirrorJob -> String
(Int -> MirrorJob -> ShowS)
-> (MirrorJob -> String)
-> ([MirrorJob] -> ShowS)
-> Show MirrorJob
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MirrorJob -> ShowS
showsPrec :: Int -> MirrorJob -> ShowS
$cshow :: MirrorJob -> String
show :: MirrorJob -> String
$cshowList :: [MirrorJob] -> ShowS
showList :: [MirrorJob] -> ShowS
Show)

{- | A serialised W3C trace-context carrier riding on a 'MirrorJob': the
@traceparent@ (and any @tracestate@) of the span that enqueued the job, in the
standard wire encoding. It is captured at enqueue time and read back by the worker's
tracing port to re-establish a span __link__ from the per-job span to the enqueueing
request, so the asynchronous mirror hand-off is navigable in a trace.

The two fields are the W3C header values verbatim; the queue carries them opaquely
(it neither parses nor validates them -- an unparseable carrier simply yields no link),
so this type names what is carried without coupling the queue to any tracing backend.
-}
data RemoteSpanContext = RemoteSpanContext
    { RemoteSpanContext -> Text
rscTraceparent :: Text
    -- ^ The W3C @traceparent@ header value of the enqueueing span.
    , RemoteSpanContext -> Text
rscTracestate :: Text
    {- ^ The W3C @tracestate@ header value (possibly empty) carried alongside, so
    vendor trace state survives the hop.
    -}
    }
    deriving stock (RemoteSpanContext -> RemoteSpanContext -> Bool
(RemoteSpanContext -> RemoteSpanContext -> Bool)
-> (RemoteSpanContext -> RemoteSpanContext -> Bool)
-> Eq RemoteSpanContext
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: RemoteSpanContext -> RemoteSpanContext -> Bool
== :: RemoteSpanContext -> RemoteSpanContext -> Bool
$c/= :: RemoteSpanContext -> RemoteSpanContext -> Bool
/= :: RemoteSpanContext -> RemoteSpanContext -> Bool
Eq, Int -> RemoteSpanContext -> ShowS
[RemoteSpanContext] -> ShowS
RemoteSpanContext -> String
(Int -> RemoteSpanContext -> ShowS)
-> (RemoteSpanContext -> String)
-> ([RemoteSpanContext] -> ShowS)
-> Show RemoteSpanContext
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> RemoteSpanContext -> ShowS
showsPrec :: Int -> RemoteSpanContext -> ShowS
$cshow :: RemoteSpanContext -> String
show :: RemoteSpanContext -> String
$cshowList :: [RemoteSpanContext] -> ShowS
showList :: [RemoteSpanContext] -> ShowS
Show)

{- | An __opaque__ handle identifying a received message for 'ack' \/
'extendVisibility'. It carries the backend's own delivery token -- an SQS receipt
handle or a Pub\/Sub @ackId@ -- as text; the constructor is hidden so neither
provider's representation leaks into worker code, and a handle is only ever
obtained from a 'QueueMessage' returned by 'receive'. Build one (in a backend)
with 'mkReceiptHandle' and read the token back with 'unReceiptHandle'.
-}
newtype ReceiptHandle = ReceiptHandle Text
    deriving stock (ReceiptHandle -> ReceiptHandle -> Bool
(ReceiptHandle -> ReceiptHandle -> Bool)
-> (ReceiptHandle -> ReceiptHandle -> Bool) -> Eq ReceiptHandle
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ReceiptHandle -> ReceiptHandle -> Bool
== :: ReceiptHandle -> ReceiptHandle -> Bool
$c/= :: ReceiptHandle -> ReceiptHandle -> Bool
/= :: ReceiptHandle -> ReceiptHandle -> Bool
Eq, Eq ReceiptHandle
Eq ReceiptHandle =>
(ReceiptHandle -> ReceiptHandle -> Ordering)
-> (ReceiptHandle -> ReceiptHandle -> Bool)
-> (ReceiptHandle -> ReceiptHandle -> Bool)
-> (ReceiptHandle -> ReceiptHandle -> Bool)
-> (ReceiptHandle -> ReceiptHandle -> Bool)
-> (ReceiptHandle -> ReceiptHandle -> ReceiptHandle)
-> (ReceiptHandle -> ReceiptHandle -> ReceiptHandle)
-> Ord ReceiptHandle
ReceiptHandle -> ReceiptHandle -> Bool
ReceiptHandle -> ReceiptHandle -> Ordering
ReceiptHandle -> ReceiptHandle -> ReceiptHandle
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: ReceiptHandle -> ReceiptHandle -> Ordering
compare :: ReceiptHandle -> ReceiptHandle -> Ordering
$c< :: ReceiptHandle -> ReceiptHandle -> Bool
< :: ReceiptHandle -> ReceiptHandle -> Bool
$c<= :: ReceiptHandle -> ReceiptHandle -> Bool
<= :: ReceiptHandle -> ReceiptHandle -> Bool
$c> :: ReceiptHandle -> ReceiptHandle -> Bool
> :: ReceiptHandle -> ReceiptHandle -> Bool
$c>= :: ReceiptHandle -> ReceiptHandle -> Bool
>= :: ReceiptHandle -> ReceiptHandle -> Bool
$cmax :: ReceiptHandle -> ReceiptHandle -> ReceiptHandle
max :: ReceiptHandle -> ReceiptHandle -> ReceiptHandle
$cmin :: ReceiptHandle -> ReceiptHandle -> ReceiptHandle
min :: ReceiptHandle -> ReceiptHandle -> ReceiptHandle
Ord, Int -> ReceiptHandle -> ShowS
[ReceiptHandle] -> ShowS
ReceiptHandle -> String
(Int -> ReceiptHandle -> ShowS)
-> (ReceiptHandle -> String)
-> ([ReceiptHandle] -> ShowS)
-> Show ReceiptHandle
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ReceiptHandle -> ShowS
showsPrec :: Int -> ReceiptHandle -> ShowS
$cshow :: ReceiptHandle -> String
show :: ReceiptHandle -> String
$cshowList :: [ReceiptHandle] -> ShowS
showList :: [ReceiptHandle] -> ShowS
Show)

{- | Wrap a backend's delivery token (an SQS receipt handle, a Pub\/Sub @ackId@)
as an opaque 'ReceiptHandle'. For backend implementations only -- worker code
obtains handles from 'receive', never builds them.
-}
mkReceiptHandle :: Text -> ReceiptHandle
mkReceiptHandle :: Text -> ReceiptHandle
mkReceiptHandle = Text -> ReceiptHandle
ReceiptHandle

{- | Recover the backend's delivery token from a 'ReceiptHandle', to pass back to
the backend on 'ack' \/ 'extendVisibility'. For backend implementations only.
-}
unReceiptHandle :: ReceiptHandle -> Text
unReceiptHandle :: ReceiptHandle -> Text
unReceiptHandle (ReceiptHandle Text
t) = Text
t

{- | A received message: the 'MirrorJob' to process together with the
'ReceiptHandle' used to 'ack' it (or 'extendVisibility' on it) once processed.
-}
data QueueMessage = QueueMessage
    { QueueMessage -> MirrorJob
msgJob :: MirrorJob
    -- ^ The job to process.
    , QueueMessage -> ReceiptHandle
msgReceipt :: ReceiptHandle
    -- ^ The handle identifying this delivery, for 'ack' \/ 'extendVisibility'.
    }
    deriving stock (QueueMessage -> QueueMessage -> Bool
(QueueMessage -> QueueMessage -> Bool)
-> (QueueMessage -> QueueMessage -> Bool) -> Eq QueueMessage
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: QueueMessage -> QueueMessage -> Bool
== :: QueueMessage -> QueueMessage -> Bool
$c/= :: QueueMessage -> QueueMessage -> Bool
/= :: QueueMessage -> QueueMessage -> Bool
Eq, Int -> QueueMessage -> ShowS
[QueueMessage] -> ShowS
QueueMessage -> String
(Int -> QueueMessage -> ShowS)
-> (QueueMessage -> String)
-> ([QueueMessage] -> ShowS)
-> Show QueueMessage
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QueueMessage -> ShowS
showsPrec :: Int -> QueueMessage -> ShowS
$cshow :: QueueMessage -> String
show :: QueueMessage -> String
$cshowList :: [QueueMessage] -> ShowS
showList :: [QueueMessage] -> ShowS
Show)

{- | A duration in whole seconds, for 'extendVisibility'. A 'newtype' so a raw
@Int@ of seconds is never confused with some other count.
-}
newtype Seconds = Seconds Int
    deriving stock (Seconds -> Seconds -> Bool
(Seconds -> Seconds -> Bool)
-> (Seconds -> Seconds -> Bool) -> Eq Seconds
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Seconds -> Seconds -> Bool
== :: Seconds -> Seconds -> Bool
$c/= :: Seconds -> Seconds -> Bool
/= :: Seconds -> Seconds -> Bool
Eq, Eq Seconds
Eq Seconds =>
(Seconds -> Seconds -> Ordering)
-> (Seconds -> Seconds -> Bool)
-> (Seconds -> Seconds -> Bool)
-> (Seconds -> Seconds -> Bool)
-> (Seconds -> Seconds -> Bool)
-> (Seconds -> Seconds -> Seconds)
-> (Seconds -> Seconds -> Seconds)
-> Ord Seconds
Seconds -> Seconds -> Bool
Seconds -> Seconds -> Ordering
Seconds -> Seconds -> Seconds
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: Seconds -> Seconds -> Ordering
compare :: Seconds -> Seconds -> Ordering
$c< :: Seconds -> Seconds -> Bool
< :: Seconds -> Seconds -> Bool
$c<= :: Seconds -> Seconds -> Bool
<= :: Seconds -> Seconds -> Bool
$c> :: Seconds -> Seconds -> Bool
> :: Seconds -> Seconds -> Bool
$c>= :: Seconds -> Seconds -> Bool
>= :: Seconds -> Seconds -> Bool
$cmax :: Seconds -> Seconds -> Seconds
max :: Seconds -> Seconds -> Seconds
$cmin :: Seconds -> Seconds -> Seconds
min :: Seconds -> Seconds -> Seconds
Ord, Int -> Seconds -> ShowS
[Seconds] -> ShowS
Seconds -> String
(Int -> Seconds -> ShowS)
-> (Seconds -> String) -> ([Seconds] -> ShowS) -> Show Seconds
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Seconds -> ShowS
showsPrec :: Int -> Seconds -> ShowS
$cshow :: Seconds -> String
show :: Seconds -> String
$cshowList :: [Seconds] -> ShowS
showList :: [Seconds] -> ShowS
Show)

{- | Why a queue operation could not be delivered to the backend, reported as a
__value__ on every handle field: the closed transport cause a consumer branches
on, and the backend's rendered detail for its log line. The cause vocabulary is
"Ecluse.Core.Fault"'s ('TransportCause'); a cloud backend's service-level
refusal (a throttle, an access denial) classifies as 'TransportProtocol' with
the service detail carried. Build one by adopting an already-classified
transport fault ('Ecluse.Core.Fault.transportFault') with 'queueTransportFault',
so the detail stays bounded.

Every fault is __safe to absorb__ under the handle's contract: an enqueue fault
is the documented best-effort loss (re-enqueued on the next demand), a receive
fault is a failed poll (retried after backoff), and an ack or visibility fault
just means the message redelivers (idempotent). The typed channel exists so each
caller makes that absorption decision explicitly, with the cause in hand.
-}
data QueueFault = QueueFault
    { QueueFault -> TransportCause
qfCause :: TransportCause
    -- ^ The closed classification a consumer or an operator reads.
    , QueueFault -> Text
qfDetail :: Text
    {- ^ The backend's rendered detail, bounded to a log-line-sized budget.
    Diagnostic text only: it is never parsed, and no decision may branch on it.
    -}
    }
    deriving stock (QueueFault -> QueueFault -> Bool
(QueueFault -> QueueFault -> Bool)
-> (QueueFault -> QueueFault -> Bool) -> Eq QueueFault
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: QueueFault -> QueueFault -> Bool
== :: QueueFault -> QueueFault -> Bool
$c/= :: QueueFault -> QueueFault -> Bool
/= :: QueueFault -> QueueFault -> Bool
Eq, Int -> QueueFault -> ShowS
[QueueFault] -> ShowS
QueueFault -> String
(Int -> QueueFault -> ShowS)
-> (QueueFault -> String)
-> ([QueueFault] -> ShowS)
-> Show QueueFault
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QueueFault -> ShowS
showsPrec :: Int -> QueueFault -> ShowS
$cshow :: QueueFault -> String
show :: QueueFault -> String
$cshowList :: [QueueFault] -> ShowS
showList :: [QueueFault] -> ShowS
Show)

{- | Adopt an already-classified 'TransportFault' (an adapter edge's
classification of its client library's exception) as a 'QueueFault'. The
'TransportFault' side ('Ecluse.Core.Fault.transportFault') truncates the detail
to the shared log-line budget, so the two vocabularies cannot drift on what
"bounded" means.
-}
queueTransportFault :: TransportFault -> QueueFault
queueTransportFault :: TransportFault -> QueueFault
queueTransportFault (TransportFault TransportCause
cause Text
detail) = TransportCause -> Text -> QueueFault
QueueFault TransportCause
cause Text
detail

{- | The mirror-queue handle -- a record of functions over a backend whose private
state the closures capture. See the module header for the @enqueue@ /
don't-@ack@-to-retry / no-@nack@ conventions; all fields are 'IO', and each
reports its backend failures as a 'QueueFault' __value__, so no queue outage
ever rides the exception channel through a caller.
-}
data MirrorQueue = MirrorQueue
    { MirrorQueue -> MirrorJob -> IO (Either QueueFault ())
enqueue :: MirrorJob -> IO (Either QueueFault ())
    {- ^ Producer. __Best-effort__: runs on the request hot path, so a 'Left' is
    counted\/logged by the caller and never fails the client response (see the
    header); the lost job is re-enqueued on the next demand for its artifact.
    -}
    , MirrorQueue -> IO (Either QueueFault [QueueMessage])
receive :: IO (Either QueueFault [QueueMessage])
    {- ^ Consumer. One long-poll for a batch of messages; @Right []@ on timeout
    (an empty, healthy poll), so the worker loop simply polls again. A 'Left' is
    a failed poll: the worker logs it and backs off, and -- unlike an empty
    poll -- it does __not__ advance the liveness heartbeat, so a persistently
    failing backend still surfaces through @\/livez@.
    -}
    , MirrorQueue -> ReceiptHandle -> IO (Either QueueFault ())
ack :: ReceiptHandle -> IO (Either QueueFault ())
    {- ^ Acknowledge a processed message so it is not redelivered. __Not__ acking
    is how a failed job is retried (the header's "retry is don't ack"), so a
    'Left' here is absorbed after logging: the processed message redelivers, and
    idempotent publishing makes the repeat harmless.
    -}
    , MirrorQueue
-> ReceiptHandle -> Seconds -> IO (Either QueueFault ())
extendVisibility :: ReceiptHandle -> Seconds -> IO (Either QueueFault ())
    {- ^ Extend a received message's visibility window to hold a long publish. An
    optimization, not correctness-critical (redelivery is harmless), so a 'Left'
    is absorbed silently by the caller.
    -}
    , MirrorQueue -> ReceiptHandle -> IO (Either QueueFault ())
deadLetter :: ReceiptHandle -> IO (Either QueueFault ())
    {- ^ Realise a __terminal__ fault: a job that can never succeed (an artifact past
    the plan-sized byte cap), decided as a verdict at the read site. Each backend
    routes it to its own dead-letter terminus -- the in-memory backend drops the
    delivery (its only terminus; observability is the worker's log and metric), the
    SQS backend returns the message with a backoff visibility timeout __without
    deleting it__, so it rides the operator's redrive policy to the dead-letter queue
    rather than being silently discarded. Distinct from 'ack' (a clean retire) and
    from not-acking (a transient redelivery); see the header's terminus convention. A
    'Left' is absorbed after logging, like 'ack'.
    -}
    }

{- | The inert queue a deployment with zero mirroring mounts carries, so the
composition-root 'Env' keeps its total shape without a backend. It is unreachable
by construction (no serve path enqueues on a mount that never mirrors, and no
worker runs to poll it); reached anyway, 'enqueue' is a typed, counted refusal --
never a crash -- and 'receive' is the empty healthy poll.
-}
noMirrorQueue :: MirrorQueue
noMirrorQueue :: MirrorQueue
noMirrorQueue =
    MirrorQueue
        { enqueue :: MirrorJob -> IO (Either QueueFault ())
enqueue = \MirrorJob
_ -> Either QueueFault () -> IO (Either QueueFault ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (QueueFault -> Either QueueFault ()
forall a b. a -> Either a b
Left QueueFault
inertFault)
        , receive :: IO (Either QueueFault [QueueMessage])
receive = Either QueueFault [QueueMessage]
-> IO (Either QueueFault [QueueMessage])
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([QueueMessage] -> Either QueueFault [QueueMessage]
forall a b. b -> Either a b
Right [])
        , ack :: ReceiptHandle -> IO (Either QueueFault ())
ack = \ReceiptHandle
_ -> Either QueueFault () -> IO (Either QueueFault ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either QueueFault ()
forall a b. b -> Either a b
Right ())
        , extendVisibility :: ReceiptHandle -> Seconds -> IO (Either QueueFault ())
extendVisibility = \ReceiptHandle
_ Seconds
_ -> Either QueueFault () -> IO (Either QueueFault ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either QueueFault ()
forall a b. b -> Either a b
Right ())
        , deadLetter :: ReceiptHandle -> IO (Either QueueFault ())
deadLetter = \ReceiptHandle
_ -> Either QueueFault () -> IO (Either QueueFault ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either QueueFault ()
forall a b. b -> Either a b
Right ())
        }
  where
    inertFault :: QueueFault
inertFault =
        TransportFault -> QueueFault
queueTransportFault
            (TransportCause -> Text -> TransportFault
transportFault TransportCause
TransportProtocol Text
"no mount mirrors, so no mirror queue is built")

{- | Hand a job to a bounded queue within the caller's transaction: write it when
there is room, or drop it at the cap (drop-newest) and return the incremented
running drop total for the caller's report policy. Dropping rather than blocking
keeps the producer non-blocking, and the loss is safe: a dropped job is
re-enqueued on the next demand for its artifact. A backend building block, shared
by the bounded in-memory backend ("Ecluse.Core.Queue.Memory") and
'newEnqueueBuffer''s hand-off so the two cannot drift on the drop policy.
-}
writeOrDrop :: TBQueue MirrorJob -> TVar Int -> MirrorJob -> STM (Maybe Int)
writeOrDrop :: TBQueue MirrorJob -> TVar Int -> MirrorJob -> STM (Maybe Int)
writeOrDrop TBQueue MirrorJob
queue TVar Int
dropCount MirrorJob
job = do
    full <- TBQueue MirrorJob -> STM Bool
forall a. TBQueue a -> STM Bool
isFullTBQueue TBQueue MirrorJob
queue
    if full
        then Just <$> bumpCount dropCount
        else writeTBQueue queue job $> Nothing

-- Increment a running counter and return the new total.
bumpCount :: TVar Int -> STM Int
bumpCount :: TVar Int -> STM Int
bumpCount TVar Int
counter = do
    n <- (Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) (Int -> Int) -> STM Int -> STM Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar Int -> STM Int
forall a. TVar a -> STM a
readTVar TVar Int
counter
    writeTVar counter n
    pure n

{- | Whether the @n@-th event in a rate-limited series should be reported: the first
(@n == 1@), then every @interval@-th. Shared by the bounded queue's drop reporting and
the composition root's enqueue-buffer reporting so the two cannot drift.
-}
reportWorthy :: Int -> Int -> Bool
reportWorthy :: Int -> Int -> Bool
reportWorthy Int
n Int
interval = Int
n Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1 Bool -> Bool -> Bool
|| Int
n Int -> Int -> Int
forall a. Integral a => a -> a -> a
`mod` Int
interval Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0

{- | Wrap a bounded __producer-side hand-off buffer__ in front of a queue, so the
serve path's 'enqueue' is an in-process STM write (microseconds) no matter how slow
the backend's own producer call is.

The motivating case is the SQS backend: its 'enqueue' is an HTTP round trip
(@SendMessage@), and the serve path runs the mirror enqueue after the response body
has been sent but before the handler returns -- so on a keep-alive connection those
milliseconds hold the connection's turn and tax the next request on it. Buffered,
the handler hands the job off and returns; the returned __drain loop__ -- which the
composition root runs alongside the server -- delivers buffered jobs to the backend
at the backend's own pace. The consumer fields ('receive', 'ack',
'extendVisibility') pass through untouched.

Loss stays safe, so the buffer keeps the handle's best-effort producer contract
(mirroring is demand-driven: a lost job is re-enqueued on the next demand for its
artifact -- the same argument "Ecluse.Core.Queue.Memory"'s bounded backend makes):

* __Drop-newest on overflow.__ A hand-off finding the buffer full drops the job and
  invokes @onDrop@ with the running drop total. The callback fires on __every__
  drop (metric-grade); the caller owns any log rate-limiting.
* __A backend failure inside the drain loop__ invokes @onDeliveryFailure@ with the
  running failure total and the failure's detail, then the loop __backs off__ (bounded,
  growing with consecutive failures) before the next job so a persistently-unreachable
  backend is retried at a bounded rate rather than hot-looping; the failed job is not
  redelivered here, and the monotonic failure count is the operator's degraded-hand-off
  surface.
* __Cancellation loses the buffer.__ The drain loop never returns, so the
  composition root races it against the services; shutdown cancels it and any
  still-buffered jobs are dropped -- the same safe loss.

The wrapped 'enqueue' never fails: it is always @Right ()@ (a drop is the
documented safe loss, reported through @onDrop@, not a fault), so the never-fails
producer contract is visible in the type.
-}
newEnqueueBuffer ::
    {- | Buffer depth: how many undelivered jobs the hand-off retains before
    dropping the newest.
    -}
    Int ->
    -- | Invoked on every hand-off drop, with the running drop total.
    (Int -> IO ()) ->
    {- | Invoked on every backend delivery failure, with the running failure total
    and the failure's detail.
    -}
    (Int -> Text -> IO ()) ->
    -- | The backend whose 'enqueue' is being decoupled from its callers.
    MirrorQueue ->
    IO (MirrorQueue, IO ())
newEnqueueBuffer :: Int
-> (Int -> IO ())
-> (Int -> Text -> IO ())
-> MirrorQueue
-> IO (MirrorQueue, IO ())
newEnqueueBuffer Int
depth Int -> IO ()
onDrop Int -> Text -> IO ()
onDeliveryFailure MirrorQueue
backend = do
    -- A capacity of at least one, so a degenerate depth can never make the
    -- hand-off an always-full drop (the same guard the bounded backend applies).
    buffer <- Natural -> IO (TBQueue MirrorJob)
forall a. Natural -> IO (TBQueue a)
newTBQueueIO (Int -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
1 Int
depth))
    dropCount <- newTVarIO (0 :: Int)
    failureCount <- newTVarIO (0 :: Int)
    let
        -- Unlike the bounded backend, every hand-off drop reports (metric-grade);
        -- the caller owns any log rate-limiting.
        handOff MirrorJob
job = do
            dropped <- STM (Maybe Int) -> IO (Maybe Int)
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (TBQueue MirrorJob -> TVar Int -> MirrorJob -> STM (Maybe Int)
writeOrDrop TBQueue MirrorJob
buffer TVar Int
dropCount MirrorJob
job)
            -- 'onDrop' is a best-effort observer (log/metric) and runs on the serve
            -- hot path, so a throwing observer must never turn a safe drop into an
            -- exception on the client response: guard it (async-safe 'tryAny').
            whenJust dropped (void . tryAny . onDrop)
            -- The hand-off is an in-process STM write with a drop policy: it has
            -- no fault to report (the header's never-fails producer contract,
            -- visible in the type as an always-'Right').
            pure (Right ())
    pure (backend{enqueue = handOff}, drainLoop buffer failureCount onDeliveryFailure backend)

{- The drain loop: deliver buffered jobs to the backend's own 'enqueue', forever. Each
iteration blocks until a job is buffered, then delivers it. A delivery failure is
reported through the best-effort failure callback (guarded, so a throwing observer
cannot tear the loop down), then the loop __backs off__ before the next delivery so a
persistently-unreachable backend is retried at a bounded rate rather than hot-looping
through the buffer and shedding every job at once. The backoff grows with consecutive
failures to a cap and resets on the next success; the failed job is not redelivered here
(the safe loss 'newEnqueueBuffer' documents: it is re-enqueued on the next demand for its
artifact, and the running failure count the callback carries is the operator's surface
for a persistently-degraded hand-off). -}
drainLoop :: TBQueue MirrorJob -> TVar Int -> (Int -> Text -> IO ()) -> MirrorQueue -> IO ()
drainLoop :: TBQueue MirrorJob
-> TVar Int -> (Int -> Text -> IO ()) -> MirrorQueue -> IO ()
drainLoop TBQueue MirrorJob
buffer TVar Int
failureCount Int -> Text -> IO ()
onDeliveryFailure MirrorQueue
backend = Int -> IO ()
forall {b}. Int -> IO b
go Int
0
  where
    go :: Int -> IO b
go Int
consecutiveFailures = do
        job <- STM MirrorJob -> IO MirrorJob
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (TBQueue MirrorJob -> STM MirrorJob
forall a. TBQueue a -> STM a
readTBQueue TBQueue MirrorJob
buffer)
        -- The backend reports its delivery failures as 'QueueFault' values, so the
        -- branch is a total match; an exception escaping here is an invariant
        -- break, left to the loop's supervisor.
        enqueue backend job >>= \case
            Right () -> Int -> IO b
go Int
0
            Left QueueFault
fault -> do
                n <- STM Int -> IO Int
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (TVar Int -> STM Int
bumpCount TVar Int
failureCount)
                -- 'onDeliveryFailure' is a best-effort observer; guard it so a throwing
                -- observer can never escape the loop and tear down the composition root.
                void (tryAny (onDeliveryFailure n (qfDetail fault)))
                threadDelay (backoffMicros drainBackoff consecutiveFailures)
                go (consecutiveFailures + 1)

{- The bounded backoff between failed deliveries (the shared
'Ecluse.Core.Supervision.BackoffSchedule' shape): from 200ms towards a 30s cap as
consecutive failures mount, so a persistently-dead backend is retried at most
once per the cap interval. This is the loop's own per-delivery pacing over the
typed fault channel; the supervision combinator wrapping the whole loop paces
only residue. -}
drainBackoff :: BackoffSchedule
drainBackoff :: BackoffSchedule
drainBackoff = BackoffSchedule{bsBaseMicros :: Int
bsBaseMicros = Int
200_000, bsCapMicros :: Int
bsCapMicros = Int
30_000_000}