feat: add request observability and Stellar reconciliation worker

.env.example

@@ -29,6 +29,16 @@ JWT_SECRET=your-super-secret-jwt-key
 JWT_EXPIRES_IN=15m
 AUTH_CHALLENGE_TTL_MS=300000
 
+# Observability
+METRICS_ENABLED=true
+
+# Background reconciliation
+STELLAR_RECONCILIATION_ENABLED=false
+STELLAR_RECONCILIATION_INTERVAL_MS=30000
+STELLAR_RECONCILIATION_BATCH_SIZE=25
+STELLAR_RECONCILIATION_GRACE_PERIOD_MS=60000
+STELLAR_RECONCILIATION_MAX_RUNTIME_MS=10000
+
 # Email
 SENDGRID_API_KEY=SG.xxxxxxxxxxxxx
 FROM_EMAIL=noreply@stellarsettle.com

Readme.md

@@ -111,6 +111,16 @@ JWT_SECRET=your-super-secret-jwt-key
 JWT_EXPIRES_IN=15m
 AUTH_CHALLENGE_TTL_MS=300000
 
+# Observability
+METRICS_ENABLED=true
+
+# Background reconciliation
+STELLAR_RECONCILIATION_ENABLED=false
+STELLAR_RECONCILIATION_INTERVAL_MS=30000
+STELLAR_RECONCILIATION_BATCH_SIZE=25
+STELLAR_RECONCILIATION_GRACE_PERIOD_MS=60000
+STELLAR_RECONCILIATION_MAX_RUNTIME_MS=10000
+
 # Email
 SENDGRID_API_KEY=SG.xxxxxxxxxxxxx
 FROM_EMAIL=noreply@stellarsettle.com
@@ -199,9 +209,16 @@ npm run test:e2e
 
 ## 📊 Monitoring
 
-- Health check: `GET /health`
+- Health check: `GET /health` (includes process uptime and request ID)
 - Metrics: `GET /metrics` (Prometheus format)
