-- SPDX-FileCopyrightText: 2026 Alexandra de Wit
--
-- SPDX-License-Identifier: MIT
{-# LANGUAGE OverloadedStrings #-}

module Ecluse.Core.Osv.Compile (
    compileOsvToSqlite,
) where

import Conduit
import Control.Monad.Catch (MonadMask)
import Data.Conduit.List qualified as CL
import Data.Time (getCurrentTime)
import Data.Time.Format.ISO8601 (iso8601Show)
import Data.Version (showVersion)
import Database.SQLite.Simple
import Katip (KatipContext, Severity (..), SimpleLogPayload, katipAddContext, logFM, ls, sl)
import Paths_ecluse (version)
import System.Directory (createDirectoryIfMissing, removeFile)
import System.FilePath ((</>))
import System.IO.Error (catchIOError)
import UnliftIO.Exception (bracket, throwIO)

import Ecluse.Core.Osv.Advisory (ExtractedOsv (..))
import Ecluse.Core.Osv.Retry (defaultOsvRetryPolicy, withOsvRetry)
import Ecluse.Core.Osv.Schema (MetaKey (..), metaTableDdl, osvDbFileName, osvSchemaEpoch, rangesTableDdl, renderMetaKey)
import Ecluse.Core.Osv.Stream (
    IngestStats (..),
    PilotIngestAborted (..),
    defaultIngestLimits,
    newOsvIngest,
    readIngestStats,
    resetIngestStats,
    streamOsvUrl,
    systemicDrop,
 )
import OpenTelemetry.Context qualified as Ctx
import OpenTelemetry.Trace.Core (SpanKind (Internal), SpanStatus (Error), TracerProvider, addAttribute, createSpan, defaultSpanArguments, endSpan, kind, makeTracer, setStatus, tracerOptions)

{- | Compile an ecosystem's OSV advisory export into the SQLite artifact and
return its path. The artifact's name, epoch stamp, and @meta@ table follow the
contract in "Ecluse.Core.Osv.Schema".
-}
compileOsvToSqlite :: (MonadResource m, MonadMask m, MonadUnliftIO m, KatipContext m) => Maybe TracerProvider -> FilePath -> Text -> String -> m FilePath
compileOsvToSqlite :: forall (m :: * -> *).
(MonadResource m, MonadMask m, MonadUnliftIO m, KatipContext m) =>
Maybe TracerProvider -> String -> Text -> String -> m String
compileOsvToSqlite Maybe TracerProvider
mTracerProvider String
outDir Text
ecosystem String
urlStr = do
    let dbFile :: String
dbFile = String
outDir String -> String -> String
</> Text -> String
osvDbFileName Text
ecosystem
        mTracer :: Maybe Tracer
mTracer = (\TracerProvider
tp -> TracerProvider -> InstrumentationLibrary -> TracerOptions -> Tracer
makeTracer TracerProvider
tp InstrumentationLibrary
"ecluse" TracerOptions
tracerOptions) (TracerProvider -> Tracer) -> Maybe TracerProvider -> Maybe Tracer
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe TracerProvider
mTracerProvider
    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
"Compiling OSV data for " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
ecosystem Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" to " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
forall a. ToText a => a -> Text
toText String
dbFile))

    -- Ensure clean state
    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Bool -> String -> IO ()
createDirectoryIfMissing Bool
True String
outDir
    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IO () -> (IOError -> IO ()) -> IO ()
forall a. IO a -> (IOError -> IO a) -> IO a
catchIOError (String -> IO ()
removeFile String
dbFile) (IO () -> IOError -> IO ()
forall a b. a -> b -> a
const (IO () -> IOError -> IO ()) -> IO () -> IOError -> IO ()
forall a b. (a -> b) -> a -> b
$ () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())

    -- The whole compile pass is one span: ecosystem and source stamped up front, the
    -- row count and drop tally once the stream settles, and a systemic-drop abort marks
    -- the span errored so an abandoned run is legible from the trace alone.
    m (Maybe Span)
