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

{- | The AWS SQS backend behind the 'MirrorQueue' handle.

Maps the handle's receive → process → ack shape onto SQS:

* 'enqueue' → @SendMessage@ (the 'MirrorJob' encoded as the message body),
* 'receive' → one long-poll @ReceiveMessage@ (a batch, @[]@ on an empty poll),
* 'ack' → @DeleteMessage@ (the message is gone, never redelivered),
* 'extendVisibility' → @ChangeMessageVisibility@ (hold a long publish),
* 'deadLetter' → @ChangeMessageVisibility@ with the 'sqsTerminalBackoff' window and
  __no @DeleteMessage@__ (a terminal fault rides the redrive policy to the DLQ).

The provider differences SQS embodies -- the visibility timeout, the long-poll
window, the batch limit -- are 'SqsConfig' knobs with sane defaults, and the SQS
receipt handle is carried opaquely in a 'ReceiptHandle' (via 'mkReceiptHandle'),
so none of it leaks past the handle. __Retry is "don't ack"__: a job whose
processing fails transiently is simply not 'ack'ed, and SQS redelivers it once the
visibility timeout lapses; persistent failures fall to the queue's native dead-letter
(max-receive-count), so there is no @nack@ (see "Ecluse.Core.Queue"). A __terminal__
fault ('deadLetter') is returned with a backoff window and never deleted, so it too
falls to the operator's dead-letter queue rather than being discarded; this assumes
the operator's redrive policy exists (the no-DLQ case is issue #935). Every
operation reports its AWS failure as the handle's typed
'Ecluse.Core.Queue.QueueFault' value, classified into the core transport
vocabulary at this edge ("Ecluse.Runtime.Aws.Fault"), so a queue outage never
rides the exception channel through a caller.

The @amazonka@ 'AWS.Env' is built once at 'newSqsQueue' and captured by the
handle's closures, so the backend's state never reaches the proxy's @Env@\/@App@
(see @docs\/architecture\/technology-stack.md@ → "Key Decisions"). The
'MirrorJob' wire mapping is a plain JSON object, decoded on 'receive'; a body that
fails to parse is dropped rather than yielded as a partial, so -- like any message
left unprocessed -- it is not 'ack'ed and SQS redelivers it, ultimately to the
dead-letter queue. Each drop (a missing body or receipt, or an undecodable body) is
logged at 'DebugS' with its reason and the SQS message id when present, so a poison
message is visible rather than cycling silently; the untrusted body is never logged.

The SQS queue is a __trusted, operator-declared destination__ (the configured queue
URL, or an endpoint override): like the OTLP telemetry endpoint (see
"Ecluse.Runtime.Telemetry.Resolve"), it is reached through @amazonka@'s own client and is
__not__ subject to the data-plane egress controls (the host allowlist and the https-only
egress posture of "Ecluse.Core.Security.Egress"), which guard only untrusted package
downloads, never a destination the operator configured.
-}
module Ecluse.Runtime.Queue.Sqs (
    -- * Configuration
    SqsConfig (..),
    SqsEndpoint (..),
    defaultSqsConfig,

    -- * The backend
    newSqsQueue,

    -- * Received-message lifting
    ReceivedMessage (..),
    liftReceivedMessages,

    -- * Job wire mapping
    encodeJob,
    decodeJob,
) where

import Amazonka qualified as AWS

import Amazonka.SQS.ChangeMessageVisibility qualified as SQS
import Amazonka.SQS.DeleteMessage qualified as SQS
import Amazonka.SQS.ReceiveMessage qualified as SQS
import Amazonka.SQS.SendMessage qualified as SQS
import Amazonka.SQS.Types qualified as SQS
import Control.Monad.Trans.Resource (runResourceT)
import Data.Aeson (
    eitherDecodeStrict',
    object,
    withObject,
    (.:),
    (.:?),
    (.=),
 )
import Data.Aeson qualified as Aeson
import Data.Aeson.Types (Parser, parseEither)
import Katip (LogEnv, Severity (DebugS), logFM, ls, sl)
import Katip.Monadic (runKatipContextT)
import Lens.Micro ((?~), (^.))

import Ecluse.Core.Ecosystem (ecosystemName, parseEcosystem)
import Ecluse.Core.Package (
    mkPackageName,
    mkScope,
    pkgEcosystem,
    pkgNamespace,
    unScope,
    unscopedName,
 )
import Ecluse.Core.Queue (
    MirrorJob (..),
    MirrorQueue (..),
    QueueFault,
    QueueMessage (..),
    RemoteSpanContext (RemoteSpanContext, rscTraceparent, rscTracestate),
    Seconds (..),
    mkReceiptHandle,
    queueTransportFault,
    unReceiptHandle,
 )
import Ecluse.Core.Security.Egress (RegistryUrl, registryUrlText)
import Ecluse.Core.Version (mkVersion, renderVersion)
import Ecluse.Runtime.Aws.Fault (classifyAwsTransport)
import Ecluse.Runtime.Log (moduleField)

