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

{- | The advisory database's sync mechanics: detect a new @osv.db@ artifact in
object storage, download it bounded, verify it, and shadow-swap it into the
read path, one ecosystem per task, driven by the configured mounts.

The write side of "Ecluse.Core.Cve.Slot": 'syncStep' performs exactly one
detect-download-verify-swap cycle over an injected 'CveFetch' (so unit tests
drive it without a network), and 'runCveSync' schedules those steps: an eager
__boot burst__ (an immediate attempt, retried with incremental backoff, that is
eventually allowed to fail so a broken bucket never wedges startup) followed by
the steady ETag poll. The proxy is rules-engine complete as early as the
artifact can be had; before then it serves deny-by-default.

The swap's file discipline: the download lands in a temp file beside the
canonical per-ecosystem path, and 'Ecluse.Core.Cve.openCveDb' verifies the
temp file (epoch stamp, table shape, ecosystem), the artifact contract's
verify-before-swap. __The connection that verified is the connection that
serves__: the accepted temp file is renamed atomically onto the canonical
name, the open connection follows the inode through the rename, and that same
'CveDb' is swapped in; there is no reopen and so no verify-to-serve gap. The
displaced generation drains and closes inside 'Ecluse.Core.Cve.Slot.swapIn', releasing
the old inode's last reference; reclamation is the kernel's, never a delete
this code could mistime. A rejected artifact is deleted, its ETag remembered
(re-downloading a known-bad object every poll buys nothing), and the last-good
generation keeps serving.
-}
module Ecluse.Runtime.Cve.Sync (
    -- * The injected transport
    CveFetch (..),
    DbEtag (..),
    OsvDbFetchFault (..),
    OsvDbCapExceeded (..),
    S3CveSource,
    newS3CveSource,
    s3CveFetchFor,
    cappedAt,

    -- * One sync cycle
    SyncEnv (..),
    SyncOutcome (..),
    syncStep,

    -- * The scheduled task
    SyncSchedule (..),
    runCveSync,
    bootBackoffDelays,
    bootBurstPolicy,
) where

import Conduit (ConduitT, await, runResourceT, yield, (.|))
import Control.Retry (RetryPolicyM, RetryStatus (rsIterNumber), retryPolicy, retrying)
import Data.ByteString qualified as BS
import Data.Conduit.Combinators qualified as C
import Katip (KatipContext, Severity (DebugS, ErrorS, InfoS), logFM, ls)
import Network.HTTP.Types.Status (statusCode)
import System.Directory (removeFile, renameFile)
import UnliftIO (MonadUnliftIO)
import UnliftIO.Concurrent (threadDelay)
import UnliftIO.Exception (catch, catchAny, mask, onException, throwIO)

import Amazonka qualified as AWS
import Amazonka.S3 qualified as S3
import Amazonka.S3.Lens qualified as S3L
import Lens.Micro ((^.))

import Ecluse.Core.Cve (CveDb (cveDbClose, cveDbMeta), CveDbRejected, DbEtag (..), openCveDb)
import Ecluse.Core.Cve.Slot (CveSlot, swapIn)
import Ecluse.Core.Ecosystem (Ecosystem)
import Ecluse.Core.Fault (TransportFault)
import Ecluse.Runtime.Aws.Fault (classifyAwsTransport)
import Ecluse.Runtime.Aws.S3 (buildS3Env)

{- | The sync transport, as data: how to learn the remote artifact's current
version and how to fetch its bytes. Injected so 'syncStep' is unit-testable
without a network; the composition root draws one from a 'newS3CveSource'.
-}
data CveFetch = CveFetch
    { CveFetch -> IO (Either OsvDbFetchFault (Maybe DbEtag))
fetchHeadEtag :: IO (Either OsvDbFetchFault (Maybe DbEtag))
    {- ^ The remote artifact's current ETag; @Right Nothing@ when the object does
    not exist (not yet published for this ecosystem). Every fetch failure --
    a transport fault included -- is the 'Left' value.
    -}
    , CveFetch -> FilePath -> IO (Either OsvDbFetchFault DbEtag)
fetchDownload :: FilePath -> IO (Either OsvDbFetchFault DbEtag)
    {- ^ Download the artifact to the given path (byte-bounded) and return the
    ETag of the bytes actually fetched, the download's own rather than an
    earlier @HEAD@'s, so a publish racing the poll is recorded truthfully.
    Every fetch failure -- an overstepped byte cap, a missing ETag, a transport
    fault -- is the 'Left' value; a 'Left' may leave a partial file at the
    given path for the caller to discard.
    -}
    }

