From e93a8fc0a0d4179aeb9c4175c5663e4d13d38f9d Mon Sep 17 00:00:00 2001 From: Opulence Chuks Date: Tue, 23 Jun 2026 12:18:52 +0100 Subject: [PATCH 1/4] feat: add rate limiting to ingest endpoint --- ingest-node/package.json | 3 ++- ingest-node/src/index.ts | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/ingest-node/package.json b/ingest-node/package.json index 32221ba4..cf87a1d2 100644 --- a/ingest-node/package.json +++ b/ingest-node/package.json @@ -13,7 +13,8 @@ "ioredis": "^5.4.1", "nats": "^2.28.2", "prom-client": "^15.1.3", - "zod": "^3.23.8" + "zod": "^3.23.8", + "@fastify/rate-limit": "^8.0.0" }, "devDependencies": { "@types/node": "^20.0.0", diff --git a/ingest-node/src/index.ts b/ingest-node/src/index.ts index e6d26934..1a98d673 100644 --- a/ingest-node/src/index.ts +++ b/ingest-node/src/index.ts @@ -31,6 +31,7 @@ import { z } from "zod"; import Redis from "ioredis"; import { connect as natsConnect, StringCodec, type NatsConnection } from "nats"; import { Registry, Counter, Histogram, collectDefaultMetrics } from "prom-client"; +import fastifyRateLimit from "@fastify/rate-limit"; // --------------------------------------------------------------------------- // Config @@ -383,6 +384,7 @@ const app = Fastify({ logger: false, // disable for benchmark — logging adds latency trustProxy: true, }); +app.register(fastifyRateLimit, { max: 100, timeWindow: 60000 }); app.post<{ Body: unknown }>("/ingest", async (req, reply) => { const requestStart = process.hrtime.bigint(); From e79734c9d976da8343d8800c8cac0a42ac1e0c92 Mon Sep 17 00:00:00 2001 From: Opulence Chuks Date: Tue, 23 Jun 2026 15:12:40 +0100 Subject: [PATCH 2/4] Add withParentTrace helper to propagate trace IDs to child jobs --- src/queue/trace.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/queue/trace.ts b/src/queue/trace.ts index 1f079c1a..9e520d82 100644 --- a/src/queue/trace.ts +++ b/src/queue/trace.ts @@ -63,3 +63,18 @@ export function childLoggerWithTrace( const traceId = traceIdFromJob(data); return traceId ? childLogger(traceId) : undefined; } + +/** + * Propagates the trace ID from a parent job's data to a new child job's data. + * If the parent job contains a trace ID (via TRACE_ID_KEY), it is copied + * into the child data. Otherwise a new UUID is generated ensuring the child + * job remains traceable. + */ +export function withParentTrace>( + parentData: Record | undefined, + childData: T, +): T & { [TRACE_ID_KEY]: string } { + const existingTraceId = traceIdFromJob(parentData); + const traceId = existingTraceId ?? crypto.randomUUID(); + return { ...childData, [TRACE_ID_KEY]: traceId } as any; +} From 2c0e1eea9c4f9c02d20d5375746ed9460bd62fc6 Mon Sep 17 00:00:00 2001 From: Opulence Chuks Date: Tue, 23 Jun 2026 16:00:29 +0100 Subject: [PATCH 3/4] feat: track request durations with timing metrics and error handling --- ingest-node/src/index.ts | 50 +++++++++++++++++++++++----------------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/ingest-node/src/index.ts b/ingest-node/src/index.ts index 1a98d673..7547601f 100644 --- a/ingest-node/src/index.ts +++ b/ingest-node/src/index.ts @@ -388,31 +388,39 @@ app.register(fastifyRateLimit, { max: 100, timeWindow: 60000 }); app.post<{ Body: unknown }>("/ingest", async (req, reply) => { const requestStart = process.hrtime.bigint(); + try { + // --- Parse + validate --- + const parseStart = process.hrtime.bigint(); + const parsed = CallbackSchema.safeParse(req.body); + const parseNs = Number(process.hrtime.bigint() - parseStart); + ingestParseDurationSeconds.observe(parseNs / 1e9); + + if (!parsed.success) { + ingestRequestsTotal.inc({ status_code: "400" }); + const totalNs = Number(process.hrtime.bigint() - requestStart); + ingestRequestDurationSeconds.observe({ status_code: "400" }, totalNs / 1e9); + return reply.status(400).send({ error: "Invalid payload", details: parsed.error.flatten() }); + } - // --- Parse + validate --- - const parseStart = process.hrtime.bigint(); - const parsed = CallbackSchema.safeParse(req.body); - const parseNs = Number(process.hrtime.bigint() - parseStart); - ingestParseDurationSeconds.observe(parseNs / 1e9); + // --- Publish to streams --- + const publishStart = process.hrtime.bigint(); + await publish(parsed.data); + const publishNs = Number(process.hrtime.bigint() - publishStart); + ingestPublishDurationSeconds.observe({ target: "all" }, publishNs / 1e9); - if (!parsed.success) { - ingestRequestsTotal.inc({ status_code: "400" }); + ingestRequestsTotal.inc({ status_code: "202" }); const totalNs = Number(process.hrtime.bigint() - requestStart); - ingestRequestDurationSeconds.observe({ status_code: "400" }, totalNs / 1e9); - return reply.status(400).send({ error: "Invalid payload", details: parsed.error.flatten() }); - } - - // --- Publish to streams --- - const publishStart = process.hrtime.bigint(); - await publish(parsed.data); - const publishNs = Number(process.hrtime.bigint() - publishStart); - ingestPublishDurationSeconds.observe({ target: "all" }, publishNs / 1e9); + ingestRequestDurationSeconds.observe({ status_code: "202" }, totalNs / 1e9); - ingestRequestsTotal.inc({ status_code: "202" }); - const totalNs = Number(process.hrtime.bigint() - requestStart); - ingestRequestDurationSeconds.observe({ status_code: "202" }, totalNs / 1e9); - - return reply.status(202).send({ status: "accepted", reference: parsed.data.reference }); + return reply.status(202).send({ status: "accepted", reference: parsed.data.reference }); + } catch (err) { + // Unexpected error handling + ingestRequestsTotal.inc({ status_code: "500" }); + const totalNs = Number(process.hrtime.bigint() - requestStart); + ingestRequestDurationSeconds.observe({ status_code: "500" }, totalNs / 1e9); + console.error('[ingest-node] unexpected error:', err); + return reply.status(500).send({ error: 'Internal server error' }); + } }); app.get("/health", async (_req, reply) => { From 0854b121003e49aad41456504527846a50ff0e4e Mon Sep 17 00:00:00 2001 From: Opulence Chuks Date: Wed, 24 Jun 2026 09:37:14 +0100 Subject: [PATCH 4/4] feat(cli): wrap transaction hashes in clickable StellarExpert explorer links --- momo-cli | 1 + src/scripts/momo-cli.ts | 50 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/momo-cli b/momo-cli index 7002400f..9dcb2a84 100755 --- a/momo-cli +++ b/momo-cli @@ -1,6 +1,7 @@ #!/usr/bin/env bash # Mobile Money Admin CLI Wrapper # Usage: ./momo-cli retry-batch +# Note: Automatically wraps transaction hashes in clickable StellarExpert links. # Set current working directory to project root DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" diff --git a/src/scripts/momo-cli.ts b/src/scripts/momo-cli.ts index 1d7d6391..df0d216e 100644 --- a/src/scripts/momo-cli.ts +++ b/src/scripts/momo-cli.ts @@ -15,6 +15,56 @@ import { addTransactionJob } from "../queue/index.js"; dotenv.config(); +const hashRegex = /\b([0-9a-fA-F]{64})\b/g; + +function formatTransactionHashes(text: string): string { + const network = + process.env.STELLAR_NETWORK === "mainnet" || + process.env.STELLAR_NETWORK === "public" + ? "public" + : "testnet"; + + return text.replace(hashRegex, (match) => { + const url = `https://stellar.expert/explorer/${network}/tx/${match}`; + return `\x1b]8;;${url}\x1b\\\x1b[36m\x1b[1m${match}\x1b[0m\x1b]8;;\x1b\\`; + }); +} + +// Intercept process.stdout.write and process.stderr.write to automatically format hashes +const originalStdoutWrite = process.stdout.write; +const originalStderrWrite = process.stderr.write; + +process.stdout.write = function ( + chunk: any, + encodingOrCb?: any, + cb?: any +): boolean { + if (typeof chunk === "string") { + chunk = formatTransactionHashes(chunk); + } else if (chunk instanceof Uint8Array) { + const text = new TextDecoder().decode(chunk); + const formatted = formatTransactionHashes(text); + chunk = new TextEncoder().encode(formatted); + } + return originalStdoutWrite.call(process.stdout, chunk, encodingOrCb, cb); +}; + +process.stderr.write = function ( + chunk: any, + encodingOrCb?: any, + cb?: any +): boolean { + if (typeof chunk === "string") { + chunk = formatTransactionHashes(chunk); + } else if (chunk instanceof Uint8Array) { + const text = new TextDecoder().decode(chunk); + const formatted = formatTransactionHashes(text); + chunk = new TextEncoder().encode(formatted); + } + return originalStderrWrite.call(process.stderr, chunk, encodingOrCb, cb); +}; + + const isTest = process.env.NODE_ENV === "test"; const colors = { reset: isTest ? "" : "\x1b[0m",