Browse.hs

{- | Bounded inspection reads. Stream names use stable UTF-8 byte order;
category enumeration retains the database's category collation. Cursors are
exclusive and describe live pages rather than a cross-request snapshot.
-}
module Kiroku.Metrics.Browse (
    StoreBrowser (..),
    storeBrowser,
    storeBrowserWith,
    BrowseLimits,
    BrowseLimitsError (..),
    mkBrowseLimits,
    defaultLimit,
    maxLimit,
    defaultBrowseLimits,
    ReadDirection (..),
    browseApp,
    browseNotConfiguredApp,
    streamInfoToJSON,
) where

import Data.Aeson (Value, encode, object, toJSON, (.=))
import Data.ByteString qualified as BS
import Data.Int (Int32, Int64)
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.Encoding qualified as TE
import Data.UUID qualified as UUID
import Data.Vector (Vector)
import Data.Vector qualified as V
import Effectful (Eff, IOE)
import Effectful.Error.Static (Error)
import Kiroku.Metrics.JSON (errorResponse, jsonResponse, storeErrorResponse)
import Kiroku.Metrics.WebSocket (recordedEventToJSONResolved)
import Kiroku.Store
import Network.HTTP.Types (Status, status200, status400, status404, status405)
import Network.Wai (Application, Request, Response, mapResponseHeaders, pathInfo, queryString, requestMethod, responseHeaders, responseLBS, responseStatus)
import Text.Read (readMaybe)

data StoreBrowser = StoreBrowser
    { runStoreRead :: forall a. Eff '[Store, Error StoreError, IOE] a -> IO (Either StoreError a)
    , limits :: !BrowseLimits
    }

data BrowseLimits = BrowseLimits {defaultLimit :: !Int, maxLimit :: !Int}
    deriving stock (Eq, Show)
data BrowseLimitsError = InvalidBrowseLimits !Int !Int deriving stock (Eq, Show)
mkBrowseLimits :: Int -> Int -> Either BrowseLimitsError BrowseLimits
mkBrowseLimits def cap
    | 1 <= def && def <= cap && cap <= 1000 = Right (BrowseLimits def cap)
    | otherwise = Left (InvalidBrowseLimits def cap)
defaultBrowseLimits :: BrowseLimits
defaultBrowseLimits = BrowseLimits 100 1000
storeBrowser :: KirokuStore -> StoreBrowser
storeBrowser = storeBrowserWith defaultBrowseLimits
storeBrowserWith :: BrowseLimits -> KirokuStore -> StoreBrowser
storeBrowserWith lims store = StoreBrowser (runStoreIO store) lims

data ReadDirection = ReadForward | ReadBackward deriving stock (Eq, Show)
type ReadProgram a = Eff '[Store, Error StoreError, IOE] a
type Failure = (Status, Text, Text, Maybe Value)

-- | Preserve GET status/headers for HEAD. Other methods never run a read.
readMethods :: (Request -> IO Response) -> Application
readMethods action req respond
    | requestMethod req == "GET" = action req >>= respond
    | requestMethod req == "HEAD" = do
        response <- action req
        respond (responseLBS (responseStatus response) (responseHeaders response) "")
    | otherwise =
        respond $
            mapResponseHeaders (("Allow", "GET, HEAD") :) $
                errorResponse status405 "method_not_allowed" "Only GET and HEAD are supported." Nothing

browseNotConfiguredApp :: Application
browseNotConfiguredApp = readMethods $ \_ -> pure (errorResponse status404 "store_browsing_not_configured" "Store browsing is not configured." Nothing)

browseApp :: StoreBrowser -> Application
browseApp browser = readMethods $ \req -> case route browser req of
    Left (status, code, message, details) -> pure (errorResponse status code message details)
    Right program ->
        runStoreRead browser program >>= \case
            Left err -> pure (storeErrorResponse "store_unavailable" err)
            Right (Left (status, code, message, details)) -> pure (errorResponse status code message details)
            Right (Right value) -> pure (jsonResponse status200 (encode value))