{- | Why an artifact fetch did not yield usable bytes: refused by this side (the
object oversteps the configured byte cap, or the response carried no ETag to
record), or not delivered at all (a transport fault, classified into the core
vocabulary at the adapter edge). A value on the 'CveFetch' channel, never an
exception: the sync task's step folds it into its outcome and the schedule
retries.
-}
data OsvDbFetchFault
    = -- | The object exceeds the configured byte cap (carried, in bytes).
      OsvDbTooLarge Int
    | -- | The response carried no ETag; nothing truthful to record.
      OsvDbNoEtag
    | -- | The transport could not deliver the object (carried, classified).
      OsvDbTransport TransportFault
    deriving stock (OsvDbFetchFault -> OsvDbFetchFault -> Bool
(OsvDbFetchFault -> OsvDbFetchFault -> Bool)
-> (OsvDbFetchFault -> OsvDbFetchFault -> Bool)
-> Eq OsvDbFetchFault
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: OsvDbFetchFault -> OsvDbFetchFault -> Bool
== :: OsvDbFetchFault -> OsvDbFetchFault -> Bool
$c/= :: OsvDbFetchFault -> OsvDbFetchFault -> Bool
/= :: OsvDbFetchFault -> OsvDbFetchFault -> Bool
Eq, Int -> OsvDbFetchFault -> ShowS
[OsvDbFetchFault] -> ShowS
OsvDbFetchFault -> FilePath
(Int -> OsvDbFetchFault -> ShowS)
-> (OsvDbFetchFault -> FilePath)
-> ([OsvDbFetchFault] -> ShowS)
-> Show OsvDbFetchFault
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> OsvDbFetchFault -> ShowS
showsPrec :: Int -> OsvDbFetchFault -> ShowS
$cshow :: OsvDbFetchFault -> FilePath
show :: OsvDbFetchFault -> FilePath
$cshowList :: [OsvDbFetchFault] -> ShowS
showList :: [OsvDbFetchFault] -> ShowS
Show)

{- | The byte cap's mid-stream escape hatch: 'cappedAt' sits inside a conduit
pipeline (no value channel of its own), so it reports an overstepped cap by
throwing this -- __confined__ typed exception, caught at the adapter boundary
('s3Download') and folded into 'OsvDbTooLarge'. It never crosses the 'CveFetch'
interface.
-}
newtype OsvDbCapExceeded = OsvDbCapExceeded Int
    deriving stock (OsvDbCapExceeded -> OsvDbCapExceeded -> Bool
(OsvDbCapExceeded -> OsvDbCapExceeded -> Bool)
-> (OsvDbCapExceeded -> OsvDbCapExceeded -> Bool)
-> Eq OsvDbCapExceeded
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: OsvDbCapExceeded -> OsvDbCapExceeded -> Bool
== :: OsvDbCapExceeded -> OsvDbCapExceeded -> Bool
$c/= :: OsvDbCapExceeded -> OsvDbCapExceeded -> Bool
/= :: OsvDbCapExceeded -> OsvDbCapExceeded -> Bool
Eq, Int -> OsvDbCapExceeded -> ShowS
[OsvDbCapExceeded] -> ShowS
OsvDbCapExceeded -> FilePath
(Int -> OsvDbCapExceeded -> ShowS)
-> (OsvDbCapExceeded -> FilePath)
-> ([OsvDbCapExceeded] -> ShowS)
-> Show OsvDbCapExceeded
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> OsvDbCapExceeded -> ShowS
showsPrec :: Int -> OsvDbCapExceeded -> ShowS
$cshow :: OsvDbCapExceeded -> FilePath
show :: OsvDbCapExceeded -> FilePath
$cshowList :: [OsvDbCapExceeded] -> ShowS
showList :: [OsvDbCapExceeded] -> ShowS
Show)

instance Exception OsvDbCapExceeded

-- | Everything one ecosystem's sync task operates on.
data SyncEnv = SyncEnv
    { SyncEnv -> CveFetch
syncFetch :: CveFetch
    -- ^ The transport for this ecosystem's object key.
    , SyncEnv -> Ecosystem
syncEcosystem :: Ecosystem
    -- ^ The ecosystem the artifact must verify as.
    , SyncEnv -> FilePath
syncDbPath :: FilePath
    -- ^ The canonical on-disk artifact path (the stable per-ecosystem name).
    , SyncEnv -> CveSlot
syncSlot :: CveSlot
    -- ^ The slot this task's swaps publish to.
    }

{- | What one 'syncStep' concluded; the caller ('runCveSync') logs it and
decides scheduling.
-}
data SyncOutcome
    = -- | A new artifact was verified and is now live (its ETag and provenance carried).
      SyncSwapped DbEtag [(Text, Text)]
    | -- | The remote ETag matches the last seen one; nothing to do.
      SyncUnchanged
    | -- | The object does not exist in the bucket (not yet published).
      SyncAbsent
    | {- | The artifact was downloaded and __refused__ by verification; the
      last-good generation keeps serving and the ETag is remembered.
      -}
      SyncRejected DbEtag CveDbRejected
    | {- | The fetch itself failed (carried); nothing was learned about the
      remote artifact, so the last seen ETag stands and the schedule retries.
      -}
      SyncFetchFaulted OsvDbFetchFault
    deriving stock (Int -> SyncOutcome -> ShowS
[SyncOutcome] -> ShowS
SyncOutcome -> FilePath
(Int -> SyncOutcome -> ShowS)
-> (SyncOutcome -> FilePath)
-> ([SyncOutcome] -> ShowS)
-> Show SyncOutcome
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SyncOutcome -> ShowS
showsPrec :: Int -> SyncOutcome -> ShowS
$cshow :: SyncOutcome -> FilePath
show :: SyncOutcome -> FilePath
$cshowList :: [SyncOutcome] -> ShowS
showList :: [SyncOutcome] -> ShowS
Show)

