-- SPDX-FileCopyrightText: 2026 Alexandra de Wit
--
-- SPDX-License-Identifier: MIT
-- The reads specialise here. Full laziness would float each member's rarely taken continuation out
-- of the element's continuation, and every member of a read would allocate it.
{-# OPTIONS_GHC -fno-full-laziness #-}

-- | Full and selected Simple-index reads share incremental extraction. Only full reads hash the source.
module Ecluse.Core.Registry.PyPI.Metadata (
    newPyPIMetadataReads,
    fetchPyPIManifest,
    pypiChargeFactors,
    readPyPIIndex,
    pypiIndexWalk,
    projectPyPIStream,
) where

import Data.JsonStream.TokenParser (TokenResult)
import Data.Map.Strict qualified as Map

import Ecluse.Core.Package (InvalidEntry, PackageInfo (infoVersions), PackageName)
import Ecluse.Core.Package.Filter (enforceArtifactLocations, enforceArtifactLocationsOf)
import Ecluse.Core.Registry (FetchFault (FetchUrlUnformable))
import Ecluse.Core.Registry.CachedDocument (pypiSimpleCached)
import Ecluse.Core.Registry.Exchange (chargedRead, digestingRead, formThen, withSuccessBody)
import Ecluse.Core.Registry.Json.Intern (InternTable, newInternTable, newTableKey)
import Ecluse.Core.Registry.Json.Walk (Step, readJsonWalk)
import Ecluse.Core.Registry.JsonStream (StreamResult (..))
import Ecluse.Core.Registry.Metadata (Manifest (..), MetadataError (..), VersionDoc (..), VersionRead (..), metadataResponse)
import Ecluse.Core.Registry.Metadata.Projection (streamError)
import Ecluse.Core.Registry.Origin (OriginClient (ocChargeFullRead, ocLimits, ocManager, ocToken), OriginFor, originBaseUrl)
import Ecluse.Core.Registry.PyPI.Document (SimpleDocument)
import Ecluse.Core.Registry.PyPI.Reader (fileUniqueFields, pypiWalk)
import Ecluse.Core.Registry.PyPI.Request (pypiArtifactHosts, simpleIndexRequest)
import Ecluse.Core.Registry.PyPI.Streaming (PyPIRead (..))
import Ecluse.Core.Registry.PyPI.StreamingProjection (PyPIProjection, collectField, emptyProjection, finishProjection, keepsFile)
import Ecluse.Core.Security (AllowedHostPorts, BodyLimit (MetadataBodyLimit), LimitError, Limits (progressFloor), ecosystemArtifactAuthorities, maxMetadataBytes, maxNestingDepth)
import Ecluse.Core.Server.Admission.Types (ChargeFactors (..))
import Ecluse.Core.Server.Metadata (MetadataReads, newMetadataReads)
import Ecluse.Core.Telemetry.Record (MetricsPort)
import Ecluse.Core.Telemetry.Span (TracingPort (spanMetadataDecode, spanMetadataFetch))
import Ecluse.Core.Version (Version, renderVersion)

-- | Bind one origin's reads to observers. PyPI retains no publication object for selected releases.
newPyPIMetadataReads ::
    TracingPort ->
    MetricsPort ->
    (PackageName -> MetadataError -> IO ()) ->
    (PackageName -> [InvalidEntry] -> IO ()) ->
    (PackageName -> IO ()) ->
    OriginFor posture ->
    MetadataReads posture
newPyPIMetadataReads :: forall posture.
TracingPort
-> MetricsPort
-> (PackageName -> MetadataError -> IO ())
-> (PackageName -> [InvalidEntry] -> IO ())
-> (PackageName -> IO ())
-> OriginFor posture
-> MetadataReads posture
newPyPIMetadataReads TracingPort
tracing MetricsPort
metrics PackageName -> MetadataError -> IO ()
logFailure PackageName -> [InvalidEntry] -> IO ()
logInvalid PackageName -> IO ()
logFetch =
    MetricsPort
-> (PackageName -> MetadataError -> IO ())
-> (PackageName -> [InvalidEntry] -> IO ())
-> (PackageName -> IO ())
-> (OriginClient
    -> PackageName -> IO (Either MetadataError Manifest))
-> (OriginClient
    -> PackageName -> Version -> IO (Either MetadataError VersionRead))
-> OriginFor posture
-> MetadataReads posture
forall posture.
MetricsPort
-> (PackageName -> MetadataError -> IO ())
-> (PackageName -> [InvalidEntry] -> IO ())
-> (PackageName -> IO ())
-> (OriginClient
    -> PackageName -> IO (Either MetadataError Manifest))
-> (OriginClient
    -> PackageName -> Version -> IO (Either MetadataError VersionRead))
-> OriginFor posture
-> MetadataReads posture
newMetadataReads MetricsPort
metrics PackageName -> MetadataError -> IO ()
logFailure PackageName -> [InvalidEntry] -> IO ()
logInvalid PackageName -> IO ()
logFetch (TracingPort
-> OriginClient
-> PackageName
-> IO (Either MetadataError Manifest)
fetchPyPIManifest TracingPort
tracing) (TracingPort
-> OriginClient
-> PackageName
-> Version
-> IO (Either MetadataError VersionRead)
fetchPyPIVersion TracingPort
tracing)

{- | PyPI's memory charges, above the largest read peak per source byte (3.07, boto3) and output
working set per basis byte of a realistic merge (1.24, boto3) from one meter step up.
-}
pypiChargeFactors :: ChargeFactors
pypiChargeFactors :: ChargeFactors
pypiChargeFactors = ChargeFactors{cfFullReadPermille :: Int
cfFullReadPermille = Int
4200, cfOutputPermille :: Int
cfOutputPermille = Int
1600}

-- | Fetch compact files and hash the complete decompressed source inside the response lifetime.
fetchPyPIManifest :: TracingPort -> OriginClient -> PackageName -> IO (Either MetadataError Manifest)
fetchPyPIManifest :: TracingPort
-> OriginClient
-> PackageName
-> IO (Either MetadataError Manifest)
fetchPyPIManifest TracingPort
tracing OriginClient
origin PackageName
name = do
    result <- TracingPort
-> OriginClient
-> PackageName
-> (IO ByteString
    -> IO
         (Either LimitError (StreamResult PyPIProjection, ContentDigest)))
-> IO
     (Either MetadataError (StreamResult PyPIProjection, ContentDigest))
forall r.
TracingPort
-> OriginClient
-> PackageName
-> (IO ByteString -> IO (Either LimitError r))
-> IO (Either MetadataError r)
fetchPyPIBody TracingPort
tracing OriginClient
origin PackageName
name ((IO ByteString
 -> IO (Either LimitError (StreamResult PyPIProjection)))
-> IO ByteString
-> IO
     (Either LimitError (StreamResult PyPIProjection, ContentDigest))
forall e a.
(IO ByteString -> IO (Either e a))
-> IO ByteString -> IO (Either e (a, ContentDigest))
digestingRead (TracingPort
-> OriginClient
-> PackageName
-> PyPIRead
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
decodePyPI TracingPort
tracing OriginClient
origin PackageName
name PyPIRead
FullRead) (IO ByteString
 -> IO
      (Either LimitError (StreamResult PyPIProjection, ContentDigest)))
-> (IO ByteString -> IO ByteString)
-> IO ByteString
-> IO
     (Either LimitError (StreamResult PyPIProjection, ContentDigest))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int -> IO ()) -> IO ByteString -> IO ByteString
chargedRead (OriginClient -> Int -> IO ()
ocChargeFullRead OriginClient
origin))
    pure $ do
        (streamed, digest) <- result
        (info, document) <- projectPyPIStream (ocLimits origin) name streamed
        pure
            Manifest
                { manifestInfo = enforceArtifactLocations pypiArtifactAuthorities (originBaseUrl origin) info
                , manifestRaw = fst pypiSimpleCached document
                , manifestBodyBytes = streamBytes streamed
                , manifestDigest = digest
                }