-- Logs: Winston with daily rotation
+- Metrics labels are intentionally low-cardinality: `method`, normalized route template, and `status_class`
+- Logs: Winston JSON logs with `X-Request-Id` correlation IDs
+
+## Background Reconciliation
+
+- Enable `STELLAR_RECONCILIATION_ENABLED=true` to start the in-process worker.
+- The worker scans a bounded batch of stale pending investments / transactions and reuses the existing Stellar payment verification path for idempotent reconciliation.
+- Current deployment assumption: run the worker on a single replica unless you add your own leader-election or advisory-lock strategy.
 
 ## 🚢 Deployment
 ```bash

src/app.ts

@@ -1,29 +1,58 @@
 import cors from "cors";
 import express from "express";
 import helmet from "helmet";
-import { errorMiddleware, notFoundMiddleware } from "./middleware/error.middleware";
+import { createErrorMiddleware, notFoundMiddleware } from "./middleware/error.middleware";
+import { createRequestObservabilityMiddleware } from "./middleware/request-observability.middleware";
+import { logger, type AppLogger } from "./observability/logger";
+import { getMetricsContentType, MetricsRegistry } from "./observability/metrics";
 import { createAuthRouter } from "./routes/auth.routes";
 import type { AuthService } from "./services/auth.service";
 
 export interface AppDependencies {
   authService: AuthService;
+  logger?: AppLogger;
+  metricsEnabled?: boolean;
+  metricsRegistry?: MetricsRegistry;
 }
 
-export function createApp({ authService }: AppDependencies) {
+export function createApp({
+  authService,
+  logger: appLogger = logger,
+  metricsEnabled = true,
+  metricsRegistry = new MetricsRegistry(),
+}: AppDependencies) {
   const app = express();
 
   app.use(helmet());
   app.use(cors());
   app.use(express.json());
+  app.use(
+    createRequestObservabilityMiddleware({
+      logger: appLogger,
+      metricsEnabled,
+      metricsRegistry,
+    }),
+  );
 
-  app.get("/health", (_req, res) => {
-    res.status(200).json({ status: "ok" });
+  app.get("/health", (req, res) => {
+    res.status(200).json({
+      status: "ok",
+      uptimeSeconds: Number(process.uptime().toFixed(3)),
+      requestId: req.requestId,
+    });
   });
 
+  if (metricsEnabled) {
+    app.get("/metrics", (_req, res) => {
+      res.setHeader("Content-Type", getMetricsContentType());
+      res.status(200).send(metricsRegistry.renderPrometheusMetrics());
+    });
+  }
+
   app.use("/api/v1/auth", createAuthRouter(authService));
 
   app.use(notFoundMiddleware);
-  app.use(errorMiddleware);
+  app.use(createErrorMiddleware(appLogger));
 
   return app;
 }

src/config/database.ts

@@ -1,3 +1,4 @@
+import "dotenv/config";
 import "reflect-metadata";
 import { DataSource } from "typeorm";
 

src/config/env.ts

@@ -1,3 +1,4 @@
+import "dotenv/config";
 import { Networks } from "stellar-sdk";
 
 type SupportedStellarNetwork = "testnet" | "mainnet" | "futurenet";
@@ -12,6 +13,16 @@ export interface AppConfig {
   auth: {
     challengeTtlMs: number;
   };
+  observability: {
+    metricsEnabled: boolean;
+  };
+  reconciliation: {
+    enabled: boolean;
+    intervalMs: number;
+    batchSize: number;
+    gracePeriodMs: number;
+    maxRuntimeMs: number;
+  };
   stellar: {
     network: SupportedStellarNetwork;
     networkPassphrase: string;
@@ -21,6 +32,12 @@ export interface AppConfig {
 const DEFAULT_PORT = 3000;
 const DEFAULT_JWT_EXPIRES_IN = "15m";
 const DEFAULT_CHALLENGE_TTL_MS = 5 * 60 * 1000;
+const DEFAULT_METRICS_ENABLED = true;
+const DEFAULT_RECONCILIATION_ENABLED = false;
+const DEFAULT_RECONCILIATION_INTERVAL_MS = 30 * 1000;
+const DEFAULT_RECONCILIATION_BATCH_SIZE = 25;
+const DEFAULT_RECONCILIATION_GRACE_PERIOD_MS = 60 * 1000;
+const DEFAULT_RECONCILIATION_MAX_RUNTIME_MS = 10 * 1000;
 
 function parsePort(value: string | undefined): number {
   if (!value) {
@@ -36,18 +53,55 @@ function parsePort(value: string | undefined): number {
   return port;
 }
 
-function parseChallengeTtl(value: string | undefined): number {
+function parsePositiveInteger(
+  value: string | undefined,
+  fallback: number,
+  name: string,
+): number {
   if (!value) {
-    return DEFAULT_CHALLENGE_TTL_MS;
+    return fallback;
   }
 
-  const ttl = Number(value);
+  const parsedValue = Number(value);
 
-  if (!Number.isInteger(ttl) || ttl <= 0) {
-    throw new Error("AUTH_CHALLENGE_TTL_MS must be a positive integer.");
+  if (!Number.isInteger(parsedValue) || parsedValue <= 0) {
+    throw new Error(`${name} must be a positive integer.`);
   }
 
-  return ttl;
+  return parsedValue;
+}
+
+function parseChallengeTtl(value: string | undefined): number {
+  return parsePositiveInteger(
+    value,
+    DEFAULT_CHALLENGE_TTL_MS,
+    "AUTH_CHALLENGE_TTL_MS",
+  );
+}
+
+function parseBoolean(
+  value: string | undefined,
+  fallback: boolean,
+  name: string,
+): boolean {
+  if (!value) {
+    return fallback;
+  }
+
+  switch (value.toLowerCase()) {
+    case "true":
+    case "1":
+    case "yes":
+    case "on":
+      return true;
+    case "false":
+    case "0":
+    case "no":
+    case "off":
+      return false;
+    default:
+      throw new Error(`${name} must be a boolean.`);
+  }
 }
 
 function resolveNetwork(network: string | undefined): AppConfig["stellar"] {
@@ -94,6 +148,40 @@ export function getConfig(): AppConfig {
     auth: {
       challengeTtlMs: parseChallengeTtl(process.env.AUTH_CHALLENGE_TTL_MS),
     },
+    observability: {
+      metricsEnabled: parseBoolean(
+        process.env.METRICS_ENABLED,
+        DEFAULT_METRICS_ENABLED,
+        "METRICS_ENABLED",
+      ),
+    },
+    reconciliation: {
+      enabled: parseBoolean(
+        process.env.STELLAR_RECONCILIATION_ENABLED,
+        DEFAULT_RECONCILIATION_ENABLED,
+        "STELLAR_RECONCILIATION_ENABLED",
+      ),
+      intervalMs: parsePositiveInteger(
+        process.env.STELLAR_RECONCILIATION_INTERVAL_MS,
+        DEFAULT_RECONCILIATION_INTERVAL_MS,
+        "STELLAR_RECONCILIATION_INTERVAL_MS",
+      ),
+      batchSize: parsePositiveInteger(
+        process.env.STELLAR_RECONCILIATION_BATCH_SIZE,
+        DEFAULT_RECONCILIATION_BATCH_SIZE,
+        "STELLAR_RECONCILIATION_BATCH_SIZE",
+      ),
+      gracePeriodMs: parsePositiveInteger(
+        process.env.STELLAR_RECONCILIATION_GRACE_PERIOD_MS,
+        DEFAULT_RECONCILIATION_GRACE_PERIOD_MS,
+        "STELLAR_RECONCILIATION_GRACE_PERIOD_MS",
+      ),
+      maxRuntimeMs: parsePositiveInteger(
+        process.env.STELLAR_RECONCILIATION_MAX_RUNTIME_MS,
+        DEFAULT_RECONCILIATION_MAX_RUNTIME_MS,
+        "STELLAR_RECONCILIATION_MAX_RUNTIME_MS",
+      ),
+    },
     stellar: resolveNetwork(process.env.STELLAR_NETWORK),
   };
 }

src/index.ts

@@ -1,23 +1,105 @@
+import type { Server } from "http";
 import { createApp } from "./app";
-import { getConfig } from "./config/env";
 import dataSource from "./config/database";
+import { getConfig } from "./config/env";
+import { getPaymentVerificationConfig } from "./config/stellar";
+import { logger } from "./observability/logger";
 import { createAuthService } from "./services/auth.service";
+import { createVerifyPaymentService } from "./services/stellar/verify-payment.service";
+import { createReconcilePendingStellarStateWorker } from "./workers/reconcile-pending-stellar-state.worker";
+
+export interface ApplicationRuntime {
+  stop(signal?: string): Promise<void>;
+  server: Server;
+}
+
+function closeServer(server: Server): Promise<void> {
+  return new Promise((resolve, reject) => {
+    server.close((error) => {
+      if (error) {
+        reject(error);
+        return;
+      }
 
-export async function bootstrap(): Promise<void> {
+      resolve();
+    });
+  });
+}
+
+export async function bootstrap(): Promise<ApplicationRuntime> {
   const config = getConfig();
 
   if (!dataSource.isInitialized) {
     await dataSource.initialize();
   }
 
   const authService = createAuthService(dataSource, config);
-  const app = createApp({ authService });
+  const app = createApp({
+    authService,
+    logger,
+    metricsEnabled: config.observability.metricsEnabled,
+  });
+  const server = await new Promise<Server>((resolve) => {
+    const listeningServer = app.listen(config.port, () => {
+      logger.info("StellarSettle API listening.", {
+        port: config.port,
+        metricsEnabled: config.observability.metricsEnabled,
+      });
+      resolve(listeningServer);
+    });
+  });
+
+  const reconciliationWorker = config.reconciliation.enabled
+    ? createReconcilePendingStellarStateWorker(
+        dataSource,
+        createVerifyPaymentService(dataSource, getPaymentVerificationConfig()),
+        config.reconciliation,
+        logger,
+      )
+    : null;
 
-  app.listen(config.port, () => {
-    process.stdout.write(`StellarSettle API listening on port ${config.port}\n`);
+  reconciliationWorker?.start();
+
+  let shutdownPromise: Promise<void> | null = null;
+
+  const stop = async (signal = "manual"): Promise<void> => {
+    if (shutdownPromise) {
+      return shutdownPromise;
+    }
+
+    shutdownPromise = (async () => {
+      logger.info("Shutting down StellarSettle API.", { signal });
+      await reconciliationWorker?.stop();
+      await closeServer(server);
+
+      if (dataSource.isInitialized) {
+        await dataSource.destroy();
+      }
+
+      logger.info("StellarSettle API stopped.", { signal });
+    })();
+
+    return shutdownPromise;
+  };
+
+  process.once("SIGTERM", () => {
+    void stop("SIGTERM");
+  });
+  process.once("SIGINT", () => {
+    void stop("SIGINT");
   });
+
+  return {
+    stop,
+    server,
+  };
 }
 
 if (require.main === module) {
-  void bootstrap();
+  void bootstrap().catch((error: unknown) => {
+    logger.error("Failed to bootstrap StellarSettle API.", {
+      error: error instanceof Error ? error.message : "Unknown error",
+    });
+    process.exitCode = 1;
+  });
 }