{- | Where an SQS-compatible endpoint lives, for pointing the backend at a
non-default host: a local emulator (@ministack@) in tests, or a VPC endpoint. A
non-default host: a local emulator (@ministack@) in tests, or a VPC endpoint.
-}
data SqsEndpoint = SqsEndpoint
    { SqsEndpoint -> Bool
endpointSecure :: Bool
    -- ^ Whether to connect over HTTPS (an emulator is usually plain HTTP).
    , SqsEndpoint -> Text
endpointHost :: Text
    -- ^ The host to connect to (e.g. @"localhost"@).
    , SqsEndpoint -> Int
endpointPort :: Int
    -- ^ The port to connect to (e.g. @4566@ for ministack).
    }
    deriving stock (SqsEndpoint -> SqsEndpoint -> Bool
(SqsEndpoint -> SqsEndpoint -> Bool)
-> (SqsEndpoint -> SqsEndpoint -> Bool) -> Eq SqsEndpoint
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SqsEndpoint -> SqsEndpoint -> Bool
== :: SqsEndpoint -> SqsEndpoint -> Bool
$c/= :: SqsEndpoint -> SqsEndpoint -> Bool
/= :: SqsEndpoint -> SqsEndpoint -> Bool
Eq, Int -> SqsEndpoint -> ShowS
[SqsEndpoint] -> ShowS
SqsEndpoint -> String
(Int -> SqsEndpoint -> ShowS)
-> (SqsEndpoint -> String)
-> ([SqsEndpoint] -> ShowS)
-> Show SqsEndpoint
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SqsEndpoint -> ShowS
showsPrec :: Int -> SqsEndpoint -> ShowS
$cshow :: SqsEndpoint -> String
show :: SqsEndpoint -> String
$cshowList :: [SqsEndpoint] -> ShowS
showList :: [SqsEndpoint] -> ShowS
Show)