-> (Maybe Span -> m ()) -> (Maybe Span -> m ()) -> m ()
forall (m :: * -> *) a b c.
MonadUnliftIO m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket
        ((Tracer -> m Span) -> Maybe Tracer -> m (Maybe Span)
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) -> Maybe a -> f (Maybe b)
traverse (\Tracer
t -> Tracer -> Context -> Text -> SpanArguments -> m Span
forall (m :: * -> *).
(MonadIO m, HasCallStack) =>
Tracer -> Context -> Text -> SpanArguments -> m Span
createSpan Tracer
t Context
Ctx.empty Text
"ecluse.pilot.osv.compile" SpanArguments
defaultSpanArguments{kind = Internal}) Maybe Tracer
mTracer)
        ((Span -> m ()) -> Maybe Span -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Span -> Maybe Timestamp -> m ()
forall (m :: * -> *). MonadIO m => Span -> Maybe Timestamp -> m ()
`endSpan` Maybe Timestamp
forall a. Maybe a
Nothing))
        ((Maybe Span -> m ()) -> m ()) -> (Maybe Span -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Maybe Span
mSpan -> do
            Maybe Span -> (Span -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe Span
mSpan ((Span -> m ()) -> m ()) -> (Span -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Span
sp -> do
                Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.ecosystem" Text
ecosystem
                Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.source_url" (String -> Text
forall a. ToText a => a -> Text
toText String
urlStr)

            m Connection
-> (Connection -> m ()) -> (Connection -> m ()) -> m ()
forall (m :: * -> *) a b c.
MonadUnliftIO m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (IO Connection -> m Connection
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Connection -> m Connection) -> IO Connection -> m Connection
forall a b. (a -> b) -> a -> b
$ String -> IO Connection
open String
dbFile) (IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> (Connection -> IO ()) -> Connection -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connection -> IO ()
close) ((Connection -> m ()) -> m ()) -> (Connection -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
                IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Connection -> IO ()
initSchema Connection
conn
                ingest <- IngestLimits -> m OsvIngest
forall (m :: * -> *). MonadIO m => IngestLimits -> m OsvIngest
newOsvIngest IngestLimits
defaultIngestLimits

                -- The fetch runs under a truncated exponential backoff (see
                -- 'Ecluse.Core.Osv.Retry'): a transient osv.dev failure is retried with
                -- jittered, capped, and count-bounded backoff rather than tight-looping, so
                -- an outage cannot get our egress IP rate-limited or banned. Batches commit
                -- incrementally, so a mid-stream drop can leave a partial table behind; each
                -- attempt therefore wipes it first and re-streams from a clean slate. (INSERT
                -- OR IGNORE alone would not suffice: a NULL introduced/fixed bound is distinct
                -- under the dedup index's uniqueness, so a re-run would duplicate those ranges.)
                -- The ingest tally is reset alongside the table so it reflects only the final
                -- attempt.
                withOsvRetry defaultOsvRetryPolicy $ do
                    resetIngestStats ingest
                    liftIO $ execute_ conn "DELETE FROM package_vulnerability_ranges"
                    runConduit $
                        streamOsvUrl mTracerProvider ingest urlStr
                            .| CL.filter ((== ecosystem) . extEcosystem)
                            .| CL.chunksOf 2000
                            .| sinkSqlite conn

                -- The stream drops an over-large or malformed advisory rather than halting, so a
                -- few poisoned records never freeze the feed. But a systemically corrupt payload
                -- must not become a fresh-looking artifact that silently omits advisories: on a
                -- systemic drop rate, abandon the run before 'writeMeta' finalises it, so a
                -- consumer keeps its last-good db instead.
                stats <- readIngestStats ingest
                forM_ mSpan $ \Span
sp -> do
                    Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.accepted" (Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statAccepted IngestStats
stats) :: Text)
                    Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.dropped_oversize" (Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statDroppedOversize IngestStats
stats) :: Text)
                    Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.dropped_malformed" (Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statDroppedMalformed IngestStats
stats) :: Text)
                when (systemicDrop stats) $ do
                    forM_ mSpan $ \Span
sp -> Span -> SpanStatus -> m ()
forall (m :: * -> *). MonadIO m => Span -> SpanStatus -> m ()
setStatus Span
sp (Text -> SpanStatus
Error Text
"systemic advisory drop rate; compile abandoned")
                    katipAddContext (dropFields ecosystem stats) $
                        logFM ErrorS (ls ("Aborting OSV compile for " <> ecosystem <> ": " <> renderDrops stats))
                    throwIO (PilotIngestAborted stats)

                rowCount <- liftIO $ writeMeta conn ecosystem urlStr
                forM_ mSpan $ \Span