fetchPyPIBody :: TracingPort -> OriginClient -> PackageName -> (IO ByteString -> IO (Either LimitError r)) -> IO (Either MetadataError r)
fetchPyPIBody :: forall r.
TracingPort
-> OriginClient
-> PackageName
-> (IO ByteString -> IO (Either LimitError r))
-> IO (Either MetadataError r)
fetchPyPIBody TracingPort
tracing OriginClient
origin PackageName
name IO ByteString -> IO (Either LimitError r)
consume =
    Either FetchFault (BodyOutcome r) -> Either MetadataError r
forall a.
Either FetchFault (BodyOutcome a) -> Either MetadataError a
metadataResponse
        (Either FetchFault (BodyOutcome r) -> Either MetadataError r)
-> IO (Either FetchFault (BodyOutcome r))
-> IO (Either MetadataError r)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TracingPort -> forall a. PackageName -> IO a -> IO a
spanMetadataFetch
            TracingPort
tracing
            PackageName
name
            ((UrlFormationError -> FetchFault)
-> (Request -> IO (Either FetchFault (BodyOutcome r)))
-> Either UrlFormationError Request
-> IO (Either FetchFault (BodyOutcome r))
forall fault a.
(UrlFormationError -> fault)
-> (Request -> IO (Either fault a))
-> Either UrlFormationError Request
-> IO (Either fault a)
formThen UrlFormationError -> FetchFault
FetchUrlUnformable (Manager
-> ProgressFloor
-> (IO ByteString -> IO (Either LimitError r))
-> Request
-> IO (Either FetchFault (BodyOutcome r))
forall a.
Manager
-> ProgressFloor
-> (IO ByteString -> IO (Either LimitError a))
-> Request
-> IO (Either FetchFault (BodyOutcome a))
withSuccessBody (OriginClient -> Manager
ocManager OriginClient
origin) (Limits -> ProgressFloor
progressFloor (OriginClient -> Limits
ocLimits OriginClient
origin)) IO ByteString -> IO (Either LimitError r)
consume) (Text
-> Maybe ClientCredential
-> PackageName
-> Either UrlFormationError Request
simpleIndexRequest (OriginClient -> Text
originBaseUrl OriginClient
origin) (OriginClient -> Maybe ClientCredential
ocToken OriginClient
origin) PackageName
name))

