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

{- | Byte limits for advisory streams. Callers own the breach effect, so ingestion
and runtime downloads can share the traversal without sharing error types.
-}
module Ecluse.Core.Stream (boundBytes, boundLines) where

import Conduit (ConduitT, await, yield)
import Data.ByteString qualified as BS

{- | Preserve chunks up to the byte cap. On breach, pass the observed byte count
to the action and stop without yielding the excess chunk, even if the action returns.
-}
boundBytes :: (Monad m) => Int -> (Int -> m ()) -> ConduitT ByteString ByteString m ()
boundBytes :: forall (m :: * -> *).
Monad m =>
Int -> (Int -> m ()) -> ConduitT ByteString ByteString m ()
boundBytes Int
cap Int -> m ()
onBreach = 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 ->
                let seen' :: Int
seen' = Int
seen Int -> Int -> Int
forall a. Num a => a -> a -> a
+ ByteString -> Int
BS.length ByteString
chunk
                 in if Int
seen' Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
cap
                        then m () -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) a.
Monad m =>
m a -> ConduitT ByteString ByteString m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (Int -> m ()
onBreach Int
seen')
                        else ByteString -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) o i. Monad m => o -> ConduitT i o m ()
yield ByteString
chunk ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
forall a b.
ConduitT ByteString ByteString m a
-> ConduitT ByteString ByteString m b
-> ConduitT ByteString ByteString m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Int -> ConduitT ByteString ByteString m ()
go Int
seen'

{- | Split a stream into lines without their newline, holding at most @cap@ bytes of one line. A
longer line passes its length to the action and stops the stream, even if the action returns.
-}
boundLines :: (Monad m) => Int -> (Int -> m ()) -> ConduitT ByteString ByteString m ()
boundLines :: forall (m :: * -> *).
Monad m =>
Int -> (Int -> m ()) -> ConduitT ByteString ByteString m ()
boundLines Int
cap Int -> m ()
onBreach = ByteString -> ConduitT ByteString ByteString m ()
go ByteString
BS.empty
  where
    go :: ByteString -> ConduitT ByteString ByteString m ()
go ByteString
pending =
        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 -> Bool
-> ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (ByteString -> Bool
BS.null ByteString
pending) (ByteString -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) o i. Monad m => o -> ConduitT i o m ()
yield ByteString
pending)
            Just ByteString
chunk -> ByteString -> ConduitT ByteString ByteString m ()
split (ByteString
pending ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
chunk)
    split :: ByteString -> ConduitT ByteString ByteString m ()
split ByteString
buffer = case Word8 -> ByteString -> Maybe Int
BS.elemIndex Word8
newline ByteString
buffer of
        Just Int
end
            | Int
end Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
cap -> m () -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) a.
Monad m =>
m a -> ConduitT ByteString ByteString m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (Int -> m ()
onBreach Int
end)
            | Bool
otherwise -> ByteString -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) o i. Monad m => o -> ConduitT i o m ()
yield (Int -> ByteString -> ByteString
BS.take Int
end ByteString
buffer) ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
-> ConduitT ByteString ByteString m ()
forall a b.
ConduitT ByteString ByteString m a
-> ConduitT ByteString ByteString m b
-> ConduitT ByteString ByteString m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> ByteString -> ConduitT ByteString ByteString m ()
split (Int -> ByteString -> ByteString
BS.drop (Int
end Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) ByteString
buffer)
        Maybe Int
Nothing
            | ByteString -> Int
BS.length ByteString
buffer Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
cap -> m () -> ConduitT ByteString ByteString m ()
forall (m :: * -> *) a.
Monad m =>
m a -> ConduitT ByteString ByteString m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (Int -> m ()
onBreach (ByteString -> Int
BS.length ByteString
buffer))
            | Bool
otherwise -> ByteString -> ConduitT ByteString ByteString m ()
go ByteString
buffer
    newline :: Word8
newline = Word8
10