feat: added fetchhistory, watchtx and watchmempool

examples/node/follow-tip.ts

@@ -1,39 +1,39 @@
-import { CardanoSyncClient, CardanoQueryClient } from "@saibdev/utxorpc-sdk";
+import { SyncClient, QueryClient } from "../../src/cardano";
 
-async function test() {
-  let syncClient = new CardanoSyncClient({
-    uri: "http://localhost:50051",
-  });
+ async function test() {
+   let syncClient = new SyncClient({
+     uri: "http://localhost:50051",
+   });
 
-  let queryClient = new CardanoQueryClient({
-    uri: "http://localhost:50051",
-  });
+   let queryClient = new QueryClient({
+     uri: "http://localhost:50051",
+   });
 
-  const params = await queryClient.readParams();
+   const params = await queryClient.readParams();
 
-  console.log(params);
+   console.log(params);
 
-  const utxos = await queryClient.searchUtxosByAddress(
-    Buffer.from(
-      "705c87cbca3a88cbfee6f6ad820acea99f484b4830fc632610f2a30146",
-      "hex"
-    )
-  );
-  console.log(
-    utxos.map((utxo) => {
-      console.log(utxo.parsedValued?.script);
-    })
-  );
+   const utxos = await queryClient.searchUtxosByAddress(
+     Buffer.from(
+       "705c87cbca3a88cbfee6f6ad820acea99f484b4830fc632610f2a30146",
+       "hex"
+     )
+   );
+   console.log(
+     utxos.map((utxo) => {
+       console.log(utxo.parsedValued?.script);
+     })
+   );
 
-  let tip = syncClient.followTip([
-    {
-      slot: 54131816,
-      hash: "34c65aba4b299113a488b74e2efe3a3dd272d25b470d25f374b2c693d4386535",
-    },
-  ]);
+   let tip = syncClient.followTip([
+     {
+       slot: 54131816,
+       hash: "34c65aba4b299113a488b74e2efe3a3dd272d25b470d25f374b2c693d4386535",
+     },
+   ]);
 
-  for await (const event of tip) {
-    console.log(event);
-  }
-}
-test().catch(console.error);
+   for await (const event of tip) {
+     console.log(event);
+   }
+ }
+ test().catch(console.error);

examples/node/package.json

@@ -9,7 +9,7 @@
   "license": "ISC",
   "description": "",
   "dependencies": {
-    "@saibdev/utxorpc-sdk": "0.6.4"
+    "@utxorpc/sdk": "^0.6.2"
   },
   "devDependencies": {
     "tsx": "^4.19.0",

src/cardano.ts

@@ -1,9 +1,6 @@
-import {
-  PromiseClient,
-  createPromiseClient,
-} from "@connectrpc/connect";
+import { PromiseClient, createPromiseClient } from "@connectrpc/connect";
 
-import { createGrpcTransport } from '@sdk/grpcTransport';
+import { createGrpcTransport } from "@sdk/grpcTransport";
 
 import { PartialMessage } from "@bufbuild/protobuf";
 
@@ -16,23 +13,50 @@ import {
   queryConnect,
   submit,
   submitConnect,
-  cardano,
+  watchConnect,
+  watch,
+  cardano
 } from "@utxorpc/spec";
 
 import {
   ClientBuilderOptions,
-  metadataInterceptor,
   GenericTipEvent,
   GenericUtxo,
+  GenericTxEvent,
+  GenericTxInMempoolEvent,
+  metadataInterceptor,
 } from "./common.js";
 
 export type ChainPoint = { slot: number | string; hash: string };
 export type Utxo = GenericUtxo<query.TxoRef, cardano.TxOutput>;
 export type TipEvent = GenericTipEvent<cardano.Block, ChainPoint>;
+export type TxEvent = GenericTxEvent<cardano.Tx>;
+export type MempoolEvent = GenericTxInMempoolEvent<cardano.Tx>;
 export type TxHash = Uint8Array;
 export type TxCbor = Uint8Array;
 
-function anyChainToBlock(msg) {
+function toMempoolEvent(txInMempool: submit.TxInMempool): MempoolEvent {
+  return {
+    txoRef: txInMempool.ref,
+    stage: txInMempool.stage,
+    nativeBytes: txInMempool.nativeBytes,
+    Tx:
+      txInMempool.parsedState.case == "cardano"
+        ? txInMempool.parsedState.value
+        : undefined,
+  };
+}
+function toTxEvent(response: watch.WatchTxResponse): TxEvent {
+  return {
+    action: response.action.case as "apply" | "undo",
+    Tx:
+      response.action.value?.chain.case == "cardano"
+        ? response.action.value?.chain.value
+        : undefined,
+  };
+}
+
+function anyChainToBlock(msg: sync.AnyChainBlock) {
   return msg.chain.case == "cardano" ? msg.chain.value : null;
 }
 
@@ -43,7 +67,7 @@ function pointToBlockRef(p: ChainPoint) {
   });
 }
 