decodePyPI :: TracingPort -> OriginClient -> PackageName -> PyPIRead -> IO ByteString -> IO (Either LimitError (StreamResult PyPIProjection))
decodePyPI :: TracingPort
-> OriginClient
-> PackageName
-> PyPIRead
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
decodePyPI TracingPort
tracing OriginClient
origin PackageName
name PyPIRead
mode = TracingPort -> forall a. PackageName -> IO a -> IO a
spanMetadataDecode TracingPort
tracing PackageName
name (IO (Either LimitError (StreamResult PyPIProjection))
 -> IO (Either LimitError (StreamResult PyPIProjection)))
-> (IO ByteString
    -> IO (Either LimitError (StreamResult PyPIProjection)))
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Limits
-> PackageName
-> PyPIRead
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
readPyPIIndex (OriginClient -> Limits
ocLimits OriginClient
origin) PackageName
name PyPIRead
mode

-- | Walk an index's chunks with the production field policy, over a table keyed afresh for the read.
readPyPIIndex :: Limits -> PackageName -> PyPIRead -> IO ByteString -> IO (Either LimitError (StreamResult PyPIProjection))
readPyPIIndex :: Limits
-> PackageName
-> PyPIRead
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
readPyPIIndex Limits
limits PackageName
name PyPIRead
mode IO ByteString
readChunk = do
    table <- SipKey -> [Text] -> InternTable
newInternTable (SipKey -> [Text] -> InternTable)
-> IO SipKey -> IO ([Text] -> InternTable)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO SipKey
newTableKey IO ([Text] -> InternTable) -> IO [Text] -> IO InternTable
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> [Text] -> IO [Text]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Text]
fileUniqueFields
    readJsonWalk (MetadataBodyLimit (maxMetadataBytes limits)) (pypiIndexWalk limits name mode table) readChunk

-- | The production Simple-index walk over a caller's intern table.
pypiIndexWalk :: Limits -> PackageName -> PyPIRead -> InternTable -> TokenResult -> Step PyPIProjection
pypiIndexWalk :: Limits
-> PackageName
-> PyPIRead
-> InternTable
-> TokenResult
-> Step PyPIProjection
pypiIndexWalk Limits
limits PackageName
name PyPIRead
mode InternTable
table = Int
-> PyPIRead
-> (PyPIProjection
    -> PyPIField -> Either LimitError PyPIProjection)