route :: StoreBrowser -> Request -> Either Failure (ReadProgram (Either Failure Value))
route browser req = case pathInfo req of
    ["streams"] -> do
        limit <- pageLimit
        category <- fmap CategoryName <$> textParam "category"
        prefix <- textParam "prefix"
        cursor <- fmap StreamName <$> textParam "from"
        size <- browseSize limit
        pure $ do
            rows <- listStreams category prefix cursor size
            pure $ Right $ pageJSON limit streamInfoToJSON (\row -> let StreamName name = row.name in AesonText name) rows
    ["streams", name] -> do
        stream <- validStream name
        pure $ getStream stream >>= pure . maybe (Left (missing "stream_not_found" "Stream not found.")) (Right . streamInfoToJSON)
    ["streams", name, "events"] -> do
        stream <- validStream name
        limit <- pageLimit
        cursor <- position
        direction <- readDirection
        pure $
            getStream stream >>= \case
                Nothing -> pure (Left (missing "stream_not_found" "Stream not found."))
                Just _ ->
                    Right
                        <$> eventPage
                            limit
                            (\event -> let StreamVersion v = event.streamVersion in AesonNumber v)
                            (case direction of ReadForward -> readStreamForward stream (StreamVersion cursor) (overfetch limit); ReadBackward -> readStreamBackward stream (StreamVersion cursor) (overfetch limit))
    ["categories"] -> do
        limit <- pageLimit
        cursor <- fmap CategoryName <$> textParam "from"
        size <- browseSize limit
        pure $ do
            rows <- listCategories cursor size
            pure $ Right $ pageJSON limit (\(CategoryName name) -> object ["name" .= name]) (\(CategoryName name) -> AesonText name) rows
    ["categories", name, "events"] -> do
        validText "category" name
        limit <- pageLimit
        cursor <- position
        -- Category reads have one supported direction; do not silently ignore it.
        direction <- readDirection
        if direction == ReadBackward
            then Left (invalid "direction" "backward" "Category events support forward reads.")
            else
                pure $ Right <$> eventPage limit globalCursor (readCategory (CategoryName name) (GlobalPosition cursor) (overfetch limit))
    ["events"] -> do
        limit <- pageLimit
        cursor <- position
        direction <- readDirection
        pure $
            Right
                <$> eventPage
                    limit
                    globalCursor
                    (case direction of ReadForward -> readAllForward (GlobalPosition cursor) (overfetch limit); ReadBackward -> readAllBackward (GlobalPosition cursor) (overfetch limit))
    ["events", rawId] -> case UUID.fromText rawId of
        Nothing -> Left (status400, "invalid_event_id", "The event id must be a UUID.", Nothing)
        Just uuid ->
            pure $
                getEvent (EventId uuid) >>= \case
                    Nothing -> pure (Left (missing "event_not_found" "Event not found."))
                    Just event -> do
                        mapping <- lookupStreamNames [event.originalStreamId]
                        pure (Right (recordedEventToJSONResolved mapping event))
    _ -> Left (missing "not_found" "Not found.")
  where
    pageLimit = do
        value <- rawParam req "limit"
        case value of
            Nothing -> Right browser.limits.defaultLimit
            Just raw -> case decimal raw of
                Just n | 1 <= n && n <= toInteger browser.limits.maxLimit -> Right (fromInteger n)
                _ -> Left (invalid "limit" raw "Expected a decimal integer within the configured page limit.")
    textParam key = do
        value <- rawParam req key
        mapM_ (validText key) value
        pure value
    position = do
        value <- rawParam req "from"
        case value of
            Nothing -> Right 0
            Just raw -> case decimal raw of
                Just n | n <= toInteger (maxBound :: Int64) -> Right (fromInteger n)
                _ -> Left (invalid "from" raw "Expected a non-negative Int64 decimal integer.")
    readDirection =
        rawParam req "direction" >>= \case
            Nothing -> Right ReadForward
            Just "forward" -> Right ReadForward
            Just "backward" -> Right ReadBackward
            Just raw -> Left (invalid "direction" raw "Expected forward or backward.")