{- | One detect-download-verify-swap cycle against the last seen ETag. Total
over the fetch and over verification -- a failed fetch and a refused artifact
are both outcomes, not exceptions -- so the caller's scheduling is a plain fold
over 'SyncOutcome'. See the module header for the file discipline.
-}
syncStep :: SyncEnv -> Maybe DbEtag -> IO SyncOutcome
syncStep :: SyncEnv -> Maybe DbEtag -> IO SyncOutcome
syncStep SyncEnv
env Maybe DbEtag
lastSeen =
    CveFetch -> IO (Either OsvDbFetchFault (Maybe DbEtag))
fetchHeadEtag (SyncEnv -> CveFetch
syncFetch SyncEnv
env) IO (Either OsvDbFetchFault (Maybe DbEtag))
-> (Either OsvDbFetchFault (Maybe DbEtag) -> IO SyncOutcome)
-> IO SyncOutcome
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Left OsvDbFetchFault
fault -> SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (OsvDbFetchFault -> SyncOutcome
SyncFetchFaulted OsvDbFetchFault
fault)
        Right Maybe DbEtag
Nothing -> SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure SyncOutcome
SyncAbsent
        Right (Just DbEtag
remote)
            | DbEtag -> Maybe DbEtag
forall a. a -> Maybe a
Just DbEtag
remote Maybe DbEtag -> Maybe DbEtag -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe DbEtag
lastSeen -> SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure SyncOutcome
SyncUnchanged
            | Bool
otherwise -> SyncEnv -> IO SyncOutcome
syncNewArtifact SyncEnv
env

-- The 'onException' guards absorb nothing: they discard the temp file when a
-- fault __below__ the typed channels escapes (a filesystem error writing or
-- opening the temp path), then re-propagate it as the residue it is.
syncNewArtifact :: SyncEnv -> IO SyncOutcome
syncNewArtifact :: SyncEnv -> IO SyncOutcome
syncNewArtifact SyncEnv
env = do
    let temp :: FilePath
temp = SyncEnv -> FilePath
syncDbPath SyncEnv
env FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> FilePath
".tmp"
    downloaded <- CveFetch -> FilePath -> IO (Either OsvDbFetchFault DbEtag)
fetchDownload (SyncEnv -> CveFetch
syncFetch SyncEnv
env) FilePath
temp IO (Either OsvDbFetchFault DbEtag)
-> IO () -> IO (Either OsvDbFetchFault DbEtag)
forall (m :: * -> *) a b. MonadUnliftIO m => m a -> m b -> m a
`onException` FilePath -> IO ()
discardTemp FilePath
temp
    case downloaded of
        Left OsvDbFetchFault
fault -> do
            -- A failed download may have written partial bytes to the temp path
            -- (the byte cap trips mid-stream); discard them.
            FilePath -> IO ()
discardTemp FilePath
temp
            SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (OsvDbFetchFault -> SyncOutcome
SyncFetchFaulted OsvDbFetchFault
fault)
        Right DbEtag
fetched -> do
            opened <- Ecosystem -> FilePath -> IO (Either CveDbRejected CveDb)
openCveDb (SyncEnv -> Ecosystem
syncEcosystem SyncEnv
env) FilePath
temp IO (Either CveDbRejected CveDb)
-> IO () -> IO (Either CveDbRejected CveDb)
forall (m :: * -> *) a b. MonadUnliftIO m => m a -> m b -> m a
`onException` FilePath -> IO ()
discardTemp FilePath
temp
            case opened of
                Left CveDbRejected
rejection -> do
                    FilePath -> IO ()
discardTemp FilePath
temp
                    SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DbEtag -> CveDbRejected -> SyncOutcome
SyncRejected DbEtag
fetched CveDbRejected
rejection)
                Right CveDb
db -> SyncEnv -> FilePath -> DbEtag -> CveDb -> IO SyncOutcome
publishVerified SyncEnv
env FilePath
temp DbEtag
fetched CveDb
db