-> (PyPIProjection -> Bool)
-> InternTable
-> PyPIProjection
-> TokenResult
-> Step PyPIProjection
forall s.
Int
-> PyPIRead
-> (s -> PyPIField -> Either LimitError s)
-> (s -> Bool)
-> InternTable
-> s
-> TokenResult
-> Step s
pypiWalk (Limits -> Int
maxNestingDepth Limits
limits) PyPIRead
mode (Limits
-> PyPIRead
-> PyPIProjection
-> PyPIField
-> Either LimitError PyPIProjection
collectField Limits
limits PyPIRead
mode) PyPIProjection -> Bool
keepsFile InternTable
table (PackageName -> PyPIProjection
emptyProjection PackageName
name)

fetchPyPIVersion :: TracingPort -> OriginClient -> PackageName -> Version -> IO (Either MetadataError VersionRead)
fetchPyPIVersion :: TracingPort
-> OriginClient
-> PackageName
-> Version
-> IO (Either MetadataError VersionRead)
fetchPyPIVersion TracingPort
tracing OriginClient
origin PackageName
name Version
version = do
    result <- TracingPort
-> OriginClient
-> PackageName
-> (IO ByteString
    -> IO (Either LimitError (StreamResult PyPIProjection)))
-> IO (Either MetadataError (StreamResult PyPIProjection))
forall r.
TracingPort
-> OriginClient
-> PackageName
-> (IO ByteString -> IO (Either LimitError r))
-> IO (Either MetadataError r)
fetchPyPIBody TracingPort
tracing OriginClient
origin PackageName
name (TracingPort
-> OriginClient
-> PackageName
-> PyPIRead
-> IO ByteString
-> IO (Either LimitError (StreamResult PyPIProjection))
decodePyPI TracingPort
tracing OriginClient
origin PackageName
name (PackageName -> Text -> PyPIRead
SelectedRead PackageName
name (Version -> Text
renderVersion Version
version)))
    pure $ do
        streamed <- result
        (info, _) <- projectPyPIStream (ocLimits origin) name streamed
        pure
            VersionRead
                { vrVersion = do
                    details <- Map.lookup (renderVersion version) (infoVersions info)
                    located <- enforceArtifactLocationsOf pypiArtifactAuthorities (originBaseUrl origin) details
                    pure VersionDoc{vdDetails = located, vdRaw = Nothing}
                , vrBodyBytes = streamBytes streamed
                , vrUpstreamLatest = Nothing
                }

-- | Finish both read modes without separating typed files from their source coordinates.
projectPyPIStream :: Limits -> PackageName -> StreamResult PyPIProjection -> Either MetadataError (PackageInfo, SimpleDocument)
projectPyPIStream :: Limits
-> PackageName
-> StreamResult PyPIProjection
-> Either MetadataError (PackageInfo, SimpleDocument)
projectPyPIStream Limits
limits PackageName
name StreamResult PyPIProjection
streamed =
    (ParseError -> MetadataError)
-> Either ParseError PyPIProjection
-> Either MetadataError PyPIProjection
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 (Limits -> ParseError -> MetadataError
streamError Limits
limits) (StreamResult PyPIProjection -> Either ParseError PyPIProjection
forall a. StreamResult a -> Either ParseError a
streamValue StreamResult PyPIProjection
streamed) Either MetadataError PyPIProjection
-> (PyPIProjection
    -> Either MetadataError (PackageInfo, SimpleDocument))
-> Either MetadataError (PackageInfo, SimpleDocument)
forall a b.
Either MetadataError a
-> (a -> Either MetadataError b) -> Either MetadataError b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= PackageName
-> PyPIProjection
-> Either MetadataError (PackageInfo, SimpleDocument)
finishProjection PackageName
name

pypiArtifactAuthorities :: AllowedHostPorts
pypiArtifactAuthorities :: AllowedHostPorts
pypiArtifactAuthorities = [Text] -> AllowedHostPorts
ecosystemArtifactAuthorities [Text]
pypiArtifactHosts