feat: implement utxorpc watch module (txpipe#270)

Author: scarmuega

src/serve/grpc/mod.rs

@@ -12,6 +12,7 @@ use crate::{prelude::*, submit::Transaction};
 mod query;
 mod submit;
 mod sync;
+mod watch;
 
 #[derive(Serialize, Deserialize, Clone)]
 pub struct Config {
@@ -29,13 +30,16 @@ pub async fn serve(
 ) -> Result<(), Error> {
     let addr = config.listen_address.parse().unwrap();
 
-    let sync_service = sync::ChainSyncServiceImpl::new(wal, ledger.clone());
+    let sync_service = sync::ChainSyncServiceImpl::new(wal.clone(), ledger.clone());
     let sync_service =
         u5c::sync::chain_sync_service_server::ChainSyncServiceServer::new(sync_service);
 
-    let query_service = query::QueryServiceImpl::new(ledger);
+    let query_service = query::QueryServiceImpl::new(ledger.clone());
     let query_service = u5c::query::query_service_server::QueryServiceServer::new(query_service);
 
+    let watch_service = watch::WatchServiceImpl::new(wal.clone(), ledger.clone());
+    let watch_service = u5c::watch::watch_service_server::WatchServiceServer::new(watch_service);
+
     let submit_service = submit::SubmitServiceImpl::new(txs_out, mempool);
     let submit_service =
         u5c::submit::submit_service_server::SubmitServiceServer::new(submit_service);
@@ -45,6 +49,7 @@ pub async fn serve(
         .register_encoded_file_descriptor_set(u5c::sync::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(u5c::query::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(u5c::submit::FILE_DESCRIPTOR_SET)
+        .register_encoded_file_descriptor_set(u5c::watch::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(protoc_wkt::google::protobuf::FILE_DESCRIPTOR_SET)
         .build()
         .unwrap();
@@ -68,6 +73,7 @@ pub async fn serve(
         .add_service(tonic_web::enable(sync_service))
         .add_service(tonic_web::enable(query_service))
         .add_service(tonic_web::enable(submit_service))
+        .add_service(tonic_web::enable(watch_service))
         .add_service(reflection)
         .serve_with_shutdown(addr, exit.cancelled())
         .await

src/serve/grpc/watch.rs

@@ -0,0 +1,92 @@
+use crate::{
+    ledger,
+    wal::{self, WalReader as _},
+};
+use futures_core::Stream;
+use futures_util::StreamExt;
+use pallas::interop::utxorpc as interop;
+use pallas::interop::utxorpc::spec as u5c;
+use pallas::ledger::traverse::MultiEraBlock;
+use std::pin::Pin;
+use tonic::{Request, Response, Status};
+
+fn block_to_txs(
+    block: &wal::RawBlock,
+    mapper: &interop::Mapper<ledger::store::LedgerStore>,
+) -> Vec<u5c::watch::AnyChainTx> {
+    let wal::RawBlock { body, .. } = block;
+    let block = MultiEraBlock::decode(body).unwrap();
+    let txs = block.txs();
+
+    txs.iter()
+        .map(|x| mapper.map_tx(x))
+        .map(|x| u5c::watch::AnyChainTx {
+            chain: Some(u5c::watch::any_chain_tx::Chain::Cardano(x)),
+        })
+        .collect()
+}
+
+fn roll_to_watch_response(
+    mapper: &interop::Mapper<ledger::store::LedgerStore>,
+    log: &wal::LogValue,
+) -> impl Stream<Item = u5c::watch::WatchTxResponse> {
+    let txs: Vec<_> = match log {
+        wal::LogValue::Apply(block) => block_to_txs(block, mapper)
+            .into_iter()
+            .map(u5c::watch::watch_tx_response::Action::Apply)
+            .map(|x| u5c::watch::WatchTxResponse { action: Some(x) })
+            .collect(),
+        wal::LogValue::Undo(block) => block_to_txs(block, mapper)
+            .into_iter()
+            .map(u5c::watch::watch_tx_response::Action::Undo)
+            .map(|x| u5c::watch::WatchTxResponse { action: Some(x) })
+            .collect(),
+        // TODO: shouldn't we have a u5c event for origin?
+        wal::LogValue::Mark(..) => vec![],
+    };
+
+    tokio_stream::iter(txs)
+}
+
+pub struct WatchServiceImpl {
+    wal: wal::redb::WalStore,
+    mapper: interop::Mapper<ledger::store::LedgerStore>,
+}
+
+impl WatchServiceImpl {
+    pub fn new(wal: wal::redb::WalStore, ledger: ledger::store::LedgerStore) -> Self {
+        Self {
+            wal,
+            mapper: interop::Mapper::new(ledger),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl u5c::watch::watch_service_server::WatchService for WatchServiceImpl {
+    type WatchTxStream = Pin<
+        Box<dyn Stream<Item = Result<u5c::watch::WatchTxResponse, tonic::Status>> + Send + 'static>,
+    >;
+
+    async fn watch_tx(
+        &self,
+        request: Request<u5c::watch::WatchTxRequest>,
+    ) -> Result<Response<Self::WatchTxStream>, Status> {
+        let _ = request.into_inner();
+
+        let from_seq = self
+            .wal
+            .find_tip()
+            .map_err(|_err| Status::internal("can't read WAL"))?
+            .map(|(x, _)| x)
+            .unwrap_or_default();
+
+        let mapper = self.mapper.clone();
+
+        let stream = wal::WalStream::start(self.wal.clone(), from_seq)
+            .flat_map(move |(_, log)| roll_to_watch_response(&mapper, &log))
+            .map(Ok);
+
+        Ok(Response::new(Box::pin(stream)))
+    }
+}