{- | What the SQS backend needs. The batch size, long-poll window, and visibility
timeout are provider knobs (see "Ecluse.Core.Queue") with defaults in
'defaultSqsConfig'.
-}
data SqsConfig = SqsConfig
    { SqsConfig -> Text
sqsQueueUrl :: Text
    -- ^ The fully-qualified SQS queue URL mirror jobs are sent to and received from.
    , SqsConfig -> Text
sqsRegion :: Text
    -- ^ The AWS region the queue lives in (e.g. @"us-east-1"@).
    , SqsConfig -> Maybe SqsEndpoint
sqsEndpoint :: Maybe SqsEndpoint
    {- ^ An endpoint override for an emulator or VPC endpoint; 'Nothing' uses
    @amazonka@'s default resolution and the ambient credential chain.
    -}
    , SqsConfig -> Int
sqsBatchSize :: Int
    {- ^ Maximum messages to pull per 'receive' (SQS caps this at 10). A larger
    batch amortises the round-trip when the queue is busy.
    -}
    , SqsConfig -> Int
sqsWaitSeconds :: Int
    {- ^ The long-poll window in seconds (SQS caps this at 20): how long a
    'receive' waits for a message before returning @[]@, so an idle worker does
    not hot-loop on empty polls.
    -}
    , SqsConfig -> Seconds
sqsVisibilityTimeout :: Seconds
    {- ^ How long a received message stays hidden from other 'receive's before SQS
    redelivers it -- the budget for processing-then-'ack', extendable per message
    via 'extendVisibility'.
    -}
    , SqsConfig -> Seconds
sqsTerminalBackoff :: Seconds
    {- ^ The visibility timeout 'deadLetter' returns a __terminal__ message with
    (@ChangeMessageVisibility@, never @DeleteMessage@): larger than the normal
    processing window so a permanently-unmirrorable artifact is not re-fetched in a
    hot loop, while it rides the operator's redrive policy to the dead-letter queue.
    A per-attempt incremental backoff would need the @ApproximateReceiveCount@
    attribute (deferred with the receive-count work in issue #935); this fixed
    backoff is the conservative default.
    -}
    }
    deriving stock (SqsConfig -> SqsConfig -> Bool
(SqsConfig -> SqsConfig -> Bool)
-> (SqsConfig -> SqsConfig -> Bool) -> Eq SqsConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SqsConfig -> SqsConfig -> Bool
== :: SqsConfig -> SqsConfig -> Bool
$c/= :: SqsConfig -> SqsConfig -> Bool
/= :: SqsConfig -> SqsConfig -> Bool
Eq, Int -> SqsConfig -> ShowS
[SqsConfig] -> ShowS
SqsConfig -> String
(Int -> SqsConfig -> ShowS)
-> (SqsConfig -> String)
-> ([SqsConfig] -> ShowS)
-> Show SqsConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SqsConfig -> ShowS
showsPrec :: Int -> SqsConfig -> ShowS
$cshow :: SqsConfig -> String
show :: SqsConfig -> String
$cshowList :: [SqsConfig] -> ShowS
showList :: [SqsConfig] -> ShowS
Show)

{- | A 'SqsConfig' for a queue URL and region with the provider knobs at sane
defaults: a full batch of 10, the maximum 20-second long poll, and a 30-second
visibility timeout. Override the record fields to tune them, or set 'sqsEndpoint'
to target an emulator.
-}
defaultSqsConfig :: Text -> Text -> SqsConfig
defaultSqsConfig :: Text -> Text -> SqsConfig
defaultSqsConfig Text
queueUrl Text
region =
    SqsConfig
        { sqsQueueUrl :: Text
sqsQueueUrl = Text
queueUrl
        , sqsRegion :: Text
sqsRegion = Text
region
        , sqsEndpoint :: Maybe SqsEndpoint
sqsEndpoint = Maybe SqsEndpoint
forall a. Maybe a
Nothing
        , sqsBatchSize :: Int
sqsBatchSize = Int
10
        , sqsWaitSeconds :: Int
sqsWaitSeconds = Int
20
        , sqsVisibilityTimeout :: Seconds
sqsVisibilityTimeout = Int -> Seconds
Seconds Int
30
        , sqsTerminalBackoff :: Seconds
sqsTerminalBackoff = Int -> Seconds
Seconds Int
300
        }

{- | Build an SQS-backed 'MirrorQueue'. The @amazonka@ 'AWS.Env' is constructed
once here -- region-scoped, and pointed at 'sqsEndpoint' with its throwaway
credentials when one is given, otherwise discovering the ambient AWS credential
chain -- and captured by the returned handle's closures.
-}
newSqsQueue :: LogEnv -> (Text -> Either Text RegistryUrl) -> SqsConfig -> IO MirrorQueue
newSqsQueue :: LogEnv
-> (Text -> Either Text RegistryUrl) -> SqsConfig -> IO MirrorQueue
newSqsQueue LogEnv
logEnv Text -> Either Text RegistryUrl
egressUrl SqsConfig
cfg = do
    env <- SqsConfig -> IO Env
mkEnv SqsConfig
cfg
    -- Every operation reports its AWS failure as the handle's 'QueueFault' value:
    -- 'AWS.sendEither' keeps the error sum out of the exception channel, and the
    -- shared classifier folds it into the core transport vocabulary at this edge.
    let run :: (AWS.AWSRequest a) => a -> IO (Either QueueFault (AWS.AWSResponse a))
        run = (Either Error (AWSResponse a) -> Either QueueFault (AWSResponse a))
-> IO (Either Error (AWSResponse a))
-> IO (Either QueueFault (AWSResponse a))
forall a b. (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ((Error -> QueueFault)
-> Either Error (AWSResponse a)
-> Either QueueFault (AWSResponse a)
forall a b c. (a -> b) -> Either a c -> Either b c
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first (TransportFault -> QueueFault
queueTransportFault (TransportFault -> QueueFault)
-> (Error -> TransportFault) -> Error -> QueueFault
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Error -> TransportFault
classifyAwsTransport)) (IO (Either Error (AWSResponse a))
 -> IO (Either QueueFault (AWSResponse a)))
-> (a -> IO (Either Error (AWSResponse a)))
-> a
-> IO (Either QueueFault (AWSResponse a))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ResourceT IO (Either Error (AWSResponse a))
-> IO (Either Error (AWSResponse a))
forall (m :: * -> *) a. MonadUnliftIO m => ResourceT m a -> m a
runResourceT (ResourceT IO (Either Error (AWSResponse a))
 -> IO (Either Error (AWSResponse a)))
-> (a -> ResourceT IO (Either Error (AWSResponse a)))
-> a
-> IO (Either Error (AWSResponse a))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Env -> a -> ResourceT IO (Either Error (AWSResponse a))
forall (m :: * -> *) a.
(MonadResource m, AWSRequest a) =>
Env -> a -> m (Either Error (AWSResponse a))
AWS.sendEither Env
env
        queueUrl = SqsConfig -> Text
sqsQueueUrl SqsConfig
cfg
        Seconds terminalBackoffSecs = sqsTerminalBackoff cfg
    pure
        MirrorQueue
            { enqueue = fmap void . run . SQS.newSendMessage queueUrl . encodeJob
            , receive = do
                outcome <- run (receiveRequest cfg)
                traverse (liftReceivedMessages logEnv egressUrl . receivedMessages) outcome
            , ack = fmap void . run . SQS.newDeleteMessage queueUrl . unReceiptHandle
            , extendVisibility = \ReceiptHandle
receipt (Seconds Int
secs) ->
                (Either QueueFault ChangeMessageVisibilityResponse
 -> Either QueueFault ())
-> IO (Either QueueFault ChangeMessageVisibilityResponse)
-> IO (Either QueueFault ())
forall a b. (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Either QueueFault ChangeMessageVisibilityResponse
-> Either QueueFault ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either QueueFault ChangeMessageVisibilityResponse)
 -> IO (Either QueueFault ()))
-> (ChangeMessageVisibility
    -> IO (Either QueueFault ChangeMessageVisibilityResponse))
-> ChangeMessageVisibility
-> IO (Either QueueFault ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChangeMessageVisibility
-> IO (Either QueueFault (AWSResponse ChangeMessageVisibility))
ChangeMessageVisibility
-> IO (Either QueueFault ChangeMessageVisibilityResponse)
forall a.
AWSRequest a =>
a -> IO (Either QueueFault (AWSResponse a))
run (ChangeMessageVisibility -> IO (Either QueueFault ()))
-> ChangeMessageVisibility -> IO (Either QueueFault ())
forall a b. (a -> b) -> a -> b
$
                    Text -> Text -> Int -> ChangeMessageVisibility
SQS.newChangeMessageVisibility Text
queueUrl (ReceiptHandle -> Text
unReceiptHandle ReceiptHandle
receipt) Int
secs
            , -- A terminal fault: return the message with the backoff visibility timeout
              -- (@ChangeMessageVisibility@), __never__ @DeleteMessage@, so it is not
              -- silently discarded but rides the operator's redrive policy to the
              -- dead-letter queue -- the well-monitored terminus with forensic retention.
              deadLetter = \ReceiptHandle
receipt ->
                (Either QueueFault ChangeMessageVisibilityResponse
 -> Either QueueFault ())
-> IO (Either QueueFault ChangeMessageVisibilityResponse)
-> IO (Either QueueFault ())
forall a b. (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Either QueueFault ChangeMessageVisibilityResponse
-> Either QueueFault ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either QueueFault ChangeMessageVisibilityResponse)
 -> IO (Either QueueFault ()))
-> (ChangeMessageVisibility
    -> IO (Either QueueFault ChangeMessageVisibilityResponse))
-> ChangeMessageVisibility
-> IO (Either QueueFault ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChangeMessageVisibility
-> IO (Either QueueFault (AWSResponse ChangeMessageVisibility))
ChangeMessageVisibility
-> IO (Either QueueFault ChangeMessageVisibilityResponse)
forall a.
AWSRequest a =>
a -> IO (Either QueueFault (AWSResponse a))
run (ChangeMessageVisibility -> IO (Either QueueFault ()))
-> ChangeMessageVisibility -> IO (Either QueueFault ())
forall a b. (a -> b) -> a -> b
$
                    Text -> Text -> Int -> ChangeMessageVisibility
SQS.newChangeMessageVisibility Text
queueUrl (ReceiptHandle -> Text
unReceiptHandle ReceiptHandle
receipt) Int
terminalBackoffSecs
            }

-- Build the region-scoped, optionally endpoint-overridden amazonka environment.
mkEnv :: SqsConfig -> IO AWS.Env
mkEnv :: SqsConfig -> IO Env
mkEnv SqsConfig
cfg = case SqsConfig -> Maybe SqsEndpoint
sqsEndpoint SqsConfig
cfg of
    Just SqsEndpoint
ep -> do
        base <- Env -> Env
regioned (Env -> Env) -> IO Env -> IO Env
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (EnvNoAuth -> IO Env) -> IO Env
forall (m :: * -> *). MonadIO m => (EnvNoAuth -> m Env) -> m Env
AWS.newEnv EnvNoAuth -> IO Env
forall (m :: * -> *) (withAuth :: * -> *).
(MonadCatch m, MonadIO m, Foldable withAuth) =>
Env' withAuth -> m Env
AWS.discover
        pure (configured ep base)
    Maybe SqsEndpoint
Nothing -> Env -> Env
regioned (Env -> Env) -> IO Env -> IO Env
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (EnvNoAuth -> IO Env) -> IO Env
forall (m :: * -> *). MonadIO m => (EnvNoAuth -> m Env) -> m Env
AWS.newEnv EnvNoAuth -> IO Env
forall (m :: * -> *) (withAuth :: * -> *).
(MonadCatch m, MonadIO m, Foldable withAuth) =>
Env' withAuth -> m Env
AWS.discover
  where
    regioned :: AWS.Env -> AWS.Env
    regioned :: Env -> Env
regioned Env
env = Env
env{AWS.region = AWS.Region' (sqsRegion cfg)}

    configured :: SqsEndpoint -> AWS.Env -> AWS.Env
    configured :: SqsEndpoint -> Env -> Env
configured SqsEndpoint
ep =
        Service -> Env -> Env
forall (withAuth :: * -> *).
Service -> Env' withAuth -> Env' withAuth
AWS.configureService
            ( Bool -> ByteString -> Int -> Service -> Service
AWS.setEndpoint
                (SqsEndpoint -> Bool
endpointSecure SqsEndpoint
ep)
                (Text -> ByteString
forall a b. ConvertUtf8 a b => a -> b
encodeUtf8 (SqsEndpoint -> Text
endpointHost SqsEndpoint
ep))
                (SqsEndpoint -> Int
endpointPort SqsEndpoint
ep)
                Service
SQS.defaultService
            )

-- One long-poll ReceiveMessage with the configured batch / wait / visibility.
-- SQS caps the long-poll ('sqsWaitSeconds') at 20s, which stays within amazonka's
-- default per-service request timeout, so the client never cuts a long-poll short
-- and no explicit response-timeout override is needed; a configured wait above the
-- SQS cap is clamped by SQS, so the relationship cannot be broken from config.
receiveRequest :: SqsConfig -> SQS.ReceiveMessage
receiveRequest :: SqsConfig -> ReceiveMessage
receiveRequest SqsConfig
cfg =
    Text -> ReceiveMessage
SQS.newReceiveMessage (SqsConfig -> Text
sqsQueueUrl SqsConfig
cfg)
        ReceiveMessage
-> (ReceiveMessage -> ReceiveMessage) -> ReceiveMessage
forall a b. a -> (a -> b) -> b
& (Maybe Int -> Identity (Maybe Int))
-> ReceiveMessage -> Identity ReceiveMessage
Lens' ReceiveMessage (Maybe Int)
SQS.receiveMessage_maxNumberOfMessages
        ((Maybe Int -> Identity (Maybe Int))
 -> ReceiveMessage -> Identity ReceiveMessage)
-> Int -> ReceiveMessage -> ReceiveMessage
forall s t a b. ASetter s t a (Maybe b) -> b -> s -> t
?~ SqsConfig -> Int
sqsBatchSize SqsConfig
cfg
            ReceiveMessage
-> (ReceiveMessage -> ReceiveMessage) -> ReceiveMessage
forall a b. a -> (a -> b) -> b
& (Maybe Int -> Identity (Maybe Int))
-> ReceiveMessage -> Identity ReceiveMessage
Lens' ReceiveMessage (Maybe Int)
SQS.receiveMessage_waitTimeSeconds
        ((Maybe Int -> Identity (Maybe Int))
 -> ReceiveMessage -> Identity ReceiveMessage)
-> Int -> ReceiveMessage -> ReceiveMessage
forall s t a b. ASetter s t a (Maybe b) -> b -> s -> t
?~ SqsConfig -> Int
sqsWaitSeconds SqsConfig
cfg
            ReceiveMessage
-> (ReceiveMessage -> ReceiveMessage) -> ReceiveMessage
forall a b. a -> (a -> b) -> b
& (Maybe Int -> Identity (Maybe Int))
-> ReceiveMessage -> Identity ReceiveMessage
Lens' ReceiveMessage (Maybe Int)
SQS.receiveMessage_visibilityTimeout
        ((Maybe Int -> Identity (Maybe Int))
 -> ReceiveMessage -> Identity ReceiveMessage)
-> Int -> ReceiveMessage -> ReceiveMessage
forall s t a b. ASetter s t a (Maybe b) -> b -> s -> t
?~ Int
visibilitySeconds
  where
    Seconds Int
visibilitySeconds = SqsConfig -> Seconds
sqsVisibilityTimeout SqsConfig
cfg

{- | The fields of a received SQS message the backend reads. Lifting them out of
the @amazonka@ 'SQS.Message' keeps the 'QueueMessage' mapping (and its drop
decision) free of the AWS type, so the receive path's drop behaviour is exercised
directly in tests.
-}
data ReceivedMessage = ReceivedMessage
    { ReceivedMessage -> Maybe Text
rmBody :: Maybe Text
    -- ^ The message body carrying the encoded 'MirrorJob' (SQS always supplies one).
    , ReceivedMessage -> Maybe Text
rmReceipt :: Maybe Text
    -- ^ The receipt handle a later 'ack' deletes the message by (SQS always supplies one).
    , ReceivedMessage -> Maybe Text
rmMessageId :: Maybe Text
    -- ^ The SQS-assigned message id, for the drop log; not part of the untrusted body.
    }
    deriving stock (ReceivedMessage -> ReceivedMessage -> Bool
(ReceivedMessage -> ReceivedMessage -> Bool)
-> (ReceivedMessage -> ReceivedMessage -> Bool)
-> Eq ReceivedMessage
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ReceivedMessage -> ReceivedMessage -> Bool
== :: ReceivedMessage -> ReceivedMessage -> Bool
$c/= :: ReceivedMessage -> ReceivedMessage -> Bool
/= :: ReceivedMessage -> ReceivedMessage -> Bool
Eq, Int -> ReceivedMessage -> ShowS
[ReceivedMessage] -> ShowS
ReceivedMessage -> String
(Int -> ReceivedMessage -> ShowS)
-> (ReceivedMessage -> String)
-> ([ReceivedMessage] -> ShowS)
-> Show ReceivedMessage
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ReceivedMessage -> ShowS
showsPrec :: Int -> ReceivedMessage -> ShowS
$cshow :: ReceivedMessage -> String
show :: ReceivedMessage -> String
$cshowList :: [ReceivedMessage] -> ShowS
showList :: [ReceivedMessage] -> ShowS
Show)

-- The read fields of an amazonka Message, lifted at the effectful edge ('receive').
receivedFields :: SQS.Message -> ReceivedMessage
receivedFields :: Message -> ReceivedMessage
receivedFields Message
message =
    ReceivedMessage
        { rmBody :: Maybe Text
rmBody = Message
message Message -> Getting (Maybe Text) Message (Maybe Text) -> Maybe Text
forall s a. s -> Getting a s a -> a
^. Getting (Maybe Text) Message (Maybe Text)
Lens' Message (Maybe Text)
SQS.message_body
        , rmReceipt :: Maybe Text
rmReceipt = Message
message Message -> Getting (Maybe Text) Message (Maybe Text) -> Maybe Text
forall s a. s -> Getting a s a -> a
^. Getting (Maybe Text) Message (Maybe Text)
Lens' Message (Maybe Text)
SQS.message_receiptHandle
        , rmMessageId :: Maybe Text
rmMessageId = Message
message Message -> Getting (Maybe Text) Message (Maybe Text) -> Maybe Text
forall s a. s -> Getting a s a -> a
^. Getting (Maybe Text) Message (Maybe Text)
Lens' Message (Maybe Text)
SQS.message_messageId
        }

-- The received batch's messages, each reduced to the fields the backend reads.
receivedMessages :: SQS.ReceiveMessageResponse -> [ReceivedMessage]
receivedMessages :: ReceiveMessageResponse -> [ReceivedMessage]
receivedMessages ReceiveMessageResponse
response =
    [ReceivedMessage]
-> ([Message] -> [ReceivedMessage])
-> Maybe [Message]
-> [ReceivedMessage]
forall b a. b -> (a -> b) -> Maybe a -> b
maybe [] ((Message -> ReceivedMessage) -> [Message] -> [ReceivedMessage]
forall a b. (a -> b) -> [a] -> [b]
map Message -> ReceivedMessage
receivedFields) (ReceiveMessageResponse
response ReceiveMessageResponse
-> Getting
     (Maybe [Message]) ReceiveMessageResponse (Maybe [Message])
-> Maybe [Message]
forall s a. s -> Getting a s a -> a
^. Getting (Maybe [Message]) ReceiveMessageResponse (Maybe [Message])
Lens' ReceiveMessageResponse (Maybe [Message])
SQS.receiveMessageResponse_messages)

-- Why a received message could not become a QueueMessage. A closed set with no
-- payload, so a drop log never echoes any of the (untrusted) message contents.
data SqsDropReason = MissingBody | MissingReceipt | UndecodableBody
    deriving stock (SqsDropReason -> SqsDropReason -> Bool
(SqsDropReason -> SqsDropReason -> Bool)
-> (SqsDropReason -> SqsDropReason -> Bool) -> Eq SqsDropReason
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SqsDropReason -> SqsDropReason -> Bool
== :: SqsDropReason -> SqsDropReason -> Bool
$c/= :: SqsDropReason -> SqsDropReason -> Bool
/= :: SqsDropReason -> SqsDropReason -> Bool
Eq, Int -> SqsDropReason -> ShowS
[SqsDropReason] -> ShowS
SqsDropReason -> String
(Int -> SqsDropReason -> ShowS)
-> (SqsDropReason -> String)
-> ([SqsDropReason] -> ShowS)
-> Show SqsDropReason
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SqsDropReason -> ShowS
showsPrec :: Int -> SqsDropReason -> ShowS
$cshow :: SqsDropReason -> String
show :: SqsDropReason -> String
$cshowList :: [SqsDropReason] -> ShowS
showList :: [SqsDropReason] -> ShowS
Show)

{- Lift one received message into a QueueMessage, or report why it cannot be. A
message missing its body or receipt (which SQS always supplies), or one whose body
does not decode, is dropped rather than crashing the poll: it is left un-acked, so
the visibility timeout redelivers it and a persistently bad message falls to the
dead-letter queue. -}
toQueueMessage :: (Text -> Either Text RegistryUrl) -> ReceivedMessage -> Either SqsDropReason QueueMessage
toQueueMessage :: (Text -> Either Text RegistryUrl)
-> ReceivedMessage -> Either SqsDropReason QueueMessage
toQueueMessage Text -> Either Text RegistryUrl
egressUrl ReceivedMessage
received = do
    body <- SqsDropReason -> Maybe Text -> Either SqsDropReason Text
forall l r. l -> Maybe r -> Either l r
maybeToRight SqsDropReason
MissingBody (ReceivedMessage -> Maybe Text
rmBody ReceivedMessage
received)
    receipt <- maybeToRight MissingReceipt (rmReceipt received)
    job <- first (const UndecodableBody) (decodeJob egressUrl body)
    pure QueueMessage{msgJob = job, msgReceipt = mkReceiptHandle receipt}

{- | Lift a received batch into deliverable 'QueueMessage's, logging each dropped
message (a missing body or receipt, or an undecodable body) at 'DebugS' so a poison
message is visible rather than cycling silently until the queue's max-receive count.
A dropped message is omitted from the result and left un-'ack'ed, so redelivery and
dead-letter behaviour are unchanged.
-}
liftReceivedMessages :: LogEnv -> (Text -> Either Text RegistryUrl) -> [ReceivedMessage] -> IO [QueueMessage]
liftReceivedMessages :: LogEnv
-> (Text -> Either Text RegistryUrl)
-> [ReceivedMessage]
-> IO [QueueMessage]
liftReceivedMessages LogEnv
logEnv Text -> Either Text RegistryUrl
egressUrl =
    ([Maybe QueueMessage] -> [QueueMessage])
-> IO [Maybe QueueMessage] -> IO [QueueMessage]
forall a b. (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [Maybe QueueMessage] -> [QueueMessage]
forall a. [Maybe a] -> [a]
catMaybes (IO [Maybe QueueMessage] -> IO [QueueMessage])
-> ([ReceivedMessage] -> IO [Maybe QueueMessage])
-> [ReceivedMessage]
-> IO [QueueMessage]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (ReceivedMessage -> IO (Maybe QueueMessage))
-> [ReceivedMessage] -> IO [Maybe QueueMessage]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse (LogEnv
-> (Text -> Either Text RegistryUrl)
-> ReceivedMessage
-> IO (Maybe QueueMessage)
liftReceivedMessage LogEnv
logEnv Text -> Either Text RegistryUrl
egressUrl)

-- Deliver a received message, or log the drop at DebugS and yield Nothing.
liftReceivedMessage :: LogEnv -> (Text -> Either Text RegistryUrl) -> ReceivedMessage -> IO (Maybe QueueMessage)
liftReceivedMessage :: LogEnv
-> (Text -> Either Text RegistryUrl)
-> ReceivedMessage
-> IO (Maybe QueueMessage)
liftReceivedMessage LogEnv
logEnv Text -> Either Text RegistryUrl
egressUrl ReceivedMessage
received =
    case (Text -> Either Text RegistryUrl)
-> ReceivedMessage -> Either SqsDropReason QueueMessage
toQueueMessage Text -> Either Text RegistryUrl
egressUrl ReceivedMessage
received of
        Right QueueMessage
queueMessage -> Maybe QueueMessage -> IO (Maybe QueueMessage)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (QueueMessage -> Maybe QueueMessage
forall a. a -> Maybe a
Just QueueMessage
queueMessage)
        Left SqsDropReason
reason -> Maybe QueueMessage
forall a. Maybe a
Nothing Maybe QueueMessage -> IO () -> IO (Maybe QueueMessage)
forall a b. a -> IO b -> IO a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ LogEnv -> SqsDropReason -> Maybe Text -> IO ()
logSqsDrop LogEnv
logEnv SqsDropReason
reason (ReceivedMessage -> Maybe Text
rmMessageId ReceivedMessage
received)

-- One DebugS line naming why a received message was dropped, and its SQS message id
-- when present. The message body is untrusted payload and is never logged.
logSqsDrop :: LogEnv -> SqsDropReason -> Maybe Text -> IO ()
logSqsDrop :: LogEnv -> SqsDropReason -> Maybe Text -> IO ()
logSqsDrop LogEnv
logEnv SqsDropReason
reason Maybe Text
messageId =
    LogEnv
-> SimpleLogPayload -> Namespace -> KatipContextT IO () -> IO ()
forall c (m :: * -> *) a.
LogItem c =>
LogEnv -> c -> Namespace -> KatipContextT m a -> m a
runKatipContextT LogEnv
logEnv SimpleLogPayload
payload Namespace
forall a. Monoid a => a
mempty (Severity -> LogStr -> KatipContextT IO ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
DebugS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls Text
message))
  where
    payload :: SimpleLogPayload
payload =
        Text -> SimpleLogPayload
moduleField Text
"Ecluse.Runtime.Queue.Sqs"
            SimpleLogPayload -> SimpleLogPayload -> SimpleLogPayload
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"reason" (SqsDropReason -> Text
dropReasonLabel SqsDropReason
reason)
            SimpleLogPayload -> SimpleLogPayload -> SimpleLogPayload
forall a. Semigroup a => a -> a -> a
<> SimpleLogPayload
-> (Text -> SimpleLogPayload) -> Maybe Text -> SimpleLogPayload
forall b a. b -> (a -> b) -> Maybe a -> b
maybe SimpleLogPayload
forall a. Monoid a => a
mempty (Text -> Text -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"messageId") Maybe Text
messageId
    message :: Text
message = Text
"dropped an unusable SQS message: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> SqsDropReason -> Text
dropReasonLabel SqsDropReason
reason

-- The operator-facing phrase for each drop reason.
dropReasonLabel :: SqsDropReason -> Text
dropReasonLabel :: SqsDropReason -> Text
dropReasonLabel = \case
    SqsDropReason
MissingBody -> Text
"missing body"
    SqsDropReason
MissingReceipt -> Text
"missing receipt"
    SqsDropReason
UndecodableBody -> Text
"undecodable body"

{- | Encode a 'MirrorJob' as the JSON text of an SQS message body. The inverse of
'decodeJob': the package identity is split into its ecosystem, optional scope, and
bare name so it round-trips through 'mkPackageName', and the version keeps its raw
string. The serve-time-admitted artifact's filename rides as a plain field: it is
the selection key the worker's ingest re-evaluation gates by, and the only thing
of the artifact the wire carries -- the digests and size the worker verifies and
publishes with are derived from current metadata, never the payload.
-}
encodeJob :: MirrorJob -> Text
encodeJob :: MirrorJob -> Text
encodeJob MirrorJob
job =
    ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 (ByteString -> Text) -> (Value -> ByteString) -> Value -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Value -> ByteString
forall a. ToJSON a => a -> ByteString
Aeson.encode (Value -> Text) -> Value -> Text
forall a b. (a -> b) -> a -> b
$
        [Pair] -> Value
object
            [ Key
"ecosystem" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= Ecosystem -> Text
ecosystemName (PackageName -> Ecosystem
pkgEcosystem PackageName
name)
            , Key
"scope" Key -> Maybe Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= (Scope -> Text
unScope (Scope -> Text) -> Maybe Scope -> Maybe Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> PackageName -> Maybe Scope
pkgNamespace PackageName
name)
            , Key
"name" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= PackageName -> Text
unscopedName PackageName
name
            , Key
"version" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= Version -> Text
renderVersion (MirrorJob -> Version
jobVersion MirrorJob
job)
            , Key
"artifactUrl" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= RegistryUrl -> Text
registryUrlText (MirrorJob -> RegistryUrl
jobArtifactUrl MirrorJob
job)
            , Key
"filename" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= MirrorJob -> Text
jobArtifactFilename MirrorJob
job
            , Key
"traceContext" Key -> Maybe Value -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= (RemoteSpanContext -> Value
encodeTraceContext (RemoteSpanContext -> Value)
-> Maybe RemoteSpanContext -> Maybe Value
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> MirrorJob -> Maybe RemoteSpanContext
jobTraceContext MirrorJob
job)
            ]
  where
    name :: PackageName
name = MirrorJob -> PackageName
jobPackage MirrorJob
job

-- Encode the optional enqueue-span trace-context carrier: the W3C traceparent and
-- tracestate verbatim, so the worker can re-establish the cross-async span link. A
-- 'Nothing' carrier (tracing was off at enqueue) serialises to a JSON null and
-- round-trips back to 'Nothing'.
encodeTraceContext :: RemoteSpanContext -> Aeson.Value
encodeTraceContext :: RemoteSpanContext -> Value
encodeTraceContext RemoteSpanContext
rsc =
    [Pair] -> Value
object
        [ Key
"traceparent" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= RemoteSpanContext -> Text
rscTraceparent RemoteSpanContext
rsc
        , Key
"tracestate" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= RemoteSpanContext -> Text
rscTracestate RemoteSpanContext
rsc
        ]

{- | Decode an SQS message body back into a 'MirrorJob', or a human-readable error
if the body is not the JSON object 'encodeJob' produces (a missing field, an
unknown ecosystem, an artifact URL the egress former refuses, malformed JSON).

The queue payload is a __trust boundary__, so the artifact URL is re-formed into
its 'RegistryUrl' egress witness on decode through the given former -- the
composition root passes the https-only 'Ecluse.Core.Security.Egress.mkRegistryUrl';
the loopback test harnesses pass their flag-gated dev former. A URL the former
refuses fails the decode, so a tampered or misproduced message can never hand the
worker's fetch an unwitnessed URL (it redelivers and falls to the dead-letter
queue, like any undecodable body).
-}
decodeJob :: (Text -> Either Text RegistryUrl) -> Text -> Either Text MirrorJob
decodeJob :: (Text -> Either Text RegistryUrl) -> Text -> Either Text MirrorJob
decodeJob Text -> Either Text RegistryUrl
egressUrl Text
body =
    (String -> Text) -> Either String Value -> Either Text Value
forall a b c. (a -> b) -> Either a c -> Either b c
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first String -> Text
forall a. ToText a => a -> Text
toText (ByteString -> Either String Value
forall a. FromJSON a => ByteString -> Either String a
eitherDecodeStrict' (Text -> ByteString
forall a b. ConvertUtf8 a b => a -> b
encodeUtf8 Text
body))
        Either Text Value
-> (Value -> Either Text MirrorJob) -> Either Text MirrorJob
forall a b. Either Text a -> (a -> Either Text b) -> Either Text b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (String -> Text)
-> Either String MirrorJob -> Either Text MirrorJob
forall a b c. (a -> b) -> Either a c -> Either b c
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first String -> Text
forall a. ToText a => a -> Text
toText (Either String MirrorJob -> Either Text MirrorJob)
-> (Value -> Either String MirrorJob)
-> Value
-> Either Text MirrorJob
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Value -> Parser MirrorJob) -> Value -> Either String MirrorJob
forall a b. (a -> Parser b) -> a -> Either String b
parseEither ((Text -> Either Text RegistryUrl) -> Value -> Parser MirrorJob
parseMirrorJob Text -> Either Text RegistryUrl
egressUrl)