sp -> Span -> Text -> Text -> m ()
forall (m :: * -> *) a.
(MonadIO m, ToAttribute a) =>
Span -> Text -> a -> m ()
addAttribute Span
sp Text
"ecluse.osv.row_count" (Int -> Text
forall b a. (Show a, IsString b) => a -> b
show Int
rowCount :: Text)
                katipAddContext (sl "row_count" rowCount <> dropFields ecosystem stats) $
                    logFM InfoS (ls ("Compiled " <> show rowCount <> " advisory ranges for " <> ecosystem <> " (" <> renderDrops stats <> ")"))

    String -> m String
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure String
dbFile

-- A one-line summary of an ingest pass's drop tally for the boot log.
renderDrops :: IngestStats -> Text
renderDrops :: IngestStats -> Text
renderDrops IngestStats
s =
    Text
"accepted "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statAccepted IngestStats
s)
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", dropped "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statDroppedOversize IngestStats
s)
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" oversize / "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show (IngestStats -> Int
statDroppedMalformed IngestStats
s)
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" malformed"

-- The drop tally as structured log fields, shared by the completion and abort lines
-- so both carry the same ecosystem/accepted/dropped shape an operator can filter on.
dropFields :: Text -> IngestStats -> SimpleLogPayload
dropFields :: Text -> IngestStats -> SimpleLogPayload
dropFields Text
ecosystem IngestStats
s =
    Text -> Text -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"ecosystem" Text
ecosystem
        SimpleLogPayload -> SimpleLogPayload -> SimpleLogPayload
forall a. Semigroup a => a -> a -> a
<> Text -> Int -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"accepted" (IngestStats -> Int
statAccepted IngestStats
s)
        SimpleLogPayload -> SimpleLogPayload -> SimpleLogPayload
forall a. Semigroup a => a -> a -> a
<> Text -> Int -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"dropped_oversize" (IngestStats -> Int
statDroppedOversize IngestStats
s)
        SimpleLogPayload -> SimpleLogPayload -> SimpleLogPayload
forall a. Semigroup a => a -> a -> a
<> Text -> Int -> SimpleLogPayload
forall a. ToJSON a => Text -> a -> SimpleLogPayload
sl Text
"dropped_malformed" (IngestStats -> Int
statDroppedMalformed IngestStats
s)

initSchema :: Connection -> IO ()
initSchema :: Connection -> IO ()
initSchema Connection
conn = do
    Connection -> Query -> IO ()
execute_ Connection
conn (Text -> Query
Query Text
rangesTableDdl)
    -- The dedup guard over a segment's five identity columns. A unique index
    -- rather than a composite PRIMARY KEY because @STRICT@ makes primary-key
    -- columns implicitly NOT NULL and the three bound columns are legitimately
    -- NULL; uniqueness behaviour is identical (INSERT OR IGNORE honours it, and
    -- NULL bounds are distinct under both forms).
    Connection -> Query -> IO ()
execute_ Connection
conn Query
"CREATE UNIQUE INDEX uq_ranges_segment ON package_vulnerability_ranges(package_name, cve_id, introduced_version, fixed_version, last_affected_version)"
    Connection -> Query -> IO ()
execute_ Connection
conn Query
"CREATE INDEX idx_package_name ON package_vulnerability_ranges(package_name)"
    -- The reader's remediation probe is an exact (name, fixed) equality; this
    -- index makes it one B-tree traversal. Additive, so epoch-neutral.
    Connection -> Query -> IO ()
execute_ Connection
conn Query
"CREATE INDEX idx_package_fixed ON package_vulnerability_ranges(package_name, fixed_version)"
    Connection -> Query -> IO ()
execute_ Connection
conn (Text -> Query
Query Text
metaTableDdl)
    Connection -> Query -> IO ()
execute_ Connection
conn (String -> Query
forall a. IsString a => String -> a
fromString (String
"PRAGMA user_version = " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall b a. (Show a, IsString b) => a -> b
show Int
osvSchemaEpoch))

