feat: KYC expiry notifications, admin confirmation, corridor analytic…

api/src/app.ts

@@ -2,13 +2,15 @@ import express, { Application, Request, Response, NextFunction } from 'express';
 import helmet from 'helmet';
 import cors from 'cors';
 import rateLimit from 'express-rate-limit';
+import { Pool } from 'pg';
 import currenciesRouter from './routes/currencies';
 import limitsRouter from './routes/limits';
 import { createAnchorsRouter } from './routes/anchors';
 import docsRouter from './routes/docs';
 import settlementsRouter from './routes/settlements';
 import { createRemittancesRouter, RemittancesRouterOptions } from './routes/remittances';
 import { createAdminRouter } from './routes/admin';
+import { createAnalyticsRouter } from './routes/analytics';
 import { ErrorResponse } from './types';
 import { AnchorStore } from './db/anchorStore';
 import { Server as SocketIOServer } from 'socket.io';
@@ -78,6 +80,14 @@ export function createApp(options: AppOptions = {}): Application {
   // Admin utilities — read-only operations (simulate-upgrade, etc.)
   app.use('/api/admin', createAdminRouter());
 
+  // Corridor analytics (Issue #482)
+  const analyticsPool = process.env.DATABASE_URL
+    ? new Pool({ connectionString: process.env.DATABASE_URL, max: 5 })
+    : null;
+  if (analyticsPool) {
+    app.use('/api/analytics', createAnalyticsRouter(analyticsPool));
+  }
+
   // API documentation
   app.use('/api/docs', docsRouter);
 

api/src/routes/admin.ts

@@ -1,5 +1,7 @@
 import { Router, Request, Response } from 'express';
 import { ErrorResponse } from '../types';
+import { Pool } from 'pg';
+import { AdminConfirmationService, HighRiskOperation } from '../../../backend/src/admin-confirmation';
 
 function timestamp(): string {
   return new Date().toISOString();
@@ -92,6 +94,15 @@ function simulateUpgrade(wasmHashHex: string): {
   };
 }
 
+const HIGH_RISK_OPS: HighRiskOperation[] = ['withdraw_fees', 'remove_agent', 'update_fee'];
+
+function getConfirmationService(): AdminConfirmationService | null {
+  const dbUrl = process.env.DATABASE_URL;
+  if (!dbUrl) return null;
+  const pool = new Pool({ connectionString: dbUrl });
+  return new AdminConfirmationService(pool);
+}
+
 export function createAdminRouter(): Router {
   const router = Router();
 
@@ -207,5 +218,93 @@ export function createAdminRouter(): Router {
     });
   });
 
+  // ── Multi-step admin confirmation (#481) ──────────────────────────────────
+
+  /**
+   * POST /api/admin/actions
+   * Initiate a high-risk operation requiring a second admin to confirm.
+   */
+  router.post('/actions', async (req: Request, res: Response) => {
+    if (!isAdminAuthorized(req)) {
+      return sendError(res, 401, 'Admin authentication required', 'UNAUTHORIZED');
+    }
+
+    const { operation, initiated_by, params } = req.body as Record<string, unknown>;
+
+    if (!operation || !HIGH_RISK_OPS.includes(operation as HighRiskOperation)) {
+      return sendError(res, 400, `operation must be one of: ${HIGH_RISK_OPS.join(', ')}`, 'INVALID_OPERATION');
+    }
+    if (typeof initiated_by !== 'string' || !initiated_by) {
+      return sendError(res, 400, 'initiated_by is required', 'MISSING_FIELD');
+    }
+
+    const svc = getConfirmationService();
+    if (!svc) return sendError(res, 503, 'Database not configured', 'DB_UNAVAILABLE');
+
+    try {
+      await svc.initTable();
+      const action = await svc.initiate(
+        operation as HighRiskOperation,
+        initiated_by,
+        (params as Record<string, unknown>) ?? {}
+      );
+      return res.status(201).json({ success: true, data: action, timestamp: timestamp() });
+    } catch (err) {
+      return sendError(res, 500, err instanceof Error ? err.message : 'Failed to initiate action', 'INITIATE_FAILED');
+    }
+  });
+
+  /**
+   * GET /api/admin/actions
+   * List all pending (unconfirmed, non-expired) high-risk actions.
+   */
+  router.get('/actions', async (req: Request, res: Response) => {
+    if (!isAdminAuthorized(req)) {
+      return sendError(res, 401, 'Admin authentication required', 'UNAUTHORIZED');
+    }
+
+    const svc = getConfirmationService();
+    if (!svc) return sendError(res, 503, 'Database not configured', 'DB_UNAVAILABLE');
+
+    try {
+      await svc.initTable();
+      const actions = await svc.listPending();
+      return res.json({ success: true, data: actions, timestamp: timestamp() });
+    } catch (err) {
+      return sendError(res, 500, err instanceof Error ? err.message : 'Failed to list actions', 'LIST_FAILED');
+    }
+  });
+
+  /**
+   * POST /api/admin/actions/:id/confirm
+   * Second admin confirms a pending high-risk action.
+   */
+  router.post('/actions/:id/confirm', async (req: Request, res: Response) => {
+    if (!isAdminAuthorized(req)) {
+      return sendError(res, 401, 'Admin authentication required', 'UNAUTHORIZED');
+    }
+
+    const { confirmed_by } = req.body as Record<string, unknown>;
+    if (typeof confirmed_by !== 'string' || !confirmed_by) {
+      return sendError(res, 400, 'confirmed_by is required', 'MISSING_FIELD');
+    }
+
+    const svc = getConfirmationService();
+    if (!svc) return sendError(res, 503, 'Database not configured', 'DB_UNAVAILABLE');
+
+    try {
+      await svc.initTable();
+      const action = await svc.confirm(req.params.id, confirmed_by);
+      return res.json({ success: true, data: action, timestamp: timestamp() });
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : 'Confirmation failed';
+      const isNotFound = msg.includes('not found');
+      const isExpired = msg.includes('expired');
+      const isSelf = msg.includes('cannot confirm');
+      const status = isNotFound ? 404 : isExpired || isSelf ? 409 : 500;
+      return sendError(res, status, msg, 'CONFIRM_FAILED');
+    }
+  });
+
   return router;
 }