publishVerified :: SyncEnv -> FilePath -> DbEtag -> CveDb -> IO SyncOutcome
publishVerified :: SyncEnv -> FilePath -> DbEtag -> CveDb -> IO SyncOutcome
publishVerified SyncEnv
env FilePath
temp DbEtag
fetched CveDb
db = ((forall a. IO a -> IO a) -> IO SyncOutcome) -> IO SyncOutcome
forall (m :: * -> *) b.
MonadUnliftIO m =>
((forall a. m a -> m a) -> m b) -> m b
mask (((forall a. IO a -> IO a) -> IO SyncOutcome) -> IO SyncOutcome)
-> ((forall a. IO a -> IO a) -> IO SyncOutcome) -> IO SyncOutcome
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
    -- The verified connection follows the inode through the rename; the
    -- canonical name now holds the newest accepted artifact and the temp name
    -- is gone. Up to here this side still owns the connection, so a failure
    -- closes it and discards the download.
    IO () -> IO ()
forall a. IO a -> IO a
restore (FilePath -> FilePath -> IO ()
renameFile FilePath
temp (SyncEnv -> FilePath
syncDbPath SyncEnv
env))
        IO () -> IO () -> IO ()
forall (m :: * -> *) a b. MonadUnliftIO m => m a -> m b -> m a
`onException` (CveDb -> IO ()
cveDbClose CveDb
db IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> FilePath -> IO ()
discardTemp FilePath
temp)
    -- 'swapIn' publishes atomically before it retires the displaced
    -- generation, and owns the connection from entry: a failure or
    -- cancellation while the displaced generation drains must never close the
    -- newly live database, so no cleanup wraps it. The mask pins the
    -- ownership handoff; the drain wait inside stays interruptible.
    CveSlot -> DbEtag -> CveDb -> IO ()
swapIn (SyncEnv -> CveSlot
syncSlot SyncEnv
env) DbEtag
fetched CveDb
db
    SyncOutcome -> IO SyncOutcome
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DbEtag -> [(Text, Text)] -> SyncOutcome
SyncSwapped DbEtag
fetched (CveDb -> [(Text, Text)]
cveDbMeta CveDb
db))

-- Best-effort: the temp may already be renamed away or never created.
discardTemp :: FilePath -> IO ()
discardTemp :: FilePath -> IO ()
discardTemp FilePath
temp = FilePath -> IO ()
removeFile FilePath
temp IO () -> (SomeException -> IO ()) -> IO ()
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> (SomeException -> m a) -> m a
`catchAny` IO () -> SomeException -> IO ()
forall a b. a -> b -> a
const IO ()
forall (f :: * -> *). Applicative f => f ()
pass

{- | The task's timing: the boot burst's backoff delays and the steady poll
interval, both in microseconds. The composition root ships 'bootBackoffDelays'
and the configured poll interval; tests inject tiny values.
-}
data SyncSchedule = SyncSchedule
    { SyncSchedule -> [Int]
schedBootBackoff :: [Int]
    -- ^ Delays before each boot-burst retry; the list's length is the budget.
    , SyncSchedule -> Int
schedPollDelay :: Int
    -- ^ The steady ETag-poll interval.
    }

{- | The shipped boot-burst backoff: an immediate first attempt, then retries
after each of these, then the burst concedes and the steady poll takes over.
Constants by design; the poll interval is the operator-facing knob.
-}
bootBackoffDelays :: [Int]
bootBackoffDelays :: [Int]
bootBackoffDelays = [Int
1_000_000, Int
2_000_000, Int
4_000_000, Int
8_000_000, Int
16_000_000]

{- | The boot-burst backoff schedule compiled to a "Control.Retry" policy: the
n-th retry waits the n-th delay (microseconds) before it, and the policy stops
(yields 'Nothing') once the list is spent, so the list's length is the retry
budget. Inspect the schedule without sleeping with 'Control.Retry.simulatePolicy'.
-}
bootBurstPolicy :: (Monad m) => [Int] -> RetryPolicyM m
bootBurstPolicy :: forall (m :: * -> *). Monad m => [Int] -> RetryPolicyM m
bootBurstPolicy [Int]
delays = (RetryStatus -> Maybe Int) -> RetryPolicyM m
forall (m :: * -> *).
Monad m =>
(RetryStatus -> Maybe Int) -> RetryPolicyM m
retryPolicy (\RetryStatus
rs -> [Int]
delays [Int] -> Int -> Maybe Int
forall a. [a] -> Int -> Maybe a
!!? RetryStatus -> Int
rsIterNumber RetryStatus
rs)

{- | One ecosystem's sync task: the boot burst, then the steady poll, forever.

The __boot burst__ attempts a sync immediately and retries per the schedule's
backoff until an artifact is live, so a healthy deployment is
rules-engine complete within seconds of boot. It concedes early on a
__rejected__ artifact (retrying the same bytes cannot end differently) and
gives up after the schedule with a warning. The proxy serves regardless, since
an empty slot only ever abstains into deny-by-default, and the poll keeps
trying.

A fetch fault arrives as a value in the step's outcome and is logged here;
residue (a filesystem fault on the temp path, a contract escape) propagates to
the supervision the composition root wraps this task in
('Ecluse.Core.Supervision.superviseLoop'), which restarts the task -- it simply
resumes from the remote artifact. @notifyFirstSync@ runs after each successful
swap (its consumer, the readiness signal, is an idempotent one-way flip).
-}
runCveSync :: (MonadUnliftIO m, KatipContext m) => SyncEnv -> SyncSchedule -> IO () -> m ()
runCveSync :: forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
SyncEnv -> SyncSchedule -> IO () -> m ()
runCveSync SyncEnv
env SyncSchedule
schedule IO ()
notifyFirstSync = do
    seen <- m (Maybe DbEtag)