-- Written once, after the stream has completed: the row count is only
-- meaningful for a complete artifact.
writeMeta :: Connection -> Text -> String -> IO Int
writeMeta :: Connection -> Text -> String -> IO Int
writeMeta Connection
conn Text
ecosystem String
urlStr = do
    now <- IO UTCTime
getCurrentTime
    counted <- query_ conn "SELECT COUNT(*) FROM package_vulnerability_ranges" :: IO [Only Int]
    let rowCount = Int -> (Only Int -> Int) -> Maybe (Only Int) -> Int
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Int
0 Only Int -> Int
forall a. Only a -> a
fromOnly ([Only Int] -> Maybe (Only Int)
forall a. [a] -> Maybe a
listToMaybe [Only Int]
counted)
    executeMany
        conn
        "INSERT INTO meta (key, value) VALUES (?, ?)"
        [ (renderMetaKey MetaPilotVersion, toText (showVersion version))
        , (renderMetaKey MetaEcosystem, ecosystem)
        , (renderMetaKey MetaBuiltAt, toText (iso8601Show now))
        , (renderMetaKey MetaSourceUrl, toText urlStr)
        , (renderMetaKey MetaRowCount, show rowCount)
        ]
    pure rowCount

sinkSqlite :: (MonadIO m) => Connection -> ConduitT [ExtractedOsv] o m ()
sinkSqlite :: forall (m :: * -> *) o.
MonadIO m =>
Connection -> ConduitT [ExtractedOsv] o m ()
sinkSqlite Connection
conn = ([ExtractedOsv] -> ConduitT [ExtractedOsv] o m ())
-> ConduitT [ExtractedOsv] o m ()
forall (m :: * -> *) i o r.
Monad m =>
(i -> ConduitT i o m r) -> ConduitT i o m ()
awaitForever (([ExtractedOsv] -> ConduitT [ExtractedOsv] o m ())
 -> ConduitT [ExtractedOsv] o m ())
-> ([ExtractedOsv] -> ConduitT [ExtractedOsv] o m ())
-> ConduitT [ExtractedOsv] o m ()
forall a b. (a -> b) -> a -> b
$ \[ExtractedOsv]
batch ->
    IO () -> ConduitT [ExtractedOsv] o m ()
forall a. IO a -> ConduitT [ExtractedOsv] o m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> ConduitT [ExtractedOsv] o m ())
-> IO () -> ConduitT [ExtractedOsv] o m ()
forall a b. (a -> b) -> a -> b
$
        Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection
-> Query
-> [(Text, Text, Maybe Text, Maybe Text, Maybe Text, Maybe Double)]
-> IO ()
forall q. ToRow q => Connection -> Query -> [q] -> IO ()
executeMany
                Connection
conn
                Query
"INSERT OR IGNORE INTO package_vulnerability_ranges (package_name, cve_id, introduced_version, fixed_version, last_affected_version, severity) VALUES (?, ?, ?, ?, ?, ?)"
                ((ExtractedOsv
 -> (Text, Text, Maybe Text, Maybe Text, Maybe Text, Maybe Double))
-> [ExtractedOsv]
-> [(Text, Text, Maybe Text, Maybe Text, Maybe Text, Maybe Double)]
forall a b. (a -> b) -> [a] -> [b]
map ExtractedOsv
-> (Text, Text, Maybe Text, Maybe Text, Maybe Text, Maybe Double)
osvToRow [ExtractedOsv]
batch)
  where
    osvToRow :: ExtractedOsv
-> (Text, Text, Maybe Text, Maybe Text, Maybe Text, Maybe Double)
osvToRow ExtractedOsv
osv = (ExtractedOsv -> Text
extPackage ExtractedOsv
osv, ExtractedOsv -> Text
extCveId ExtractedOsv
osv, ExtractedOsv -> Maybe Text
extIntroduced ExtractedOsv
osv, ExtractedOsv -> Maybe Text
extFixed ExtractedOsv
osv, ExtractedOsv -> Maybe Text
extLastAffected ExtractedOsv
osv, ExtractedOsv -> Maybe Double
extSeverity ExtractedOsv
osv)