-- Closed bounds make over-fetch safe before conversion to the store's Int32.
overfetch :: Int -> Int32
overfetch limit = fromIntegral (limit + 1)
browseSize :: Int -> Either Failure BrowsePageSize
browseSize limit = case mkBrowsePageSize (limit + 1) of
    Right size -> Right size
    Left _ -> Left (invalid "limit" (T.pack (show limit)) "Invalid page limit.")

data Cursor = AesonText Text | AesonNumber Int64
cursorJSON :: Cursor -> Value
cursorJSON (AesonText text) = toJSON text
cursorJSON (AesonNumber n) = toJSON n

pageJSON :: Int -> (a -> Value) -> (a -> Cursor) -> Vector a -> Value
pageJSON limit encodeItem cursor rows =
    object $ ["items" .= V.map encodeItem items] <> ["next_cursor" .= cursorJSON (cursor (V.last items)) | V.length rows > limit && not (V.null items)]
  where
    items = V.take limit rows

eventPage :: Int -> (RecordedEvent -> Cursor) -> ReadProgram (Vector RecordedEvent) -> ReadProgram Value
eventPage limit cursor readPage = do
    rows <- readPage
    mapping <- resolveNames (V.take limit rows)
    pure (pageJSON limit (recordedEventToJSONResolved mapping) cursor rows)
resolveNames :: Vector RecordedEvent -> ReadProgram (Map StreamId StreamName)
resolveNames rows
    | V.null rows = pure Map.empty
    | otherwise = lookupStreamNames (Set.toList (Set.fromList (map (\event -> event.originalStreamId) (V.toList rows))))
globalCursor :: RecordedEvent -> Cursor
globalCursor event = let GlobalPosition n = event.globalPosition in AesonNumber n

streamInfoToJSON :: StreamInfo -> Value
streamInfoToJSON stream =
    object
        [ "stream_id" .= (let StreamId n = stream.id in n)
        , "name" .= (let StreamName name = stream.name in name)
        , "category" .= (let CategoryName category = categoryName stream.name in category)
        , "version" .= (let StreamVersion n = stream.version in n)
        , "created_at" .= stream.createdAt
        , "deleted_at" .= stream.deletedAt
        , "truncate_before" .= (let StreamVersion n = stream.truncateBefore in n)
        ]

missing :: Text -> Text -> Failure
missing code message = (status404, code, message, Nothing)
invalid :: Text -> Text -> Text -> Failure
invalid key raw reason = (status400, "invalid_query_parameter", "Invalid query parameter.", Just (object ["parameter" .= key, "value" .= raw, "reason" .= reason]))
rawParam :: Request -> BS.ByteString -> Either Failure (Maybe Text)
rawParam req key = case lookup key (queryString req) of
    Nothing -> Right Nothing
    Just Nothing -> Right (Just "")
    Just (Just bytes) -> case TE.decodeUtf8' bytes of
        Right value -> Right (Just value)
        Left _ -> Left (invalid (TE.decodeUtf8 key) "<invalid UTF-8>" "Expected valid UTF-8.")
validText :: BS.ByteString -> Text -> Either Failure ()
validText key value
    | T.any (== '\0') value || BS.length (TE.encodeUtf8 value) > 512 = Left (invalid (TE.decodeUtf8 key) value "Expected at most 512 UTF-8 bytes without NUL.")
    | otherwise = Right ()
validStream :: Text -> Either Failure StreamName
validStream value = case validateStreamName (StreamName value) of
    Right () | not (T.any (== '\0') value) -> Right (StreamName value)
    _ -> Left (status400, "invalid_stream_name", "Invalid stream name.", Just (object ["stream_name" .= value]))
decimal :: Text -> Maybe Integer
decimal value
    | T.null value || not (T.all (\c -> '0' <= c && c <= '9') value) = Nothing
    | T.length value > 20 = Nothing
    | otherwise = readMaybe (T.unpack value)