burst
    poll seen
  where
    eco :: Text
eco = Ecosystem -> Text
forall b a. (Show a, IsString b) => a -> b
show (SyncEnv -> Ecosystem
syncEcosystem SyncEnv
env) :: Text

    -- The boot burst under 'Control.Retry': an immediate first attempt, then a
    -- retry on each not-settled outcome per 'bootBurstPolicy' until an artifact
    -- settles the step or the schedule is spent. 'lastSeen' is fixed at 'Nothing'
    -- because the only not-settled outcomes ('SyncAbsent', 'SyncFetchFaulted')
    -- return it untouched, so it never changes across the burst; the settled ETag
    -- is what 'poll' resumes from.
    burst :: m (Maybe DbEtag)
burst = do
        (settled, seen') <-
            RetryPolicyM m
-> (RetryStatus -> (Bool, Maybe DbEtag) -> m Bool)
-> (RetryStatus -> m (Bool, Maybe DbEtag))
-> m (Bool, Maybe DbEtag)
forall (m :: * -> *) b.
MonadIO m =>
RetryPolicyM m
-> (RetryStatus -> b -> m Bool) -> (RetryStatus -> m b) -> m b
retrying
                ([Int] -> RetryPolicyM m
forall (m :: * -> *). Monad m => [Int] -> RetryPolicyM m
bootBurstPolicy (SyncSchedule -> [Int]
schedBootBackoff SyncSchedule
schedule))
                (\RetryStatus
_ (Bool
done, Maybe DbEtag
_) -> Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool -> Bool
not Bool
done))
                (\RetryStatus
_ -> SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
loggedStep SyncEnv
env Text
eco IO ()
notifyFirstSync Maybe DbEtag
forall a. Maybe a
Nothing)
        unless settled $
            -- The boot budget is spent without an artifact. This ecosystem stays
            -- not-ready (the readiness gate reads 'csReady'), so its rules deny by
            -- default and no traffic is served against a missing advisory database;
            -- the poll continues in case the artifact appears later. Logged at
            -- 'ErrorS' because a persistent failure here is a real misconfiguration
            -- (bucket, object key, or IAM), not a condition a healthy deploy hits.
            logFM ErrorS (ls ("cve-sync[" <> eco <> "]: boot fetch did not acquire an advisory database within the boot budget; this ecosystem stays not-ready and denies by default until one is acquired. Continuing to poll; investigate the bucket, object, or IAM if this persists."))
        pure seen'

    poll :: Maybe DbEtag -> m b
poll Maybe DbEtag
lastSeen = do
        Int -> m ()
forall (m :: * -> *). MonadIO m => Int -> m ()
threadDelay (SyncSchedule -> Int
schedPollDelay SyncSchedule
schedule)
        (_, seen') <- SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
loggedStep SyncEnv
env Text
eco IO ()
notifyFirstSync Maybe DbEtag
lastSeen
        poll seen'

-- One logged step: (the burst may stop, the ETag now last seen). 'syncStep' is
-- total over the fetch and over verification, so this is a plain fold over
-- 'SyncOutcome' -- nothing is caught, and residue propagates to the task's
-- supervision at the composition root.
loggedStep :: (MonadUnliftIO m, KatipContext m) => SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
loggedStep :: forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
SyncEnv -> Text -> IO () -> Maybe DbEtag -> m (Bool, Maybe DbEtag)
loggedStep SyncEnv
env Text
eco IO ()
notifyFirstSync Maybe DbEtag
lastSeen =
    IO SyncOutcome -> m SyncOutcome
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (SyncEnv -> Maybe DbEtag -> IO SyncOutcome
syncStep SyncEnv
env Maybe DbEtag
lastSeen) m SyncOutcome
-> (SyncOutcome -> m (Bool, Maybe DbEtag))
-> m (Bool, Maybe DbEtag)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        SyncFetchFaulted OsvDbFetchFault
fault -> do
            -- Nothing was learned about the remote artifact: keep the last seen
            -- ETag, let the burst retry (or the poll try again next interval).
            Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
ErrorS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"cve-sync[" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eco Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"]: sync fetch failed: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> OsvDbFetchFault -> Text
forall b a. (Show a, IsString b) => a -> b
show OsvDbFetchFault
fault))
            (Bool, Maybe DbEtag) -> m (Bool, Maybe DbEtag)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool
False, Maybe DbEtag
lastSeen)
        SyncSwapped DbEtag
etag [(Text, Text)]
meta -> do
            Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
InfoS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"cve-sync[" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eco Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"]: advisory database swapped in: etag=" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> DbEtag -> Text
forall b a. (Show a, IsString b) => a -> b
show DbEtag
etag Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" meta=" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> [(Text, Text)] -> Text
forall b a. (Show a, IsString b) => a -> b
show [(Text, Text)]
meta))
            IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO ()
notifyFirstSync
            (Bool, Maybe DbEtag) -> m (Bool, Maybe DbEtag)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool
True, DbEtag -> Maybe DbEtag
forall a. a -> Maybe a
Just DbEtag
etag)
        SyncOutcome
SyncUnchanged -> do
            Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
DebugS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"cve-sync[" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eco Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"]: advisory database unchanged"))
            (Bool, Maybe DbEtag) -> m (Bool, Maybe DbEtag)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool
True, Maybe DbEtag
lastSeen)
        SyncOutcome
SyncAbsent -> do
            Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
DebugS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"cve-sync[" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eco Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"]: no advisory database published yet"))
            (Bool, Maybe DbEtag) -> m (Bool, Maybe DbEtag)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool
False, Maybe DbEtag
lastSeen)
        SyncRejected DbEtag
etag CveDbRejected
rejection -> do
            Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
ErrorS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"cve-sync[" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eco Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"]: downloaded artifact refused (keeping last good): " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CveDbRejected -> Text
forall b a. (Show a, IsString b) => a -> b
show CveDbRejected
rejection))
            -- Remembered so the same bad artifact is not re-downloaded
            -- every poll; a fixed re-publish carries a new ETag. The burst
            -- stops: retrying identical bytes cannot end differently.
            (Bool, Maybe DbEtag) -> m (Bool, Maybe DbEtag)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool
True, DbEtag -> Maybe DbEtag
forall a. a -> Maybe a
Just DbEtag
etag)

{- | An S3-backed advisory-fetch source: the @amazonka@ 'AWS.Env' is built once at
'newS3CveSource' and captured, so 's3CveFetchFor' yields a 'CveFetch' per (bucket,
object key, byte cap) without re-discovering credentials per mount. The runtime
adapter that seals the SDK env behind the sync's transport, matching
'Ecluse.Runtime.Queue.Sqs.newSqsQueue'; the composition shell never handles the env.
-}
newtype S3CveSource = S3CveSource
    { S3CveSource -> Text -> Text -> Int -> CveFetch
s3CveFetchFor :: Text -> Text -> Int -> CveFetch
    -- ^ A 'CveFetch' against one bucket, object key, and byte cap, over the captured env.
    }

{- | Build an 'S3CveSource', constructing the S3 @amazonka@ env once (honouring the
pre-parsed endpoint override) and capturing it, so every ecosystem's 'CveFetch'
shares one credential discovery.
-}
newS3CveSource :: Maybe (Bool, Text, Int) -> IO S3CveSource
newS3CveSource :: Maybe (Bool, Text, Int) -> IO S3CveSource
newS3CveSource Maybe (Bool, Text, Int)
mEndpoint = do
    awsEnv <- Maybe (Bool, Text, Int) -> IO Env
buildS3Env Maybe (Bool, Text, Int)
mEndpoint
    pure (S3CveSource (s3CveFetch awsEnv))

{- | The real transport over the captured env: S3 @HEAD@ for the ETag, bounded
streaming @GET@ for the bytes, against one bucket and key. Internal to the adapter,
sealed by 'newS3CveSource'; a @404@ on @HEAD@ is the honest @Right Nothing@ (not yet
published), and every other service or transport fault is the 'Left' value, classified
into the core vocabulary at this edge.
-}
s3CveFetch :: AWS.Env -> Text -> Text -> Int -> CveFetch
s3CveFetch :: Env -> Text -> Text -> Int -> CveFetch
s3CveFetch Env
awsEnv Text
bucket Text
key Int
maxBytes =
    CveFetch
        { fetchHeadEtag :: IO (Either OsvDbFetchFault (Maybe DbEtag))
fetchHeadEtag = Env -> Text -> Text -> IO (Either OsvDbFetchFault (Maybe DbEtag))
s3HeadEtag Env
awsEnv Text
bucket Text
key
        , fetchDownload :: FilePath -> IO (Either OsvDbFetchFault DbEtag)
fetchDownload = Env
-> Text
-> Text
-> Int
-> FilePath
-> IO (Either OsvDbFetchFault DbEtag)
s3Download Env
awsEnv Text
bucket Text
key Int
maxBytes
        }

s3HeadEtag :: AWS.Env -> Text -> Text -> IO (Either OsvDbFetchFault (Maybe DbEtag))
s3HeadEtag :: Env -> Text -> Text -> IO (Either OsvDbFetchFault (Maybe DbEtag))
s3HeadEtag Env
awsEnv Text
bucket Text
key =
    ResourceT IO (Either Error HeadObjectResponse)