-function blockRefToPoint(r) {
+function blockRefToPoint(r: sync.BlockRef) {
   return {
     slot: r.index.toString(),
     hash: Buffer.from(r.hash).toString("hex"),
@@ -122,6 +146,26 @@ export class SyncClient {
     const res = await this.inner.fetchBlock({ ref: [req] });
     return anyChainToBlock(res.block[0])!;
   }
+
+  async fetchHistory(p: ChainPoint, maxItems = 1): Promise<cardano.Block> {
+    const req = new sync.DumpHistoryRequest({
+      startToken: new sync.BlockRef({
+        index: BigInt(p.slot),
+        hash: Buffer.from(p.hash, "hex"),
+      }),
+      maxItems: maxItems,
+    });
+
+    const res = await this.inner.dumpHistory(req);
+
+    if (res.block.length === 0) {
+      throw new Error("No block history found for the provided ChainPoint.");
+    }
+
+    const block = anyChainToBlock(res.block[0]);
+
+    return block!;
+  }
 }
 
 export class QueryClient {
@@ -187,7 +231,9 @@ export class QueryClient {
     });
   }
 
-  async searchUtxosByDelegationPart(delegationPart: Uint8Array): Promise<Utxo[]> {
+  async searchUtxosByDelegationPart(
+    delegationPart: Uint8Array
+  ): Promise<Utxo[]> {
     return this.searchUtxosByMatch({
       address: {
         delegationPart: delegationPart,
@@ -250,4 +296,144 @@ export class SubmitClient {
       yield change.stage;
     }
   }
+
+  async *watchMempoolByMatch(
+    pattern: PartialMessage<cardano.TxPattern>
+  ): AsyncIterable<MempoolEvent> {
+    const stream = this.inner.watchMempool({
+      predicate: {
+        match: { chain: { value: pattern, case: "cardano" } },
+      },
+    });
+
+    for await (const response of stream) {
+      if (response.tx) {
+        yield toMempoolEvent(response.tx);
+      }
+    }
+  }
+
+  async *watchMempool(): AsyncIterable<MempoolEvent> {
+    yield* this.watchMempoolByMatch({});
+  }
+
+  async *watchMempoolForAddress(
+    address: Uint8Array
+  ): AsyncIterable<MempoolEvent> {
+    yield* this.watchMempoolByMatch({
+      hasAddress: { exactAddress: address },
+    });
+  }
+
+  async *watchMempoolForPaymentPart(
+    paymentPart: Uint8Array
+  ): AsyncIterable<MempoolEvent> {
+    yield* this.watchMempoolByMatch({
+      hasAddress: { paymentPart: paymentPart },
+    });
+  }
+
+  async *watchMempoolForDelegationPart(
+    delegationPart: Uint8Array
+  ): AsyncIterable<MempoolEvent> {
+    yield* this.watchMempoolByMatch({
+      hasAddress: { delegationPart: delegationPart },
+    });
+  }
+
+  async *watchMempoolForAsset(
+    policyId?: Uint8Array,
+    assetName?: Uint8Array
+  ): AsyncIterable<MempoolEvent> {
+    yield* this.watchMempoolByMatch({
+      movesAsset: policyId ? { policyId } : { assetName },
+    });
+  }
+}
+
+export class WatchClient {
+  inner: PromiseClient<typeof watchConnect.WatchService>;
+
+  constructor(options: ClientBuilderOptions) {
+    let headerInterceptor = metadataInterceptor(options);
+
+    const transport = createGrpcTransport({
+      httpVersion: "2",
+      baseUrl: options.uri,
+      interceptors: [headerInterceptor],
+    });
+
+    this.inner = createPromiseClient(watchConnect.WatchService, transport);
+  }
+
+  async *watchTxByMatch(
+    pattern: PartialMessage<cardano.TxPattern>,
+    intersect?: ChainPoint[]
+  ): AsyncIterable<TxEvent> {
+    const request: watch.WatchTxRequest = new watch.WatchTxRequest({
+      intersect: intersect ? intersect.map(pointToBlockRef) : [],
+      predicate: {
+        match: {
+          chain: {
+            value: pattern,
+            case: "cardano",
+          },
+        },
+      },
+    });
+
+    const stream = this.inner.watchTx(request);
+
+    for await (const response of stream) {
+      switch (response.action.case) {
+        case "apply":
+          yield toTxEvent(response);
+          break;
+
+        case "undo":
+          yield toTxEvent(response);
+          break;
+      }
+    }
+  }
+
+  async *watchTx(intersect?: ChainPoint[]): AsyncIterable<TxEvent> {
+    const pattern = {};
+    yield* this.watchTxByMatch(pattern, intersect);
+  }
+
+  async *watchTxForAddress(
+    address: Uint8Array,
+    intersect?: ChainPoint[]
+  ): AsyncIterable<TxEvent> {
+    const pattern = { hasAddress: { exactAddress: address } };
+    yield* this.watchTxByMatch(pattern, intersect);
+  }
+
+  async *watchTxForPaymentPart(
+    paymentPart: Uint8Array,
+    intersect?: ChainPoint[]
+  ): AsyncIterable<TxEvent> {
+    const pattern = { hasAddress: { paymentPart } };
+    yield* this.watchTxByMatch(pattern, intersect);
+  }
+
+  async *watchTxForDelegationPart(
+    delegationPart: Uint8Array,
+    intersect?: ChainPoint[]
+  ): AsyncIterable<TxEvent> {
+    const pattern = { hasAddress: { delegationPart } };
+    yield* this.watchTxByMatch(pattern, intersect);
+  }
+
+  async *watchTxForAsset(
+    policyId?: Uint8Array,
+    assetName?: Uint8Array,
+    intersect?: ChainPoint[]
+  ): AsyncIterable<TxEvent> {
+    const pattern = policyId
+      ? { movesAsset: { policyId } }
+      : { movesAsset: { assetName } };
+    yield* this.watchTxByMatch(pattern, intersect);
+  }
 }

src/common.ts

@@ -1,7 +1,5 @@
-import {
-  Interceptor,
-} from "@connectrpc/connect";
-
+import { Interceptor } from "@connectrpc/connect";
+import { submit } from "@utxorpc/spec";
 export function metadataInterceptor(
   options?: ClientBuilderOptions
 ): Interceptor {
@@ -21,6 +19,17 @@ export type GenericTipEvent<Block, Point> =
   | { action: "undo"; block: Block }
   | { action: "reset"; point: Point };
 
+export type GenericTxEvent<Tx> =
+  | { action: "apply"; Tx: Tx | undefined }
+  | { action: "undo"; Tx: Tx | undefined };
+
+export type GenericTxInMempoolEvent<Tx> = {
+  stage: submit.Stage;
+  txoRef: Uint8Array;
+  nativeBytes: Uint8Array;
+  Tx: Tx | undefined;
+};
+
 export type GenericUtxo<Ref, Parsed> = {
   txoRef: Ref;
   parsedValued: Parsed | undefined;