feat(txs-tracer-core): ✨ improve tx trace machine

improved tx trace machine state structer and add websocket query to rpc node.

package.json

@@ -5,9 +5,11 @@
   "scripts": {},
   "private": true,
   "dependencies": {
+    "@cosmjs/encoding": "^0.29.5",
     "@cosmjs/stargate": "^0.29.5",
     "@cosmjs/tendermint-rpc": "^0.29.5",
-    "xstate": "^4.37.0"
+    "xstate": "^4.37.0",
+    "xstream": "^11.14.0"
   },
   "devDependencies": {
     "@nrwl/eslint-plugin-nx": "15.7.2",

packages/txs-tracer-core/src/lib/index.ts

@@ -1,2 +1,3 @@
 export * from './types';
 export * from './machines';
+export * from './utils';

packages/txs-tracer-core/src/lib/machines/txs-trace-machine/txs-trace-machine.spec.ts

@@ -1,19 +1,60 @@
 import { interpret } from 'xstate';
 import { txTraceMachine } from './txs-trace-machine';
 
-test('should eventually reach "pending"', () =>
-	new Promise<void>(done => {
-		const fetchService = interpret(txTraceMachine).onTransition(state => {
-			if (state.matches('pending')) {
-				done();
-			}
-		});
+describe('txs-trace-machine', () => {
+	test('should eventually reach "pending_subscription"', () =>
+		new Promise<void>(done => {
+			const fetchService = interpret(
+				txTraceMachine.withContext({
+					...txTraceMachine.context,
+				}),
+			)
+				.onTransition(state => {
+					console.log(state.value);
+					if (state.matches('pending_subscription')) {
+						done();
+					}
+				})
+				.onEvent(event => {
+					console.log(event.type);
+				});
 
-		fetchService.start();
+			fetchService.start();
 
-		/*
-		 *  Send zero or more events to the service that should
-		 *  cause it to eventually reach its expected state
-		 */
-		fetchService.send({ type: 'TRACE' });
-	}));
+			/*
+			 *  Send zero or more events to the service that should
+			 *  cause it to eventually reach its expected state
+			 */
+			fetchService.send({ type: 'TRACE' });
+		}));
+
+	test(
+		'should eventually reach "results"',
+		() =>
+			new Promise<void>(done => {
+				const fetchService = interpret(
+					txTraceMachine.withContext({
+						...txTraceMachine.context,
+						subscribeTimeout: 1000,
+						query: "message.action='/osmosis.gamm.v1beta1.MsgSwapExactAmountIn'",
+					}),
+				).onTransition(state => {
+					if (state.matches('results')) {
+						console.log(state.context.txs);
+						done();
+					}
+				});
+
+				fetchService.start();
+
+				/*
+				 *  Send zero or more events to the service that should
+				 *  cause it to eventually reach its expected state
+				 */
+				fetchService.send({ type: 'TRACE' });
+			}),
+		{
+			timeout: 30_000,
+		},
+	);
+});

packages/txs-tracer-core/src/lib/machines/txs-trace-machine/txs-trace-machine.ts

@@ -1,6 +1,8 @@
 import { createMachine, raise } from 'xstate';
-import { Tendermint34Client } from '@cosmjs/tendermint-rpc';
-import { TxTraceContext } from '../../types';
+import { Tendermint34Client, TxEvent } from '@cosmjs/tendermint-rpc';
+import { Listener, Stream } from 'xstream';
+import { TxTraceContext, TxTraceEvents } from '../../types';
+import { mapIndexedTx } from '../../utils';
 
 export const txTraceMachine = createMachine(
 	{
@@ -37,12 +39,7 @@ export const txTraceMachine = createMachine(
 					},
 				},
 			},
-			pending: {
-				after: {
-					TX_TIMEOUT: {
-						target: 'pending',
-					},
-				},
+			pending_search_txs: {
 				on: {
 					TX_RESULTS: {
 						target: 'result',
@@ -55,39 +52,50 @@ export const txTraceMachine = createMachine(
 					},
 				},
 			},
+			pending_subscription: {
+				entry: ['subscribeToQuery'],
+				after: {
+					SUBSCRIBE_TIMEOUT: {
+						target: 'pending_search_txs',
+					},
+				},
+				on: {
+					TX_RESULTS: {
+						target: 'result',
+					},
+					CONNECTION_DISCONNECT: {
+						target: 'connection_error',
+					},
+				},
+			},
 			result: {
 				type: 'final',
+				data: (ctx, event) => ({
+					event,
+					txs: ctx.txs,
+				}),
+				entry: ['closeConnection'],
 			},
 			connected: {
 				on: {
-					SEND_QUERY_MESSAGE: { target: 'pending' },
+					SEND_QUERY_MESSAGE: { target: 'pending_subscription' },
 				},
 			},
 			closed: {
 				type: 'final',
-				entry: ctx => {
-					if (ctx.tendermintClient) {
-						ctx.tendermintClient.disconnect();
-
-						ctx.tendermintClient = undefined;
-					}
-				},
+				entry: ['closeConnection'],
 			},
 			connection_timeout: {
 				type: 'final',
-				entry: ctx => {
-					if (ctx.tendermintClient) {
-						ctx.tendermintClient.disconnect();
-
-						ctx.tendermintClient = undefined;
-					}
-				},
+				entry: ['closeConnection'],
 			},
 			not_found_error: {
 				type: 'final',
+				entry: ['closeConnection'],
 			},
 			connection_error: {
 				type: 'final',
+				entry: ['closeConnection'],
 			},
 			idle: {
 				on: {
@@ -99,30 +107,55 @@ export const txTraceMachine = createMachine(
 		},
 		schema: {
 			context: {} as TxTraceContext,
-			events: {} as
-				| { type: 'TX_RESULTS' }
-				| { type: 'CONNECTION_SUCCESS' }
-				| { type: 'CONNECTION_ERROR' }
-				| { type: 'SEND_QUERY_MESSAGE' }
-				| { type: 'CONNECTION_DISCONNECT' }
-				| { type: 'TRACE' }
-				| { type: 'TX_SEARCH_EMPTY' },
+			events: {} as TxTraceEvents,
 		},
 		context: {
 			tendermintClient: undefined,
-			txTimeout: 5000,
+			subscribeTimeout: 5000,
 			connectionTimeout: 5000,
 			websocketUrl: 'wss://rpc-osmosis.blockapsis.com',
-			query: "acknowledge_packet.packet_sequence='1753590'",
-			method: 'tx_search',
+			query: "acknowledge_packet.packet_sequence='1777404'",
+			txs: [],
 		},
 		predictableActionArguments: true,
 		preserveActionOrder: true,
 	},
 	{
+		actions: {
+			closeConnection: ctx => {
+				if (ctx.tendermintClient) {
+					ctx.tendermintClient.disconnect();
+
+					ctx.tendermintClient = undefined;
+				}
+			},
+			subscribeToQuery: ctx => {
+				let subscription: Stream<TxEvent>;
+				let listener: Listener<TxEvent>;
+
+				if (ctx.tendermintClient) {
+					subscription = ctx.tendermintClient.subscribeTx(ctx.query).take(1);
+
+					listener = {
+						next: txResult => {
+							ctx.txs = [mapIndexedTx(txResult)];
+						},
+						error: err => {
+							console.error(err);
+							raise('CONNECTION_DISCONNECT');
+						},
+						complete: () => {
+							raise('TX_RESULTS');
+						},
+					};
+
+					subscription.addListener(listener);
+				}
+			},
+		},
 		delays: {
-			TX_TIMEOUT: context => {
-				return context.txTimeout;
+			SUBSCRIBE_TIMEOUT: context => {
+				return context.subscribeTimeout;
 			},
 			CONNECTION_TIMEOUT: context => {
 				return context.connectionTimeout;

packages/txs-tracer-core/src/lib/types/tx-trace.ts

@@ -1,10 +1,20 @@
+import { IndexedTx } from '@cosmjs/stargate';
 import { Tendermint34Client } from '@cosmjs/tendermint-rpc';
 
 export interface TxTraceContext {
-	txTimeout: number;
+	subscribeTimeout: number;
 	connectionTimeout: number;
 	websocketUrl: string;
 	query: string;
-	method: string;
+	txs: IndexedTx[];
 	tendermintClient?: Tendermint34Client;
 }
+
+export type TxTraceEvents =
+	| { type: 'TX_RESULTS' }
+	| { type: 'CONNECTION_SUCCESS' }
+	| { type: 'CONNECTION_ERROR' }
+	| { type: 'SEND_QUERY_MESSAGE' }
+	| { type: 'CONNECTION_DISCONNECT' }
+	| { type: 'TRACE' }
+	| { type: 'TX_SEARCH_EMPTY' };

packages/txs-tracer-core/src/lib/utils/index.ts

@@ -0,0 +1 @@
+export * from './tx';

packages/txs-tracer-core/src/lib/utils/tx.ts

@@ -0,0 +1,14 @@
+import { TxEvent, TxResponse } from '@cosmjs/tendermint-rpc';
+import { fromTendermint34Event, IndexedTx } from '@cosmjs/stargate';
+import { toHex } from '@cosmjs/encoding';
+
+export const mapIndexedTx = (tx: TxResponse | TxEvent): IndexedTx => ({
+	height: tx.height,
+	hash: toHex(tx.hash).toUpperCase(),
+	code: tx.result.code,
+	events: tx.result.events.map(fromTendermint34Event),
+	rawLog: tx.result.log || '',
+	tx: tx.tx,
+	gasUsed: tx.result.gasUsed,
+	gasWanted: tx.result.gasWanted,
+});