-- Parse the top-level job object 'encodeJob' writes, delegating the nested
-- trace-context carrier to 'parseTraceContext'.
parseMirrorJob :: (Text -> Either Text RegistryUrl) -> Aeson.Value -> Parser MirrorJob
parseMirrorJob :: (Text -> Either Text RegistryUrl) -> Value -> Parser MirrorJob
parseMirrorJob Text -> Either Text RegistryUrl
egressUrl = String -> (Object -> Parser MirrorJob) -> Value -> Parser MirrorJob
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"MirrorJob" ((Object -> Parser MirrorJob) -> Value -> Parser MirrorJob)
-> (Object -> Parser MirrorJob) -> Value -> Parser MirrorJob
forall a b. (a -> b) -> a -> b
$ \Object
o -> do
    ecoName <- Object
o Object -> Key -> Parser Text
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"ecosystem"
    eco <- maybe (fail (unknownEcosystem ecoName)) pure (parseEcosystem ecoName)
    scope <- o .:? "scope"
    rawName <- o .: "name"
    rawVersion <- o .: "version"
    rawArtifactUrl <- o .: "artifactUrl"
    -- Re-form the egress witness at the wire boundary: the type the worker's fetch
    -- requires cannot be fabricated from an unvalidated payload string.
    artifactUrl <- either (fail . toString) pure (egressUrl rawArtifactUrl)
    filename <- o .: "filename"
    -- The trace-context carrier is optional: a job from an older producer (or one
    -- enqueued with tracing off) carries no "traceContext", which decodes to
    -- 'Nothing' and simply yields no span link in the worker.
    traceContext <- o .:? "traceContext" >>= traverse parseTraceContext
    pure
        MirrorJob
            { jobPackage = mkPackageName eco (mkScope <$> scope) rawName
            , jobVersion = mkVersion eco rawVersion
            , jobArtifactUrl = artifactUrl
            , jobArtifactFilename = filename
            , jobTraceContext = traceContext
            }
  where
    unknownEcosystem :: Text -> a
