WebSocket.hs

{-# LANGUAGE ScopedTypeVariables #-}

{- | The WebSocket endpoint for the Kiroku metrics server.

Fills the IP-3 seam EP-2 left stubbed ('Kiroku.Metrics.Server.stubWebSocketApp')
with a real 'WS.ServerApp' that dispatches on the request path:

  * @\/ws\/metrics@ — the /metrics channel/: a 'MetricsSnapshot' on connect,
    then a fresh snapshot every @wsPushIntervalUs@ microseconds; @ping@ → @pong@.
  * @\/ws\/events@ — the /event channel/: after a @subscribe_events@ message, a
    JSON message per appended 'RecordedEvent' in global-position order, live.
    Optionally replays history from a chosen @from_position@ and/or restricts to
    a single @category@.

The event tail is built on the /public/ 'EventPublisher' broadcast
('subscribePublisher') for live "from-now" delivery and the public effectful
reads ('readAllForward' / 'readCategory' via 'runStoreIO') for replay and
category filtering. It creates /no persistent subscription/ and writes nothing to
the @subscriptions@ checkpoint table — transient watchers leave no trace. See the
plan's Decision Log.

This module owns IP-4: the 'ClientMessage' / 'ServerMessage' protocol and the
explicit 'recordedEventToJSON' encoder (an orphan-free function, not a @ToJSON@
instance — 'RecordedEvent' has none and @kiroku-store@ keeps its @Types@ module
instance-light).
-}
module Kiroku.Metrics.WebSocket (
    -- * Protocol
    ClientMessage (..),
    ServerMessage (..),
    recordedEventToJSON,
    recordedEventToJSONResolved,
    errorCodeReplayFailed,
    errorCodeCategoryReadFailed,
    errorCodeEventStreamOverflowed,
    errorCodeLiveDecodeFailed,
    overflowNotice,

    -- * Tail delivery building blocks
    StreamNameCache,
    newStreamNameCache,
    streamNameCacheSize,
    resolveEventNames,
    broadcastEventsWith,
    withWorkerSlot,

    -- * Connection limiting
    WebSocketState (..),
    newWebSocketState,

    -- * The WebSocket application (the IP-3 seam)
    websocketApp,
) where

import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (asyncWithUnmask, link, uninterruptibleCancel)
import Control.Concurrent.STM (
    STM,
    TVar,
    atomically,
    check,
    modifyTVar',
    newTVarIO,
    readTBQueue,
    readTVar,
    writeTVar,
 )
import Control.Exception (bracket, catch, finally, mask, mask_)
import Control.Monad (forever)
import Data.Aeson (
    FromJSON (..),
    ToJSON (..),
    Value (..),
    eitherDecode',
    encode,
    object,
    withObject,
    (.:),
    (.:?),
    (.=),
 )
import Data.Aeson.KeyMap qualified as KeyMap
import Data.ByteString qualified as BS
import Data.ByteString.Lazy qualified as LBS
import Data.Foldable (foldl', for_)
import Data.IORef (IORef, newIORef, readIORef, writeIORef)
import Data.Int (Int64)
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Sequence (Seq, ViewL (..), (|>))
import Data.Sequence qualified as Seq
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Vector (Vector)
import Data.Vector qualified as V
import Data.Word (Word64)
import Network.WebSockets qualified as WS

import Kiroku.Metrics.Collector (KirokuMetrics, snapshotMetrics)
import Kiroku.Metrics.Config (MetricsServerConfig (..))
import Kiroku.Metrics.Types (MetricsSnapshot)
import Kiroku.Store (
    CategoryName (..),
    GlobalPosition (..),
    KirokuStore (..),
    RecordedEvent (..),
    lookupStreamNames,
    readAllForward,
    readCategory,
    runStoreIO,
 )
import Kiroku.Store.Settings (DecodedBatch (..), DecodedEvent (..), decodedEventRecorded)
import Kiroku.Store.Subscription.EventPublisher (
    PublisherSubscription (..),
    SubscriberStatus (..),
    publisherPosition,
    subscribePublisherWith,
 )
import Kiroku.Store.Subscription.Types (OverflowPolicy (..))
import Kiroku.Store.Types (
    EventId (..),
    EventType (..),
    StreamId (..),
    StreamName (..),
    StreamVersion (..),
 )

--------------------------------------------------------------------------------
-- Protocol (IP-4)
--------------------------------------------------------------------------------

{- | Messages from a WebSocket client to the server (tagged on a @"type"@ field).

The constructors are field-less (positional) on purpose: 'SubscribeEvents' and
'ServerMessage'\'s 'EventStreamStarted' would otherwise both define a
@fromPosition@ record selector, which collides when @Kiroku.Metrics@ re-exports
both @(..)@ lists.
-}
data ClientMessage
    = -- | Keepalive request; answered with 'Pong'.
      Ping
    | -- | (Metrics channel) request a fresh snapshot now.
      SubscribeMetrics
    | {- | (Event channel) start streaming. The first field is @from_position@
      ('Nothing' = "from now"); the second is @category@ ('Nothing' = all streams).
      -}
      SubscribeEvents !(Maybe Int64) !(Maybe Text)
    | -- | (Event channel) stop the current tail.
      UnsubscribeEvents
    | -- | Stop periodic metrics snapshots; subscribe_metrics resumes them.
      UnsubscribeMetrics
    deriving stock (Eq, Show)

-- | Messages from the server to a WebSocket client (tagged on a @"type"@ field).
data ServerMessage
    = -- | Answer to 'Ping'.
      Pong
    | -- | (Metrics channel) a metrics snapshot, embedded under @"metrics"@.
      Snapshot !MetricsSnapshot
    | -- | (Event channel) one appended event, embedded under @"event"@.
      Event !Value
    | {- | (Event channel) acknowledgement that streaming has begun from a
      given global position (the @from_position@ field on the wire).
      -}
      EventStreamStarted !Int64
    | -- | The connection is being torn down.
      Goodbye
    | -- | A non-fatal error message for the client.
      ErrorMsg !Text
    | -- | An error with a stable machine-readable code.
      CodedError !Text !Text
    deriving stock (Eq, Show)

instance FromJSON ClientMessage where
    parseJSON = withObject "ClientMessage" $ \v -> do
        msgType <- v .: "type"
        case msgType :: Text of
            "ping" -> pure Ping
            "subscribe_metrics" -> pure SubscribeMetrics
            "unsubscribe_metrics" -> pure UnsubscribeMetrics
            "subscribe_events" ->
                SubscribeEvents <$> v .:? "from_position" <*> v .:? "category"
            "unsubscribe_events" -> pure UnsubscribeEvents
            other -> fail ("Unknown client message type: " <> T.unpack other)

instance ToJSON ServerMessage where
    toJSON Pong = object ["type" .= ("pong" :: Text)]
    toJSON (Snapshot snap) = object ["type" .= ("snapshot" :: Text), "metrics" .= snap]
    toJSON (Event ev) = object ["type" .= ("event" :: Text), "event" .= ev]
    toJSON (EventStreamStarted p) =
        object ["type" .= ("event_stream_started" :: Text), "from_position" .= p]
    toJSON Goodbye = object ["type" .= ("goodbye" :: Text)]
    toJSON (ErrorMsg msg) = object ["type" .= ("error" :: Text), "message" .= msg]
    toJSON (CodedError code msg) = object ["type" .= ("error" :: Text), "code" .= code, "message" .= msg]

-- | Published error codes (human messages may change).
errorCodeReplayFailed, errorCodeCategoryReadFailed, errorCodeEventStreamOverflowed, errorCodeLiveDecodeFailed :: Text
errorCodeReplayFailed = "replay_failed"
errorCodeCategoryReadFailed = "category_read_failed"
errorCodeEventStreamOverflowed = "event_stream_overflowed"
errorCodeLiveDecodeFailed = "live_decode_failed"

-- | A notice on any counter change, including a Word64 wrap (modular subtraction).
overflowNotice :: Word64 -> Word64 -> Maybe ServerMessage
overflowNotice previous current
    | previous /= current =
        Just
            ( CodedError
                errorCodeEventStreamOverflowed
                ( "event stream overflowed; "
                    <> T.pack (show (current - previous))
                    <> " undelivered batch(es) dropped since the last notice; re-read from your last position"
                )
            )
    | otherwise = Nothing

{- | Encode a 'RecordedEvent' to a JSON 'Value' (IP-4). An explicit function
rather than a @ToJSON@ instance: 'RecordedEvent' has no instance today and a
library-level orphan is undesirable. EP-4's user guide documents this shape.
-}
recordedEventToJSON :: RecordedEvent -> Value
recordedEventToJSON e =
    object
        [ "eventId" .= unEventId e.eventId
        , "eventType" .= unEventType e.eventType
        , "streamVersion" .= unStreamVersion e.streamVersion
        , "globalPosition" .= unGlobalPosition e.globalPosition
        , "originalStreamId" .= unStreamId e.originalStreamId
        , "originalVersion" .= unStreamVersion e.originalVersion
        , "payload" .= e.payload
        , "metadata" .= e.metadata
        , "causationId" .= e.causationId
        , "correlationId" .= e.correlationId
        , "createdAt" .= e.createdAt
        ]
  where
    unEventId (EventId u) = u
    unEventType (EventType t) = t
    unStreamVersion (StreamVersion n) = n
    unGlobalPosition (GlobalPosition n) = n
    unStreamId (StreamId n) = n

--------------------------------------------------------------------------------
-- Connection limiting (mirrors shibuya-metrics)
--------------------------------------------------------------------------------

{- | Shared state bounding the number of concurrent WebSocket connections.
Allocated once per server in 'Kiroku.Metrics.Server.startMetricsServerWithStore'
and captured by 'websocketApp', so the bound is shared across connections.
-}
data WebSocketState = WebSocketState
    { connectionCount :: !(TVar Int)
    , maxConnections :: !Int
    }

-- | Create a 'WebSocketState' bounding connections at @maxConns@.
newWebSocketState :: Int -> IO WebSocketState
newWebSocketState maxConns = do
    countVar <- newTVarIO 0
    pure WebSocketState{connectionCount = countVar, maxConnections = maxConns}

-- | Try to claim a connection slot; 'True' on success.
acquireConnection :: WebSocketState -> STM Bool
acquireConnection st = do
    count <- readTVar st.connectionCount
    if count < st.maxConnections
        then writeTVar st.connectionCount (count + 1) >> pure True
        else pure False

-- | Release a connection slot.
releaseConnection :: WebSocketState -> STM ()
releaseConnection st = modifyTVar' st.connectionCount (\c -> max 0 (c - 1))

--------------------------------------------------------------------------------
-- Path dispatch
--------------------------------------------------------------------------------

data WsPath = MetricsPath | EventsPath | UnknownPath

-- | Dispatch on the request path, ignoring any query string.
dispatchPath :: BS.ByteString -> WsPath
dispatchPath raw =
    case BS.takeWhile (/= 0x3f) raw of -- 0x3f = '?'
        p
            | p == "/ws/metrics" -> MetricsPath
            | p == "/ws/events" -> EventsPath
            | otherwise -> UnknownPath

--------------------------------------------------------------------------------
-- The WebSocket application
--------------------------------------------------------------------------------

{- | The real WebSocket app filling the IP-3 seam. Closes over the config, the
collector, the store (for event streaming — EP-2 deliberately kept the store out
of the /server/ signature, so it enters here), and the shared connection-limiting
state. Each upgrade is dispatched by path; an over-capacity upgrade is rejected.
-}
websocketApp ::
    MetricsServerConfig ->
    KirokuMetrics ->
    KirokuStore ->
    WebSocketState ->
    WS.ServerApp
websocketApp cfg m store st pending =
    case dispatchPath (WS.requestPath (WS.pendingRequest pending)) of
        MetricsPath -> guarded (handleMetrics cfg m)
        EventsPath -> guarded (handleEvents cfg store)
        UnknownPath ->
            WS.rejectRequest pending "Unknown WebSocket path; use /ws/metrics or /ws/events"
  where
    guarded run = do
        acquired <- atomically (acquireConnection st)
        if not acquired
            then WS.rejectRequest pending "Too many connections"
            else
                (run pending `catch` ignoreClosed)
                    `finally` atomically (releaseConnection st)
    -- A normal client disconnect surfaces as a 'WS.ConnectionException' from the
    -- receive loop; treat it as a clean end-of-connection rather than letting it
    -- escape to the server's exception reporter.
    ignoreClosed (_ :: WS.ConnectionException) = pure ()

--------------------------------------------------------------------------------
-- Metrics channel
--------------------------------------------------------------------------------

-- | Handle a @/ws/metrics@ connection: snapshot on connect, periodic push, ping/pong.
handleMetrics :: MetricsServerConfig -> KirokuMetrics -> WS.PendingConnection -> IO ()
handleMetrics cfg m pending = do
    conn <- WS.acceptRequest pending
    WS.withPingThread conn 30 (pure ()) $
        withWorkerSlot $ \start stop -> do
            sendMsg conn . Snapshot =<< snapshotMetrics m
            start (metricsPushLoop cfg m conn)
            metricsReceiveLoop m conn (start (metricsPushLoop cfg m conn)) stop
                `finally` (stop >> sendMsg conn Goodbye)

{- | Scope one linked worker. Creation and registration are masked, the worker
body is unmasked, and cleanup cancels and joins even during acquisition. Starting
an already occupied slot is a no-op. The receive loop is the sole slot owner.
-}
withWorkerSlot :: ((IO () -> IO ()) -> IO () -> IO a) -> IO a
withWorkerSlot body = mask $ \restore -> do
    workerVar <- newTVarIO Nothing
    let stop = mask_ $ do
            worker <- atomically (readTVar workerVar)
            for_ worker uninterruptibleCancel
            atomically (writeTVar workerVar Nothing)
        start action = mask_ $ do
            worker <- atomically (readTVar workerVar)
            case worker of
                Just _ -> pure ()
                Nothing -> do
                    child <- asyncWithUnmask (\unmask -> unmask action)
                    atomically (writeTVar workerVar (Just child))
                    link child
    restore (body start stop) `finally` stop

-- | Periodically push a fresh snapshot every @wsPushIntervalUs@.
metricsPushLoop :: MetricsServerConfig -> KirokuMetrics -> WS.Connection -> IO ()
metricsPushLoop cfg m conn = forever $ do
    threadDelay cfg.wsPushIntervalUs
    sendMsg conn . Snapshot =<< snapshotMetrics m

-- | Answer @ping@ with @pong@ and @subscribe_metrics@ with a fresh snapshot.
metricsReceiveLoop :: KirokuMetrics -> WS.Connection -> IO () -> IO () -> IO ()
metricsReceiveLoop m conn start stop = forever $ do
    cmd <- recvMsg conn
    case cmd of
        Just Ping -> sendMsg conn Pong
        Just SubscribeMetrics -> do
            sendMsg conn . Snapshot =<< snapshotMetrics m
            start
        Just UnsubscribeMetrics -> stop
        _ -> pure ()

--------------------------------------------------------------------------------
-- Event channel
--------------------------------------------------------------------------------

{- | Handle a @/ws/events@ connection. The connection starts idle; a
@subscribe_events@ message starts (or restarts) a tail, @unsubscribe_events@
stops it, and @ping@ → @pong@. The tail runs in a child thread tracked in a
'TVar' so the receive loop can cancel and replace it; disconnect tears it down.
-}
handleEvents :: MetricsServerConfig -> KirokuStore -> WS.PendingConnection -> IO ()
handleEvents cfg store pending = do
    conn <- WS.acceptRequest pending
    WS.withPingThread conn 30 (pure ()) $
        withWorkerSlot $ \start stop ->
            finally
                ( forever $ do
                    cmd <- recvMsg conn
                    case cmd of
                        Just Ping -> sendMsg conn Pong
                        Just (SubscribeEvents from cat) -> stop >> start (eventTail cfg store conn from cat)
                        Just UnsubscribeEvents -> stop
                        _ -> pure ()
                )
                (stop >> sendMsg conn Goodbye)

-- | A reasonable replay/category page size.
eventReadLimit :: Int
eventReadLimit = 500

{- | Stream events to the client. Dispatches on the request shape:

  * no @from@, no @category@: live "from-now" tail via the broadcast.
  * @from@, no @category@: replay history from @from@, tracking the highest
    delivered position, then live with the broadcast filtered to positions above
    that covered boundary. A replay read error sends an @error@ frame and ends
    the tail instead of entering live mode with a gap.
  * any @category@: a DB-driven loop over 'readCategory' gated on the publisher
    position (the broadcast carries no stream names, so it cannot be filtered by
    category in-process — see the plan).
-}
eventTail ::
    MetricsServerConfig ->
    KirokuStore ->
    WS.Connection ->
    Maybe Int64 ->
    Maybe Text ->
    IO ()
eventTail cfg store conn mFrom mCategory = do
    cache <- newStreamNameCache
    case mCategory of
        Just cat -> do
            start <- case mFrom of
                Just p -> pure p
                Nothing -> unGP <$> atomically (publisherPosition store.publisher)
            sendMsg conn (EventStreamStarted start)
            categoryLoop store cache conn (CategoryName cat) start
        Nothing ->
            bracket
                ( atomically $ do
                    sub <- subscribePublisherWith store.publisher cfg.wsEventQueueCap DropOldest
                    attachPos <- unGP <$> publisherPosition store.publisher
                    pure (sub, attachPos)
                )
                (unsubscribe . fst)
                $ \(sub, attachPos) ->
                    case mFrom of
                        Nothing -> do
                            sendMsg conn (EventStreamStarted attachPos)
                            broadcastEventsWith (sendMsg conn) cache (lookupNames store) sub (const True)
                        Just p -> do
                            sendMsg conn (EventStreamStarted p)
                            mCovered <- replayHistory store cache conn p attachPos
                            for_ mCovered $ \covered ->
                                broadcastEventsWith (sendMsg conn) cache (lookupNames store) sub (\e -> unGP e.globalPosition > covered)

{- | Page history from the requested position up to @attachPos@ with
'readAllForward'. Returns @Just covered@, the highest global position the
client is now guaranteed to have received (at least @attachPos@; more when the
final page read past it), or @Nothing@ after a read error, which has already
been surfaced to the client as an 'ErrorMsg'. The caller must terminate the
tail on @Nothing@ rather than continue live with a gap.
-}
replayHistory :: KirokuStore -> StreamNameCache -> WS.Connection -> Int64 -> Int64 -> IO (Maybe Int64)
replayHistory store cache conn from attachPos = go from attachPos
  where
    go cursor covered
        | cursor >= attachPos = pure (Just covered)
        | otherwise = do
            res <- runStoreIO store (readAllForward (GlobalPosition cursor) (fromIntegral eventReadLimit))
            case res of
                Left _ -> do
                    sendMsg conn (CodedError errorCodeReplayFailed "replay error: history unavailable")
                    pure Nothing
                Right evs
                    | V.null evs -> pure (Just covered)
                    | otherwise -> do
                        sendEvents store cache conn evs
                        let lastPos = unGP (V.last evs).globalPosition
                        go lastPos (max covered lastPos)

{- | Production live delivery with an injectable frame writer and name lookup.
Queue, status and counter are sampled atomically. Loss is signalled before any
surviving event, preserving the client's last contiguous recovery cursor.
-}
broadcastEventsWith ::
    (ServerMessage -> IO ()) ->
    StreamNameCache ->
    ([StreamId] -> IO (Map StreamId StreamName)) ->
    PublisherSubscription ->
    (RecordedEvent -> Bool) ->
    IO ()
broadcastEventsWith send cache lookupBatch sub keep = go 0 False
  where
    go previous warned = do
        (batch, status, dropped) <-
            atomically $
                (,,)
                    <$> readTBQueue sub.subscriptionQueue
                    <*> readTVar sub.subscriptionStatus
                    <*> readTVar sub.subscriptionDropped
        let notice = case overflowNotice previous dropped of
                Just msg -> Just msg
                Nothing | status == Overflowed && not warned -> Just (CodedError errorCodeEventStreamOverflowed "event stream overflowed; some events dropped")
                Nothing -> Nothing
        for_ notice send
        let decoded = case batch of
                UnchangedBatch events -> Right (V.filter keep events)
                TransformedBatch events -> V.mapM unwrap (V.filter (keep . decodedEventRecorded) events)
            unwrap (Decoded event) = Right event
            unwrap (Undecodable _ failure) = Left failure
        case decoded of
            Left _ -> send (CodedError errorCodeLiveDecodeFailed "live event decoding failed")
            Right events -> do
                names <- resolveEventNames cache lookupBatch events
                V.mapM_ (send . Event . recordedEventToJSONResolved names) events
                go dropped (status == Overflowed)

{- | DB-driven category live loop. Mirrors the subscription worker's
@liveLoopDbDriven@: gate on the publisher advancing past the /last observed/
position (not the cursor) so an unmatched category does not busy-spin, then drain
the category to empty before waiting again.
-}
categoryLoop :: KirokuStore -> StreamNameCache -> WS.Connection -> CategoryName -> Int64 -> IO ()
categoryLoop store cache conn cat startPos = go startPos 0
  where
    go cursor waitFrom = do
        pubPos <- atomically $ do
            GlobalPosition p <- publisherPosition store.publisher
            check (p > waitFrom)
            pure p
        drained <- drainTo cursor
        case drained of
            Nothing -> pure () -- a DB error already surfaced; stop the tail
            Just cursor' -> go cursor' pubPos
    drainTo cursor = do
        res <- runStoreIO store (readCategory cat (GlobalPosition cursor) (fromIntegral eventReadLimit))
        case res of
            Left _ -> do
                sendMsg conn (CodedError errorCodeCategoryReadFailed "category read error: events unavailable")
                pure Nothing
            Right evs
                | V.null evs -> pure (Just cursor)
                | otherwise -> do
                    sendEvents store cache conn evs
                    drainTo (unGP (V.last evs).globalPosition)

--------------------------------------------------------------------------------
-- Send / receive helpers
--------------------------------------------------------------------------------

-- | Send each event in a batch as an 'Event' message.
sendEvents :: KirokuStore -> StreamNameCache -> WS.Connection -> Vector RecordedEvent -> IO ()
sendEvents store cache conn events = do
    names <- resolveEventNames cache (lookupNames store) events
    V.mapM_ (sendMsg conn . Event . recordedEventToJSONResolved names) events

-- Typed lookup failures fall back to null names; thrown exceptions propagate.
lookupNames :: KirokuStore -> [StreamId] -> IO (Map StreamId StreamName)
lookupNames store ids = either (const Map.empty) id <$> runStoreIO store (lookupStreamNames ids)

data NameCache = NameCache !(Map StreamId StreamName) !(Seq StreamId)

-- | Per-tail FIFO name cache. Missing names are not retained. Capacity: 4096.
newtype StreamNameCache = StreamNameCache (IORef NameCache)

-- | Allocate once for a tail; dispose on unsubscribe/disconnect.
newStreamNameCache :: IO StreamNameCache
newStreamNameCache = StreamNameCache <$> newIORef (NameCache Map.empty Seq.empty)

-- | Retained map and FIFO sizes (both bounded by 4096).
streamNameCacheSize :: StreamNameCache -> IO (Int, Int)
streamNameCacheSize (StreamNameCache ref) = do
    NameCache names order <- readIORef ref
    pure (Map.size names, Seq.length order)

{- | Resolve distinct misses in at most one lookup. The temporary encoding map
contains all current-batch names even when that batch exceeds cache capacity.
The tail owns the cache; calls on one cache must be serialized.
-}
resolveEventNames :: StreamNameCache -> ([StreamId] -> IO (Map StreamId StreamName)) -> Vector RecordedEvent -> IO (Map StreamId StreamName)
resolveEventNames (StreamNameCache ref) lookupBatch events = do
    NameCache cached order <- readIORef ref
    let wanted = Set.fromList (V.toList (V.map (.originalStreamId) events))
        -- Inspect the requested ids rather than materializing all retained keys
        -- for each small batch after the cache has filled.
        missing = Set.filter (`Map.notMember` cached) wanted
    found <- if Set.null missing then pure Map.empty else Map.restrictKeys <$> lookupBatch (Set.toList missing) <*> pure missing
    let current = Map.union cached found
        inserted = foldl' (|>) order (Map.keys found)
        trimmed = evict current inserted
    trimmed `seq` writeIORef ref trimmed
    pure (Map.restrictKeys current wanted)
  where
    evict names order
        | Map.size names <= 4096 = NameCache names order
        | otherwise = case Seq.viewl order of
            oldest :< rest -> evict (Map.delete oldest names) rest
            EmptyL -> NameCache Map.empty Seq.empty

{- | Send a 'ServerMessage', swallowing a closed-connection exception so cleanup
in a @finally@ never re-throws on an already-dead socket.
-}
sendMsg :: WS.Connection -> ServerMessage -> IO ()
sendMsg conn msg =
    WS.sendTextData conn (encode msg)
        `catch` \(_ :: WS.ConnectionException) -> pure ()

{- | Receive and decode one 'ClientMessage'. 'Nothing' on an undecodable frame
(ignored by the caller).
-}
recvMsg :: WS.Connection -> IO (Maybe ClientMessage)
recvMsg conn = do
    raw <- WS.receiveData conn :: IO LBS.ByteString
    pure (either (const Nothing) Just (eitherDecode' raw))

-- | Unwrap a 'GlobalPosition' to its underlying 'Int64'.
unGP :: GlobalPosition -> Int64
unGP (GlobalPosition n) = n

-- | The frozen event shape plus the resolved original name (or null).
recordedEventToJSONResolved :: Map StreamId StreamName -> RecordedEvent -> Value
recordedEventToJSONResolved names event = case recordedEventToJSON event of
    Object fields -> Object (KeyMap.insert "original_stream_name" (toJSON (fmap (\(StreamName name) -> name) (Map.lookup event.originalStreamId names))) fields)
    _ -> error "recordedEventToJSON must return an object"