Read.hs

module Kiroku.Store.Read
  ( readStreamForward,
    readStreamForwardStream,
    readStreamBackward,
    readAllForward,
    readAllBackward,
    visibleGlobalHeadPosition,
    readCategory,
    getStream,
    getStreamWithHead,
    listStreams,
    listCategories,
    getEvent,
    lookupStreamId,
    eventExistsInStream,
    lookupStreamName,
    lookupStreamNames,
  )
where

import Control.Lens ((^.))
import Data.Generics.Labels ()
import Data.Int (Int32)
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Text (Text)
import Data.Vector (Vector)
import Data.Vector qualified as V
import Effectful (Eff, (:>))
import Effectful.Dispatch.Dynamic (send)
import GHC.Stack (HasCallStack)
import Kiroku.Store.Effect (Store (..))
import Kiroku.Store.Types
import Streamly.Data.Stream (Stream)
import Streamly.Data.Stream qualified as Stream

-- | Read events from a named stream in forward (ascending version)
-- order.
--
-- The cursor is exclusive: events with @streamVersion > startVer@ are
-- returned. To read the entire stream from the beginning, pass
-- @'StreamVersion' 0@. Returns an empty vector for nonexistent or
-- soft-deleted streams. The @limit@ caps the batch size; pass a large
-- value for \"read everything\".
readStreamForward ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  StreamVersion ->
  Int32 ->
  Eff es (Vector RecordedEvent)
readStreamForward name startVer limit = send (ReadStreamForward name startVer limit)

-- | Forward read a single stream as a constant-memory Streamly 'Stream'.
--
-- The streaming sibling of 'readStreamForward'. Identical SQL path and identical
-- error semantics: this function dispatches 'readStreamForward' repeatedly with
-- the supplied @pageSize@ as the per-call limit, advancing the exclusive
-- 'StreamVersion' cursor across pages until a page comes back shorter than
-- @pageSize@. If the stream length is an exact multiple of @pageSize@, one
-- final empty-page probe is still required because the wrapper does not know
-- the total stream length up front.
--
-- The exclusive-cursor convention is preserved end-to-end: passing
-- @'StreamVersion' 0@ reads from the first event in the stream. Empty and
-- nonexistent streams terminate the stream immediately with zero elements.
--
-- The recommended @pageSize@ is @256@. Callers reading very wide events (large
-- payloads / metadata) should pass a smaller value to keep per-page memory
-- bounded; callers reading very long streams of small events may pass a larger
-- value to reduce round-trip count.
readStreamForwardStream ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  StreamVersion ->
  Int32 ->
  Stream (Eff es) RecordedEvent