-> IO (Either Error HeadObjectResponse)
forall (m :: * -> *) a. MonadUnliftIO m => ResourceT m a -> m a
runResourceT (Env
-> HeadObject
-> ResourceT IO (Either Error (AWSResponse HeadObject))
forall (m :: * -> *) a.
(MonadResource m, AWSRequest a) =>
Env -> a -> m (Either Error (AWSResponse a))
AWS.sendEither Env
awsEnv (BucketName -> ObjectKey -> HeadObject
S3.newHeadObject (Text -> BucketName
S3.BucketName Text
bucket) (Text -> ObjectKey
S3.ObjectKey Text
key))) IO (Either Error HeadObjectResponse)
-> (Either Error HeadObjectResponse
    -> Either OsvDbFetchFault (Maybe DbEtag))
-> IO (Either OsvDbFetchFault (Maybe DbEtag))
forall (f :: * -> *) a b. Functor f => f a -> (a -> b) -> f b
<&> \case
        Right HeadObjectResponse
resp -> Maybe DbEtag -> Either OsvDbFetchFault (Maybe DbEtag)
forall a b. b -> Either a b
Right (ETag -> DbEtag
dbEtag (ETag -> DbEtag) -> Maybe ETag -> Maybe DbEtag
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> HeadObjectResponse
resp HeadObjectResponse
-> Getting (Maybe ETag) HeadObjectResponse (Maybe ETag)
-> Maybe ETag
forall s a. s -> Getting a s a -> a
^. Getting (Maybe ETag) HeadObjectResponse (Maybe ETag)
Lens' HeadObjectResponse (Maybe ETag)
S3L.headObjectResponse_eTag)
        Left Error
err
            | Error -> Bool
isNotFound Error
err -> Maybe DbEtag -> Either OsvDbFetchFault (Maybe DbEtag)
forall a b. b -> Either a b
Right Maybe DbEtag
forall a. Maybe a
Nothing
            | Bool
otherwise -> OsvDbFetchFault -> Either OsvDbFetchFault (Maybe DbEtag)
forall a b. a -> Either a b
Left (TransportFault -> OsvDbFetchFault
OsvDbTransport (Error -> TransportFault
classifyAwsTransport Error
err))

s3Download :: AWS.Env -> Text -> Text -> Int -> FilePath -> IO (Either OsvDbFetchFault DbEtag)
s3Download :: Env
-> Text
-> Text
-> Int
-> FilePath
-> IO (Either OsvDbFetchFault DbEtag)
s3Download Env
awsEnv Text
bucket Text
key Int
maxBytes FilePath
dest = IO (Either OsvDbFetchFault DbEtag)
-> IO (Either OsvDbFetchFault DbEtag)
classified (IO (Either OsvDbFetchFault DbEtag)
 -> IO (Either OsvDbFetchFault DbEtag))
-> (ResourceT IO (Either OsvDbFetchFault DbEtag)
    -> IO (Either OsvDbFetchFault DbEtag))
-> ResourceT IO (Either OsvDbFetchFault DbEtag)
-> IO (Either OsvDbFetchFault DbEtag)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ResourceT IO (Either OsvDbFetchFault DbEtag)
-> IO (Either OsvDbFetchFault DbEtag)
forall (m :: * -> *) a. MonadUnliftIO m => ResourceT m a -> m a
runResourceT (ResourceT IO (Either OsvDbFetchFault DbEtag)
 -> IO (Either OsvDbFetchFault DbEtag))
-> ResourceT IO (Either OsvDbFetchFault DbEtag)
-> IO (Either OsvDbFetchFault DbEtag)
forall a b. (a -> b) -> a -> b
$ do
    resp <- Env -> GetObject -> ResourceT IO (AWSResponse GetObject)
forall (m :: * -> *) a.
(MonadResource m, AWSRequest a) =>
Env -> a -> m (AWSResponse a)
AWS.send Env
awsEnv (BucketName -> ObjectKey -> GetObject
S3.newGetObject (Text -> BucketName
S3.BucketName Text
bucket) (Text -> ObjectKey
S3.ObjectKey Text
key))
    -- The declared length fails fast; the streaming cap is the
    -- enforcement (a declared length is not a guarantee).
    for_ (resp ^. S3L.getObjectResponse_contentLength) $ \Integer
len ->
        Bool -> ResourceT IO () -> ResourceT IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Integer
