feat(cli): wrap transaction hashes in clickable StellarExpert links

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",

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,34 +384,43 @@ 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();
+  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) => {

momo-cli

@@ -1,6 +1,7 @@
 #!/usr/bin/env bash
 # Mobile Money Admin CLI Wrapper
 # Usage: ./momo-cli retry-batch <batch_id>
+# Note: Automatically wraps transaction hashes in clickable StellarExpert links.
 
 # Set current working directory to project root
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

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<T extends Record<string, unknown>>(
+  parentData: Record<string, unknown> | 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;
+}

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",