readStreamForwardStream name startVer pageSize =
  Stream.concatMap fromVector pages
  where
    pages = Stream.unfoldrM nextPage (Just startVer)
    nextPage Nothing = pure Nothing
    nextPage (Just cursor) = do
      events <- readStreamForward name cursor pageSize
      if V.null events
        then pure Nothing
        else do
          let nextState
                | V.length events < fromIntegral pageSize = Nothing
                | otherwise = Just (V.last events ^. #streamVersion)
          pure (Just (events, nextState))

    fromVector :: Vector a -> Stream (Eff es) a
    fromVector v = Stream.unfoldr (\i -> fmap (\x -> (x, i + 1)) (v V.!? i)) 0

-- | Read events from a named stream in backward (descending version)
-- order.
--
-- The cursor is exclusive: events with @streamVersion < startVer@ are
-- returned (events older than the cursor). To read the entire stream from
-- the latest event backward, pass @'StreamVersion' 0@. The interpreter maps
-- 0 to the maximum cursor value, so it never collides with a real stream
-- version; versions start at 1. Returns an empty vector for nonexistent or
-- soft-deleted streams.
readStreamBackward ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  StreamVersion ->
  Int32 ->
  Eff es (Vector RecordedEvent)
readStreamBackward name startVer limit = send (ReadStreamBackward name startVer limit)

-- | Read events from the global @$all@ stream in forward
-- ('GlobalPosition'-ascending) order.
--
-- Cursor exclusive: events with @globalPosition > startPos@ are returned.
-- @$all@ contains every event ever appended (including events from
-- soft-deleted streams; they survive in @$all@ even after their owning
-- stream is hidden). Hard-deleted streams' events do /not/ appear in
-- @$all@. The seed row at @globalPosition = 0@ is internal and is never
-- returned.
readAllForward ::
  (HasCallStack, Store :> es) =>
  GlobalPosition ->
  Int32 ->
  Eff es (Vector RecordedEvent)
readAllForward startPos limit = send (ReadAllForward startPos limit)

-- | Read events from the global @$all@ stream in backward
-- ('GlobalPosition'-descending) order.
--
-- Cursor exclusive. To start from the most recent event, pass
-- @'GlobalPosition' 0@. The interpreter maps 0 to the maximum cursor value,
-- so it never collides with a real global position; positions start at 1.
readAllBackward ::
  (HasCallStack, Store :> es) =>
  GlobalPosition ->
  Int32 ->
  Eff es (Vector RecordedEvent)
readAllBackward startPos limit = send (ReadAllBackward startPos limit)

-- | Return the greatest 'GlobalPosition' still visible in the global @$all@
-- stream, or @'GlobalPosition' 0@ when no event remains.
--
-- Logical truncation and soft deletion leave @$all@ junctions intact, so neither
-- changes this value. Hard deletion removes junctions and can make the visible
-- head move backward or return to zero. This differs from
-- 'Kiroku.Store.Subscription.Types.storePosition' in
-- 'Kiroku.Store.Subscription.Types.SubscriptionCheckpointInventory', which is
-- the authoritative append frontier and remains monotonic after hard deletion.
--
-- The value is observed by one database statement. A concurrent append or hard
-- deletion may change the visible head immediately after this function returns;
-- the result does not retain a database snapshot.
visibleGlobalHeadPosition ::
  (HasCallStack, Store :> es) =>
  Eff es GlobalPosition
visibleGlobalHeadPosition = send GetVisibleGlobalHeadPosition

-- | Read events whose source stream's category prefix matches the given
-- 'CategoryName', in 'GlobalPosition' order.
--
-- The category is the substring of a 'StreamName' before the first @-@:
-- @StreamName "orders-1"@ has @CategoryName "orders"@. Linked events
-- appear at their /source/ position; the category is the source's
-- category, not the link target's.
readCategory ::
  (HasCallStack, Store :> es) =>
  CategoryName ->
  GlobalPosition ->
  Int32 ->
  Eff es (Vector RecordedEvent)
readCategory cat startPos limit = send (ReadCategoryForward cat startPos limit)

-- | Query stream metadata.
--
-- Returns 'Just' for both live and soft-deleted streams (with @deletedAt@
-- populated). Returns 'Nothing' for hard-deleted streams and streams that
-- have never been created.
getStream ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  Eff es (Maybe StreamInfo)
getStream name = send (GetStream name)

-- | Read metadata and the newest surviving /originated/ event's global
-- position in one SQL statement, from one database snapshot.
--
-- Outer 'Nothing' means absent or hard-deleted. Inner 'Nothing' means the
-- stream exists without surviving originated events, including empty and
-- link-only streams. @$all@ always has an inner 'Nothing': it aggregates events
-- but originates none. Links advance the stream version without advancing this
-- head. Soft deletion and logical truncation preserve the originated head;
-- physical retention can remove it. This is neither the store-wide visible
-- head nor the monotonically allocated append frontier.
--
-- For an origin-only stream with retained required history, capture this pair,
-- require @version >= N@ and @Just head@, then wait for the projection cursor to
-- reach @head@. Waiting and timeouts belong to the consumer. The observation does
-- not lock history or freeze subsequent appends; later physical removal can
-- require a timeout. A positive linked version gives no such guarantee.
--
-- Only this opt-in operation pays for the indexed head probe. 'getStream' and
-- 'StreamInfo' retain their existing representation and cost.
getStreamWithHead ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  Eff es (Maybe (StreamInfo, Maybe GlobalPosition))
getStreamWithHead name = send (GetStreamWithHead name)

-- | Look up a stream's surrogate id by name.
--
-- Returns 'Just' the 'StreamId' for both live and soft-deleted streams (mirroring
-- 'getStream'\'s soft-delete behavior). Returns 'Nothing' for streams that have
-- never been created and for streams that have been hard-deleted.
--
-- This is a lighter-weight alternative to 'getStream' when the caller only needs
-- the surrogate id: it decodes one @int8@ column instead of the five columns
-- that 'StreamInfo' carries. Equivalent to projecting @info ^. #id@ from a
-- successful 'getStream' result, but cheaper.
lookupStreamId ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  Eff es (Maybe StreamId)
lookupStreamId name = send (LookupStreamId name)

-- | Return 'True' when the supplied event id is linked into the supplied live
-- stream, and 'False' when the event id is absent, belongs only to another stream,
-- or the stream is nonexistent or soft-deleted.
--
-- This is a point lookup for idempotency checks. Prefer it over scanning a stream
-- when the caller already has the event id it is looking for.
eventExistsInStream ::
  (HasCallStack, Store :> es) =>
  StreamName ->
  EventId ->
  Eff es Bool
eventExistsInStream name eid = send (EventExistsInStream name eid)

-- | Resolve a batch of surrogate 'StreamId's to their 'StreamName's in a single
-- round trip, returning a 'Map' that omits any id which does not name an existing
-- stream (hard-deleted or never created). Live and soft-deleted streams are
-- included, mirroring 'lookupStreamId'.
--
-- This is the inverse of 'lookupStreamId' and the supported way to recover the
-- human-readable source stream for events obtained from /fan-in/ reads — the
-- global @$all@ stream, 'readCategory', the "Kiroku.Store.Causation" queries, and
-- subscriptions — where each 'RecordedEvent' carries only its surrogate
-- @originalStreamId@. Collect the distinct ids from a batch and resolve them once,
-- rather than paying a round trip per event:
--
-- @
-- let ids = 'Data.List.nub' (map (^. #originalStreamId) events)
-- names <- 'lookupStreamNames' ids
-- -- names '!?' (event '^.' #originalStreamId) :: Maybe StreamName
-- @
--
-- Passing @[]@ returns an empty map without any database round trip because
-- the interpreter short-circuits.
lookupStreamNames ::
  (HasCallStack, Store :> es) =>
  [StreamId] ->
  Eff es (Map StreamId StreamName)
lookupStreamNames sids = send (LookupStreamNames sids)

-- | Resolve a single surrogate 'StreamId' to its 'StreamName', or 'Nothing' if
-- no such stream exists. A convenience wrapper over 'lookupStreamNames'; prefer
-- the batch form when resolving the ids of a whole read batch, to avoid one round
-- trip per id.
lookupStreamName ::
  (HasCallStack, Store :> es) =>
  StreamId ->
  Eff es (Maybe StreamName)
lookupStreamName sid = Map.lookup sid <$> lookupStreamNames [sid]

-- | Page streams in stable UTF-8 byte order, independent of database locale.
-- Category equality includes its bare name; prefix matching is literal, including
-- @%@ and @_@. Use the last returned name as the exclusive cursor. Soft-deleted
-- streams remain visible, hard-deleted streams do not. Only the reserved row is
-- excluded, so @$all-x@ and empty categories are valid. A short page proves current
-- exhaustion. Concurrent lifecycle changes are not a snapshot-isolation guarantee.
-- TypeID timestamps give generation order, not event/commit order.
listStreams :: (HasCallStack, Store :> es) => Maybe CategoryName -> Maybe Text -> Maybe StreamName -> BrowsePageSize -> Eff es (Vector StreamInfo)
listStreams category prefix after limit = send (ListStreams category prefix after limit)

-- | Distinct categories, in deployment locale order, after an exclusive cursor.
listCategories :: (HasCallStack, Store :> es) => Maybe CategoryName -> BrowsePageSize -> Eff es (Vector CategoryName)
listCategories after limit = send (ListCategories after limit)

-- | An event as seen in the global log, using the configured typed decode hook.
getEvent :: (HasCallStack, Store :> es) => EventId -> Eff es (Maybe RecordedEvent)
getEvent eid = send (GetEvent eid)