api/src/routes/analytics.ts

@@ -0,0 +1,136 @@
+/**
+ * GET /api/analytics/corridors (#482)
+ *
+ * Returns remittance volume, fees, and success/failure rates per corridor
+ * (currency/country pair) sourced from the contract_events table.
+ *
+ * Query params:
+ *   range  {string}  - Time range: 7d | 30d | 90d (default: 30d)
+ */
+
+import { Router, Request, Response } from 'express';
+import { Pool } from 'pg';
+import { ErrorResponse } from '../types';
+
+export interface CorridorStat {
+  source_currency: string;
+  destination_country: string;
+  total_volume: number;
+  transaction_count: number;
+  success_count: number;
+  failure_count: number;
+  success_rate: number;
+  avg_fee: number;
+  total_fees: number;
+}
+
+export interface CorridorAnalyticsResponse {
+  success: true;
+  data: {
+    range: string;
+    corridors: CorridorStat[];
+    top_by_volume: CorridorStat[];
+  };
+  timestamp: string;
+}
+
+const VALID_RANGES: Record<string, string> = {
+  '7d': '7 days',
+  '30d': '30 days',
+  '90d': '90 days',
+};
+
+function timestamp(): string {
+  return new Date().toISOString();
+}
+
+function sendError(res: Response, status: number, message: string, code: string): Response<ErrorResponse> {
+  return res.status(status).json({ success: false, error: { message, code }, timestamp: timestamp() });
+}
+
+export function createAnalyticsRouter(pool: Pool): Router {
+  const router = Router();
+
+  router.get('/corridors', async (req: Request, res: Response) => {
+    const rangeParam = typeof req.query.range === 'string' ? req.query.range : '30d';
+
+    if (!VALID_RANGES[rangeParam]) {
+      return sendError(res, 400, `range must be one of: ${Object.keys(VALID_RANGES).join(', ')}`, 'INVALID_RANGE');
+    }
+
+    const interval = VALID_RANGES[rangeParam];
+
+    try {
+      // contract_events stores raw_data JSONB with currency/country fields when available.
+      // We aggregate by (source_currency, destination_country) from raw_data.
+      const result = await pool.query<{
+        source_currency: string;
+        destination_country: string;
+        total_volume: string;
+        transaction_count: string;
+        success_count: string;
+        failure_count: string;
+        avg_fee: string;
+        total_fees: string;
+      }>(        `SELECT
+           COALESCE(raw_data->>'source_currency', raw_data->>'currency', 'USDC') AS source_currency,
+           COALESCE(raw_data->>'destination_country', raw_data->>'country', 'UNKNOWN') AS destination_country,
+           SUM(COALESCE(amount, 0))                                                    AS total_volume,
+           COUNT(*)                                                                    AS transaction_count,
+           COUNT(*) FILTER (WHERE event_type = 'remittance_completed')                AS success_count,
+           COUNT(*) FILTER (WHERE event_type IN ('remittance_failed', 'remittance_cancelled')) AS failure_count,
+           AVG(COALESCE(fee, 0))                                                       AS avg_fee,
+           SUM(COALESCE(fee, 0))                                                       AS total_fees
+         FROM contract_events
+         WHERE timestamp >= NOW() - INTERVAL '${interval}'
+           AND event_type IN ('remittance_created', 'remittance_completed', 'remittance_failed', 'remittance_cancelled')
+         GROUP BY source_currency, destination_country
+         ORDER BY total_volume DESC NULLS LAST`,
+        []
+      );
+
+      const corridors: CorridorStat[] = result.rows.map((row: {
+        source_currency: string;
+        destination_country: string;
+        total_volume: string;
+        transaction_count: string;
+        success_count: string;
+        failure_count: string;
+        avg_fee: string;
+        total_fees: string;
+      }) => {
+        const total = parseInt(row.transaction_count, 10);
+        const success = parseInt(row.success_count, 10);
+        return {
+          source_currency: row.source_currency,
+          destination_country: row.destination_country,
+          total_volume: parseFloat(row.total_volume) || 0,
+          transaction_count: total,
+          success_count: success,
+          failure_count: parseInt(row.failure_count, 10),
+          success_rate: total > 0 ? Math.round((success / total) * 10000) / 100 : 0,
+          avg_fee: parseFloat(row.avg_fee) || 0,
+          total_fees: parseFloat(row.total_fees) || 0,
+        };
+      });
+
+      const top_by_volume = [...corridors]
+        .sort((a, b) => b.total_volume - a.total_volume)
+        .slice(0, 10);
+
+      const response: CorridorAnalyticsResponse = {
+        success: true,
+        data: { range: rangeParam, corridors, top_by_volume },
+        timestamp: timestamp(),
+      };
+
+      return res.json(response);
+    } catch (err) {
+      // eslint-disable-next-line no-console
+      void err;
+      return sendError(res, 500, 'Failed to fetch corridor analytics', 'ANALYTICS_ERROR');
+    }
+  });
+
+  return router;
+}