fix: settlement & reconciliation engine — eliminate in-memory state, add middleware integration

drizzle/0046_settlement_reconciliation.sql

@@ -0,0 +1,100 @@
+-- Settlement & Reconciliation Engine — PostgreSQL Persistence Tables
+-- All polyglot services (Go, Rust, Python, TypeScript) auto-create tables on startup,
+-- but this migration ensures schema consistency across environments.
+
+-- Go settlement-batch-processor (port 9211)
+CREATE TABLE IF NOT EXISTS "settlement_batches" (
+    "batch_id" TEXT PRIMARY KEY,
+    "status" TEXT NOT NULL DEFAULT 'pending',
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    "completed_at" TIMESTAMPTZ,
+    "agent_count" INT NOT NULL DEFAULT 0,
+    "total_volume" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "total_fees" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "total_commission" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "net_settlement" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "entries_json" JSONB NOT NULL DEFAULT '[]'
+);
+CREATE INDEX IF NOT EXISTS "idx_settlement_batches_status" ON "settlement_batches"("status");
+CREATE INDEX IF NOT EXISTS "idx_settlement_batches_created" ON "settlement_batches"("created_at" DESC);
+
+-- Go revenue-reconciler (port 9101)
+CREATE TABLE IF NOT EXISTS "reconciliation_reports" (
+    "id" BIGINT PRIMARY KEY,
+    "period" TEXT NOT NULL,
+    "status" TEXT NOT NULL DEFAULT 'pending',
+    "projected_json" JSONB NOT NULL DEFAULT '{}',
+    "actual_json" JSONB NOT NULL DEFAULT '{}',
+    "revenue_variance_pct" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "volume_variance_pct" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "agent_variance_pct" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "insights_json" JSONB NOT NULL DEFAULT '[]',
+    "generated_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    "approved_by" TEXT,
+    "approved_at" TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS "idx_recon_reports_period" ON "reconciliation_reports"("period");
+CREATE INDEX IF NOT EXISTS "idx_recon_reports_status" ON "reconciliation_reports"("status");
+
+CREATE TABLE IF NOT EXISTS "discrepancy_alerts" (
+    "id" SERIAL PRIMARY KEY,
+    "period" TEXT NOT NULL,
+    "metric" TEXT NOT NULL,
+    "projected" DOUBLE PRECISION NOT NULL,
+    "actual" DOUBLE PRECISION NOT NULL,
+    "variance_pct" DOUBLE PRECISION NOT NULL,
+    "severity" TEXT NOT NULL DEFAULT 'warning',
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS "idx_discrepancy_severity" ON "discrepancy_alerts"("severity");
+
+-- Rust fund-flow-settlement (port 8251)
+CREATE TABLE IF NOT EXISTS "fx_rates" (
+    "corridor" TEXT PRIMARY KEY,
+    "from_currency" TEXT NOT NULL,
+    "to_currency" TEXT NOT NULL,
+    "rate" DOUBLE PRECISION NOT NULL,
+    "spread_bps" INT NOT NULL,
+    "effective_rate" DOUBLE PRECISION NOT NULL,
+    "updated_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS "installment_schedules" (
+    "application_id" BIGINT PRIMARY KEY,
+    "schedule_json" JSONB NOT NULL,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS "reconciliation_results" (
+    "id" UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    "agent_id" BIGINT NOT NULL,
+    "float_balance" DOUBLE PRECISION NOT NULL,
+    "gl_net" DOUBLE PRECISION NOT NULL,
+    "transaction_total" DOUBLE PRECISION NOT NULL,
+    "discrepancy" DOUBLE PRECISION NOT NULL,
+    "is_reconciled" BOOLEAN NOT NULL,
+    "recommendations" JSONB NOT NULL DEFAULT '[]',
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS "idx_recon_results_agent" ON "reconciliation_results"("agent_id");
+
+CREATE TABLE IF NOT EXISTS "settlement_batches_rust" (
+    "batch_id" TEXT PRIMARY KEY,
+    "total_settlements" INT NOT NULL DEFAULT 0,
+    "total_amount" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "status" TEXT NOT NULL DEFAULT 'initiated',
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Billing projections (for revenue-reconciler real data queries)
+CREATE TABLE IF NOT EXISTS "billing_projections" (
+    "id" SERIAL PRIMARY KEY,
+    "period" TEXT NOT NULL,
+    "agent_id" BIGINT NOT NULL,
+    "projected_tx_count" BIGINT NOT NULL DEFAULT 0,
+    "projected_volume" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "projected_platform_revenue" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "projected_client_revenue" DOUBLE PRECISION NOT NULL DEFAULT 0,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS "idx_billing_projections_period" ON "billing_projections"("period");

server/routers/temporalSagaOrchestrator.ts

@@ -12,7 +12,7 @@ import { z } from "zod";
 import { router, protectedProcedure, adminProcedure } from "../_core/trpc";
 import { getDb } from "../db";
 import { sql } from "drizzle-orm";
-import { publishEvent } from "../kafkaClient";
+import { publishEvent, type KafkaTopic } from "../kafkaClient";
 import { cacheGet, cacheSet, cacheInvalidate } from "../lib/cacheClient";
 import { tbCreateTransfer } from "../tbClient";
 import { fluvioPublish } from "../lib/fluvioClient";
@@ -433,4 +433,134 @@ export const temporalSagaRouter = router({
 
       return rows;
     }),
+
+  startSettlementSaga: protectedProcedure
+    .input(z.object({
+      terminalId: z.string().min(1),
+      batchRef: z.string().optional(),
+      settlementDate: z.string().optional(),
+    }))
+    .mutation(async ({ input }) => {
+      const workflowId = `settle_${input.terminalId}_${Date.now()}`;
+
+      const steps: SagaStep[] = [
+        {
+          name: "create_batch",
+          execute: async (ctx) => {
+            const db = (await getDb())!;
+            if (!db) return { batchId: 0 };
+            const batchRef = ctx.data.batchRef || `BATCH-${Date.now()}`;
+            const result = await db.execute(sql`
+              INSERT INTO pos_settlement_batches (batch_ref, terminal_id, status, created_at)
+              VALUES (${batchRef}, ${ctx.data.terminalId}, 'pending', NOW())
+              RETURNING id
+            `);
+            const batchId = Array.isArray(result) && result[0] ? (result[0] as Record<string, unknown>).id : 0;
+            ctx.results.batchId = batchId;
+            ctx.results.batchRef = batchRef;
+            return { batchId, batchRef };
+          },
+          compensate: async (ctx) => {
+            const db = (await getDb())!;
+            if (db && ctx.results.batchId) {
+              await db.execute(sql`DELETE FROM pos_settlement_batches WHERE id = ${ctx.results.batchId}`);
+            }
+          },
+        },
+        {
+          name: "process_transactions",
+          execute: async (ctx) => {
+            const db = (await getDb())!;
+            if (!db) return { processed: 0 };
+            const txResult = await db.execute(sql`
+              SELECT COUNT(*) as cnt, COALESCE(SUM(amount), 0) as total
+              FROM transactions
+              WHERE terminal_id = ${ctx.data.terminalId}
+                AND status = 'completed'
+            `);
+            const row = Array.isArray(txResult) ? txResult[0] as Record<string, unknown> : {};
+            const cnt = Number(row?.cnt ?? 0);
+            const total = Number(row?.total ?? 0);
+            if (ctx.results.batchId) {
+              await db.execute(sql`
+                UPDATE pos_settlement_batches
+                SET status = 'processing', transaction_count = ${cnt}, total_amount = ${total}
+                WHERE id = ${ctx.results.batchId}
+              `);
+            }
+            return { processed: cnt, totalAmount: total };
+          },
+          compensate: async (ctx) => {
+            const db = (await getDb())!;
+            if (db && ctx.results.batchId) {
+              await db.execute(sql`UPDATE pos_settlement_batches SET status = 'failed' WHERE id = ${ctx.results.batchId}`);
+            }
+          },
+        },
+        {
+          name: "settle_batch",
+          execute: async (ctx) => {
+            const db = (await getDb())!;
+            if (!db) return { settled: false };
+            const settleRef = `SETTLE-${ctx.results.batchRef}-${Date.now()}`;
+            if (ctx.results.batchId) {
+              await db.execute(sql`
+                UPDATE pos_settlement_batches
+                SET status = 'settled', settlement_ref = ${settleRef}, settled_at = NOW()
+                WHERE id = ${ctx.results.batchId}
+              `);
+            }
+            ctx.results.settleRef = settleRef;
+            return { settled: true, settleRef };
+          },
+          compensate: async (ctx) => {
+            const db = (await getDb())!;
+            if (db && ctx.results.batchId) {
+              await db.execute(sql`UPDATE pos_settlement_batches SET status = 'processing', settlement_ref = NULL WHERE id = ${ctx.results.batchId}`);
+            }
+          },
+        },
+        {
+          name: "reconcile_batch",
+          execute: async (ctx) => {
+            const db = (await getDb())!;
+            if (!db) return { reconciled: false };
+            if (ctx.results.batchId) {
+              await db.execute(sql`
+                UPDATE pos_settlement_batches SET status = 'reconciled' WHERE id = ${ctx.results.batchId}
+              `);
+            }
+            return { reconciled: true };
+          },
+          compensate: async (ctx) => {
+            const db = (await getDb())!;
+            if (db && ctx.results.batchId) {
+              await db.execute(sql`UPDATE pos_settlement_batches SET status = 'settled' WHERE id = ${ctx.results.batchId}`);
+            }
+          },
+        },
+      ];
+
+      const sagaCtx: SagaContext = {
+        workflowId,
+        data: {
+          terminalId: input.terminalId,
+          batchRef: input.batchRef,
+          settlementDate: input.settlementDate,
+        },
+        amount: 0,
+        agentId: 0,
+        results: {},
+      };
+
+      const result = await executeSaga("settlement_saga", steps, sagaCtx);
+
+      publishEvent("settlement.saga" as KafkaTopic, workflowId, {
+        workflowId,
+        success: result.success,
+        terminalId: input.terminalId,
+      }).catch(() => {});
+
+      return result;
+    }),
 });

server/routers/transactionReconciliation.ts

@@ -236,4 +236,135 @@ export const transactionReconciliationRouter = router({
       };
     }
   }),
+
+  updateStatus: protectedProcedure
+    .input(
+      z.object({
+        id: z.number().min(1),
+        status: z.string().min(1).max(50),
+        notes: z.string().max(500).optional(),
+      })
+    )
+    .mutation(async ({ input }) => {
+      const database = await getDb();
+      if (!database)
+        throw new TRPCError({ code: "INTERNAL_SERVER_ERROR" });
+
+      const [record] = await database
+        .select()
+        .from(transactions)
+        .where(eq(transactions.id, input.id))
+        .limit(1);
+      if (!record) throw new TRPCError({ code: "NOT_FOUND" });
+
+      enforceTransition(record.status ?? "initiated", input.status);
+
+      const [updated] = await database
+        .update(transactions)
+        .set({ status: input.status, updatedAt: new Date() })
+        .where(eq(transactions.id, input.id))
+        .returning();
+
+      logOperation("STATUS_UPDATED", {
+        transactionId: input.id,
+        from: record.status,
+        to: input.status,
+        notes: input.notes,
+      });
+
+      publishEvent("transaction.reconciliation" as KafkaTopic, String(input.id), {
+        type: "status_updated",
+        transactionId: input.id,
+        from: record.status,
+        to: input.status,
+      }).catch(() => {});
+
+      return { success: true, transaction: updated };
+    }),
+
+  markDisputed: protectedProcedure
+    .input(
+      z.object({
+        id: z.number().min(1),
+        reason: z.string().min(1).max(1000),
+        disputeRef: z.string().max(100).optional(),
+      })
+    )
+    .mutation(async ({ input }) => {
+      const database = await getDb();
+      if (!database)
+        throw new TRPCError({ code: "INTERNAL_SERVER_ERROR" });
+
+      const [record] = await database
+        .select()
+        .from(transactions)
+        .where(eq(transactions.id, input.id))
+        .limit(1);
+      if (!record) throw new TRPCError({ code: "NOT_FOUND" });
+
+      enforceTransition(record.status ?? "completed", "disputed");
+
+      const [updated] = await database
+        .update(transactions)
+        .set({ status: "disputed", updatedAt: new Date() })
+        .where(eq(transactions.id, input.id))
+        .returning();
+
+      logOperation("MARKED_DISPUTED", {
+        transactionId: input.id,
+        reason: input.reason,
+        disputeRef: input.disputeRef,
+      });
+
+      publishEvent("transaction.reconciliation" as KafkaTopic, String(input.id), {
+        type: "disputed",
+        transactionId: input.id,
+        reason: input.reason,
+      }).catch(() => {});
+
+      return { success: true, transaction: updated };
+    }),
+
+  markResolved: protectedProcedure
+    .input(
+      z.object({
+        id: z.number().min(1),
+        resolution: z.string().min(1).max(1000),
+        resolvedBy: z.string().max(200).optional(),
+      })
+    )
+    .mutation(async ({ input }) => {
+      const database = await getDb();
+      if (!database)
+        throw new TRPCError({ code: "INTERNAL_SERVER_ERROR" });
+
+      const [record] = await database
+        .select()
+        .from(transactions)
+        .where(eq(transactions.id, input.id))
+        .limit(1);
+      if (!record) throw new TRPCError({ code: "NOT_FOUND" });
+
+      enforceTransition(record.status ?? "disputed", "resolved");
+
+      const [updated] = await database
+        .update(transactions)
+        .set({ status: "resolved", updatedAt: new Date() })
+        .where(eq(transactions.id, input.id))
+        .returning();
+
+      logOperation("MARKED_RESOLVED", {
+        transactionId: input.id,
+        resolution: input.resolution,
+        resolvedBy: input.resolvedBy,
+      });
+
+      publishEvent("transaction.reconciliation" as KafkaTopic, String(input.id), {
+        type: "resolved",
+        transactionId: input.id,
+        resolution: input.resolution,
+      }).catch(() => {});
+
+      return { success: true, transaction: updated };
+    }),
 });