len Integer -> Integer -> Bool
forall a. Ord a => a -> a -> Bool
> Int -> Integer
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
maxBytes) (OsvDbCapExceeded -> ResourceT IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
throwIO (Int -> OsvDbCapExceeded
OsvDbCapExceeded Int
maxBytes))
    AWS.sinkBody (resp ^. S3L.getObjectResponse_body) (cappedAt maxBytes .| C.sinkFile dest)
    pure (maybe (Left OsvDbNoEtag) (Right . dbEtag) (resp ^. S3L.getObjectResponse_eTag))
  where
    -- The adapter boundary: fold the two typed escapes into the value channel --
    -- amazonka's error sum ('AWS.send' throws it) into 'OsvDbTransport', the
    -- streaming cap's confined 'OsvDbCapExceeded' into 'OsvDbTooLarge'. Nothing
    -- else is caught: a filesystem fault writing the destination propagates as
    -- residue for the sync task's supervision to log.
    classified :: IO (Either OsvDbFetchFault DbEtag) -> IO (Either OsvDbFetchFault DbEtag)
    classified :: IO (Either OsvDbFetchFault DbEtag)
-> IO (Either OsvDbFetchFault DbEtag)
classified IO (Either OsvDbFetchFault DbEtag)
act =
        IO (Either OsvDbFetchFault DbEtag)
act
            IO (Either OsvDbFetchFault DbEtag)
-> (Error -> IO (Either OsvDbFetchFault DbEtag))
-> IO (Either OsvDbFetchFault DbEtag)
forall (m :: * -> *) e a.
(MonadUnliftIO m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` (\(Error
err :: AWS.Error) -> Either OsvDbFetchFault DbEtag -> IO (Either OsvDbFetchFault DbEtag)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (OsvDbFetchFault -> Either OsvDbFetchFault DbEtag
forall a b. a -> Either a b
Left (TransportFault -> OsvDbFetchFault
OsvDbTransport (Error -> TransportFault
classifyAwsTransport Error
err))))
            IO (Either OsvDbFetchFault DbEtag)
-> (OsvDbCapExceeded -> IO (Either OsvDbFetchFault DbEtag))
-> IO (Either OsvDbFetchFault DbEtag)
forall (m :: * -> *) e a.
(MonadUnliftIO m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` (\(OsvDbCapExceeded Int
n) -> Either OsvDbFetchFault DbEtag -> IO (Either OsvDbFetchFault DbEtag)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (OsvDbFetchFault -> Either OsvDbFetchFault DbEtag
forall a b. a -> Either a b
Left (Int -> OsvDbFetchFault
OsvDbTooLarge Int
n)))

dbEtag :: S3.ETag -> DbEtag
dbEtag :: ETag -> DbEtag
dbEtag (S3.ETag ByteString
bytes) = Text -> DbEtag
DbEtag (ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 ByteString
bytes)

isNotFound :: AWS.Error -> Bool
isNotFound :: Error -> Bool
isNotFound = \case
    AWS.ServiceError ServiceError
se -> Status -> Int
statusCode (ServiceError
se ServiceError -> Getting Status ServiceError Status -> Status
forall s a. s -> Getting a s a -> a
^. Getting Status ServiceError Status
Lens' ServiceError Status
AWS.serviceError_status) Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
404
    Error
_ -> Bool
False

{- | A pass-through conduit that refuses to stream past the byte cap: the
enforcement behind the source's bounded download, where the declared content
length is only the fast-fail. A breach throws the confined 'OsvDbCapExceeded'
(a conduit has no value channel of its own); the adapter boundary folds it into
'OsvDbTooLarge'.
-}
cappedAt :: (MonadIO m) => Int -> ConduitT ByteString ByteString m ()
cappedAt :: forall (m :: * -> *).
MonadIO m =>
Int -> ConduitT ByteString ByteString m ()
cappedAt Int
maxBytes = Int -> ConduitT ByteString ByteString m ()
forall (m :: * -> *).
MonadIO m =>
Int -> ConduitT ByteString ByteString m ()
go Int
0
  where
    go :: Int -> ConduitT ByteString ByteString m ()
go Int
seen =
        ConduitT ByteString ByteString m (Maybe ByteString)
forall (m :: * -> *) i o. Monad m => ConduitT i o m (Maybe i)
await ConduitT ByteString ByteString m (Maybe ByteString)
-> (Maybe ByteString -> ConduitT ByteString ByteString m ())
-> ConduitT ByteString ByteString m ()
forall a b.
ConduitT ByteString ByteString m a
-> (a -> ConduitT ByteString ByteString m b)
-> ConduitT ByteString ByteString m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            Maybe ByteString
Nothing -> ConduitT ByteString ByteString m ()
forall (f :: * -> *). Applicative f => f ()
pass
            Just ByteString
chunk -> do
                let seen' :: Int
seen' = Int
seen Int -> Int -> Int
forall a. Num a => a -> a -> a
+ ByteString -> Int
BS.length ByteString
chunk
                Bool
-> ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
seen' Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maxBytes) (OsvDbCapExceeded -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
throwIO (Int -> OsvDbCapExceeded
OsvDbCapExceeded Int
maxBytes))
                ByteString -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) o i. Monad m => o -> ConduitT i o m ()
yield ByteString
chunk
                Int -> ConduitT ByteString ByteString m ()
go Int
seen'