{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE ScopedTypeVariables #-}

-- | Handler for the @WatchMempool@ SubmitService RPC method: stream
-- mempool transactions matching a predicate as they enter the mempool.
module Cardano.Rpc.Server.Internal.UtxoRpc.Mempool
  ( watchMempoolMethod
  , watchMempoolStream
  )
where

import Cardano.Rpc.Proto.Api.UtxoRpc.Submit qualified as U5c
import Cardano.Rpc.Server.Internal.Monad (MonadRpc)
import Cardano.Rpc.Server.Internal.UtxoRpc.Predicate (matchesTxPredicate)
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Mempool (txInModeToTxInMempool)
import Cardano.Rpc.Server.NodeKernelAccess
  ( MempoolWatchSnapshot (..)
  , grabNodeKernelAccess
  , nextMempoolWatchSnapshot
  , watchMempoolSnapshot
  )

import Ouroboros.Consensus.Mempool.API qualified as Consensus (TicketNo, zeroTicketNo)

import RIO

import Data.ProtoLens (defMessage)
import Network.GRPC.Spec (NextElem (NextElem), Proto)

-- | Handle the @WatchMempool@ SubmitService RPC method.
--
-- Streams new mempool entries matching the request's predicate, each with
-- 'U5c.stage' always @STAGE_MEMPOOL@ (the first locally observable stage;
-- see 'Cardano.Rpc.Server.Internal.UtxoRpc.Predicate.matchesTxPredicate').
--
-- Runs until the client disconnects or the stream is otherwise closed.
watchMempoolMethod
  :: MonadRpc e m
  => Proto U5c.WatchMempoolRequest
  -- ^ Request containing a filter predicate
  -> (NextElem (Proto U5c.WatchMempoolResponse) -> IO ())
  -- ^ Callback used to send each streamed response
  -> m ()
watchMempoolMethod :: forall e (m :: * -> *).
MonadRpc e m =>
Proto WatchMempoolRequest
-> (NextElem (Proto WatchMempoolResponse) -> IO ()) -> m ()
watchMempoolMethod Proto WatchMempoolRequest
request NextElem (Proto WatchMempoolResponse) -> IO ()
send = do
  nodeKernelAccess <- m NodeKernelAccess
forall e (m :: * -> *). MonadRpc e m => m NodeKernelAccess
grabNodeKernelAccess
  watchMempoolStream
    (watchMempoolSnapshot nodeKernelAccess)
    (nextMempoolWatchSnapshot nodeKernelAccess)
    (request ^. U5c.predicate)
    send

-- | The @WatchMempool@ streaming loop. Emits every mempool entry newer than
-- the last one seen, oldest first, filtered by the predicate.
--
-- Removals and slot-only changes (no new ticket) produce no message: the
-- proto has no representation for eviction, and a max-ticket comparison
-- upstream would otherwise miss exactly these transitions (see
-- 'Cardano.Rpc.Server.NodeKernelAccess.nextMempoolWatchSnapshot').
watchMempoolStream
  :: forall m
   . MonadIO m
  => m MempoolWatchSnapshot
  -- ^ Read the current mempool watch snapshot, without blocking - used
  -- once, for the initial baseline
  -> (MempoolWatchSnapshot -> m MempoolWatchSnapshot)
  -- ^ Block until the snapshot differs from the given one, then return the
  -- new one
  -> Proto U5c.TxPredicate
  -- ^ Predicate filtering which new entries are sent
  -> (NextElem (Proto U5c.WatchMempoolResponse) -> IO ())
  -- ^ Callback used to send each streamed response
  -> m ()
watchMempoolStream :: forall (m :: * -> *).
MonadIO m =>
m MempoolWatchSnapshot
-> (MempoolWatchSnapshot -> m MempoolWatchSnapshot)
-> Proto TxPredicate
-> (NextElem (Proto WatchMempoolResponse) -> IO ())
-> m ()
watchMempoolStream m MempoolWatchSnapshot
readSnapshot MempoolWatchSnapshot -> m MempoolWatchSnapshot
nextSnapshot Proto TxPredicate
predicate NextElem (Proto WatchMempoolResponse) -> IO ()
send = do
  initial <- m MempoolWatchSnapshot
readSnapshot
  go Consensus.zeroTicketNo initial
 where
  go :: Consensus.TicketNo -> MempoolWatchSnapshot -> m ()
  go :: TicketNo -> MempoolWatchSnapshot -> m ()
go TicketNo
lastSeenTicket snapshot :: MempoolWatchSnapshot
snapshot@MempoolWatchSnapshot{mempoolWatchTxsAfter :: MempoolWatchSnapshot -> TicketNo -> [(TxInMode, TicketNo)]
mempoolWatchTxsAfter = TicketNo -> [(TxInMode, TicketNo)]
txsAfter} = do
    let newEntries :: [(TxInMode, TicketNo)]
newEntries = TicketNo -> [(TxInMode, TicketNo)]
txsAfter TicketNo
lastSeenTicket
    [(TxInMode, TicketNo)] -> ((TxInMode, TicketNo) -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [(TxInMode, TicketNo)]
newEntries (((TxInMode, TicketNo) -> m ()) -> m ())
-> ((TxInMode, TicketNo) -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \(TxInMode
txInMode, TicketNo
_ticketNo) ->
      Maybe (Proto TxInMempool) -> (Proto TxInMempool -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (TxInMode -> Maybe (Proto TxInMempool)
txInModeToTxInMempool TxInMode
txInMode) ((Proto TxInMempool -> m ()) -> m ())
-> (Proto TxInMempool -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Proto TxInMempool
txInMempool ->
        Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Proto TxPredicate -> Proto Tx -> Bool
matchesTxPredicate Proto TxPredicate
predicate (Proto TxInMempool
txInMempool Proto TxInMempool
-> Getting (Proto Tx) (Proto TxInMempool) (Proto Tx) -> Proto Tx
forall s a. s -> Getting a s a -> a
^. Getting (Proto Tx) (Proto TxInMempool) (Proto Tx)
forall (f :: * -> *) s a.
(Functor f, HasField s "cardano" a) =>
LensLike' f s a
U5c.cardano)) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
          IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ())
-> (Proto WatchMempoolResponse -> IO ())
-> Proto WatchMempoolResponse
-> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NextElem (Proto WatchMempoolResponse) -> IO ()
send (NextElem (Proto WatchMempoolResponse) -> IO ())
-> (Proto WatchMempoolResponse
    -> NextElem (Proto WatchMempoolResponse))
-> Proto WatchMempoolResponse
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Proto WatchMempoolResponse -> NextElem (Proto WatchMempoolResponse)
forall a. a -> NextElem a
NextElem (Proto WatchMempoolResponse -> m ())
-> Proto WatchMempoolResponse -> m ()
forall a b. (a -> b) -> a -> b
$
            Proto WatchMempoolResponse
forall msg. Message msg => msg
defMessage Proto WatchMempoolResponse
-> (Proto WatchMempoolResponse -> Proto WatchMempoolResponse)
-> Proto WatchMempoolResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto WatchMempoolResponse) (Proto TxInMempool)
forall (f :: * -> *) s a.
(Functor f, HasField s "tx" a) =>
LensLike' f s a
U5c.tx LensLike' Identity (Proto WatchMempoolResponse) (Proto TxInMempool)
-> Proto TxInMempool
-> Proto WatchMempoolResponse
-> Proto WatchMempoolResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Proto TxInMempool
txInMempool
    let lastSeenTicket' :: TicketNo
lastSeenTicket' = case [(TxInMode, TicketNo)]
newEntries of
          [] -> TicketNo
lastSeenTicket
          [(TxInMode, TicketNo)]
_ -> (TxInMode, TicketNo) -> TicketNo
forall a b. (a, b) -> b
snd ([(TxInMode, TicketNo)] -> (TxInMode, TicketNo)
forall a. HasCallStack => [a] -> a
last [(TxInMode, TicketNo)]
newEntries)
    MempoolWatchSnapshot -> m MempoolWatchSnapshot
nextSnapshot MempoolWatchSnapshot
snapshot m MempoolWatchSnapshot -> (MempoolWatchSnapshot -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= TicketNo -> MempoolWatchSnapshot -> m ()
go TicketNo
lastSeenTicket'