unknownEcosystem Text
n = a
"unknown ecosystem " a -> a -> a
forall a. Semigroup a => a -> a -> a
<> Text -> a
forall b a. (Show a, IsString b) => a -> b
show (Text
n :: Text)

-- Parse the optional trace-context carrier back into a 'RemoteSpanContext': the W3C
-- traceparent and tracestate verbatim. The carrier is untrusted opaque transport, so
-- both fields are taken as-is -- an unparseable W3C value is the tracing port's concern
-- (it yields no link), never a decode failure that would strand a serviceable job.
parseTraceContext :: Aeson.Value -> Parser RemoteSpanContext
parseTraceContext :: Value -> Parser RemoteSpanContext
parseTraceContext = String
-> (Object -> Parser RemoteSpanContext)
-> Value
-> Parser RemoteSpanContext
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"RemoteSpanContext" ((Object -> Parser RemoteSpanContext)
 -> Value -> Parser RemoteSpanContext)
-> (Object -> Parser RemoteSpanContext)
-> Value
-> Parser RemoteSpanContext
forall a b. (a -> b) -> a -> b
$ \Object
t ->
    Text -> Text -> RemoteSpanContext
RemoteSpanContext (Text -> Text -> RemoteSpanContext)
-> Parser Text -> Parser (Text -> RemoteSpanContext)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Object
t Object -> Key -> Parser Text
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"traceparent" Parser (Text -> RemoteSpanContext)
-> Parser Text -> Parser RemoteSpanContext
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
t Object -> Key -> Parser Text
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"tracestate"