feat: implement account activity anomaly detection

migrations/20260630_create_security_events.sql

@@ -0,0 +1,34 @@
+-- Security Events table for audit and anomaly detection
+CREATE TABLE IF NOT EXISTS security_events (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id UUID REFERENCES users(id) ON DELETE SET NULL,
+  api_key_id UUID REFERENCES api_keys(id) ON DELETE SET NULL,
+  event_type TEXT NOT NULL,
+  severity TEXT NOT NULL CHECK (severity IN ('low', 'medium', 'high', 'critical')),
+  ip_address INET,
+  country_code CHAR(2),
+  user_agent TEXT,
+  metadata JSONB,
+  acknowledged BOOLEAN DEFAULT false,
+  acknowledged_at TIMESTAMP,
+  acknowledged_by UUID REFERENCES users(id),
+  created_at TIMESTAMP DEFAULT NOW(),
+  updated_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_security_events_user_id ON security_events(user_id);
+CREATE INDEX IF NOT EXISTS idx_security_events_event_type ON security_events(event_type);
+CREATE INDEX IF NOT EXISTS idx_security_events_created_at ON security_events(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_security_events_country_code ON security_events(country_code);
+CREATE INDEX IF NOT EXISTS idx_security_events_acknowledged ON security_events(acknowledged, created_at DESC);
+
+-- Account Activity Baseline table for tracking historical patterns
+CREATE TABLE IF NOT EXISTS account_activity_baseline (
+  user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+  countries TEXT[] DEFAULT '{}',
+  ip_addresses TEXT[] DEFAULT '{}',
+  typical_hours JSONB,
+  last_updated TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_account_activity_baseline_user_id ON account_activity_baseline(user_id);

src/auth/2fa.ts

@@ -131,7 +131,7 @@ export async function verifyBackupCode(
  * @returns True if 2FA is enabled
  */
 interface TwoFactorUser {
-  two_factor_secret?: string;
+  two_factor_secret?: string | null;
   two_factor_enabled?: boolean;
   two_factor_verified?: boolean;
 }

src/auth/oauth.ts

@@ -236,7 +236,7 @@ function createRedisBackedStore(
           return null;
         }
         await redisClient.del(key);
-        const rawStr = typeof raw === "string" ? raw : raw.toString();
+        const rawStr = raw;
         const record = JSON.parse(rawStr) as AuthorizationCodeRecord;
         return record.expiresAt > Date.now() ? record : null;
       }
@@ -269,7 +269,7 @@ function createRedisBackedStore(
         if (!raw) {
           return null;
         }
-        const rawStr = typeof raw === "string" ? raw : raw.toString();
+        const rawStr = raw;
         const record = JSON.parse(rawStr) as RefreshTokenRecord;
         return record.expiresAt > Date.now() ? record : null;
       }

src/config/index.ts

@@ -8,19 +8,19 @@
 // Initialize config system - must be imported first
 import './init';
 import { getConfigValue } from './appConfig';
-import { PROVIDER_LIMITS } from './providers';
-import { TRANSACTION_LIMITS } from './limits';
+import { PROVIDER_LIMITS, MobileMoneyProvider } from './providers';
+import { TRANSACTION_LIMITS, KYCLevel } from './limits';
 
 export { getConfig, getConfigValue } from './appConfig';
 export { PROVIDER_LIMITS, getProviderLimitsConfig, MobileMoneyProvider } from './providers';
 export { TRANSACTION_LIMITS, MIN_TRANSACTION_AMOUNT, MAX_TRANSACTION_AMOUNT, KYCLevel } from './limits';
 
 // Helper functions for commonly accessed config values
-export function getProviderLimit(provider: string): { minAmount: number; maxAmount: number } | null {
+export function getProviderLimit(provider: MobileMoneyProvider): { minAmount: number; maxAmount: number } | null {
   return PROVIDER_LIMITS[provider] || null;
 }
 
-export function getKycLimit(level: string): number | null {
+export function getKycLimit(level: KYCLevel): number | null {
   return TRANSACTION_LIMITS[level] || null;
 }
 

src/constants/errorCodes.ts

@@ -53,6 +53,8 @@ export const ERROR_CODES = {
   RATE_LIMIT: "RATE_LIMIT",
   /** Destination Stellar account has not established a trustline for the payment asset. */
   TRUSTLINE_MISSING: "TRUSTLINE_MISSING",
+  /** Quote has expired or is invalid. */
+  QUOTE_EXPIRED: "QUOTE_EXPIRED",
 
   // Server errors (5000+) - HTTP 500+
   INTERNAL_ERROR: "INTERNAL_ERROR",
@@ -134,7 +136,8 @@ export const getHttpStatus = (code: string): number => {
     code === ERROR_CODES.INSUFFICIENT_BALANCE ||
     code === ERROR_CODES.INSUFFICIENT_FUNDS ||
     code === ERROR_CODES.TRANSACTION_FAILED ||
-    code === ERROR_CODES.TRUSTLINE_MISSING
+    code === ERROR_CODES.TRUSTLINE_MISSING ||
+    code === ERROR_CODES.QUOTE_EXPIRED
   ) {
     return 400;
   }

src/index.ts

@@ -78,6 +78,7 @@ import { createAdminSep10Router } from "./stellar/adminSep10";
 import tomlRouter from "./routes/toml";
 import feeStrategiesRouter from "./routes/feeStrategies";
 import crossChainRouter from "./routes/crossChain";
+import securityRouter from "./routes/security";
 import stellarRouter from "./routes/stellar";
 import reconciliationRoutes from "./routes/reconciliation";
 import accountingReconciliationRoutes from "./routes/accountingReconciliation";
@@ -385,6 +386,7 @@ app.use("/api/admin/assets", adminAssetRoutes);
 app.use("/api/settings", settingsRoutes);
 app.use("/api/statements", statementsRoutes);
 app.use("/", paymentLinkRoutes);
+app.use("/security", securityRouter);
 
 // GDPR
 app.use("/api/gdpr", privacyRoutes);

src/jobs/anomalyDetectionJob.ts

@@ -0,0 +1,59 @@
+import { Pool } from "pg";
+import { queryRead, queryWrite } from "../config/database";
+import { securityAnomalyService } from "../services/securityAnomalyService";
+
+interface BulkOperationRecord {
+  userId: string;
+  ipAddress: string;
+  createdAt: Date;
+}
+
+const BULK_OPERATION_THRESHOLD = 10;
+const UNUSUAL_HOURS_START = 2;
+const UNUSUAL_HOURS_END = 5;
+
+export async function runAnomalyDetectionJob(): Promise<void> {
+  const now = new Date();
+  const currentHour = now.getHours();
+  const isUnusualHours = currentHour >= UNUSUAL_HOURS_START && currentHour <= UNUSUAL_HOURS_END;
+
+  if (isUnusualHours) {
+    const result = await queryRead<BulkOperationRecord>(
+      `SELECT user_id as "userId", ip_address as "ipAddress", created_at as "createdAt"
+       FROM transactions 
+       WHERE created_at > NOW() - INTERVAL '1 hour'
+       GROUP BY user_id, ip_address, created_at
+       HAVING COUNT(*) >= $1`,
+      [BULK_OPERATION_THRESHOLD]
+    );
+
+    for (const record of result.rows) {
+      const hourCount = await queryRead<{ count: number }>(
+        `SELECT COUNT(*) as count FROM transactions 
+         WHERE user_id = $1 AND date_trunc('hour', created_at) = date_trunc('hour', NOW())`,
+        [record.userId]
+      );
+
+      if (hourCount.rows[0]?.count >= BULK_OPERATION_THRESHOLD) {
+        await securityAnomalyService.createSecurityEvent({
+          userId: record.userId,
+          eventType: "bulk_operation_unusual_hours",
+          severity: "medium",
+          ipAddress: record.ipAddress,
+          metadata: {
+            operationCount: hourCount.rows[0].count,
+            hour: currentHour,
+          },
+        });
+      }
+    }
+  }
+
+  const userResult = await queryRead(
+    `SELECT id FROM users WHERE status = 'active'`
+  );
+
+  for (const user of userResult.rows) {
+    await securityAnomalyService.buildBaselineFromHistory(user.id);
+  }
+}

src/jobs/scheduler.ts

@@ -27,6 +27,7 @@ import {
 import { runIndexReindexJob } from "./indexReindexJob";
 import { runSanctionSyncJob } from "./sanctionSyncJob";
 import { startNotificationWorker } from "../workers/notificationWorker";
+import { runAnomalyDetectionJob } from "./anomalyDetectionJob";
 
 interface JobConfig {
   name: string;
@@ -37,10 +38,14 @@ interface JobConfig {
 const JOBS: JobConfig[] = [
   {
     name: "sanction-sync",
-    // Daily at 1:00 AM - syncs internal sanction list with global lists
     schedule: process.env.SANCTION_SYNC_CRON || "0 1 * * *",
     handler: runSanctionSyncJob,
   },
+  {
+    name: "anomaly-detection",
+    schedule: process.env.ANOMALY_DETECTION_CRON || "*/15 * * * *",
+    handler: runAnomalyDetectionJob,
+  },
   {
     name: "cleanup",
     // Daily at 2:00 AM - deletes old completed/failed transactions

src/locales/en.json

@@ -24,6 +24,7 @@
     "INTERNAL_ERROR": "Internal server error",
     "SERVICE_UNAVAILABLE": "Service temporarily unavailable",
     "DATABASE_ERROR": "Database operation failed",
+    "QUOTE_EXPIRED": "Quote expired or invalid",
     "DEFAULT": "An error occurred"
   },
   "sms": {

src/middleware/auth.ts

@@ -7,6 +7,8 @@ import { getAdminSep10Service } from "../stellar/adminSep10";
 import { evaluateGeoLoginAccess } from "../auth/geo";
 import { queryRead } from "../config/database";
 import { ScopeGroup } from "../auth/apikeys";
+import { securityAnomalyService, SecurityEventSeverity } from "../services/securityAnomalyService";
+import { getCurrentRequestIp } from "../services/logger";
 
 type RequestUser = {
   id: string;
@@ -170,11 +172,6 @@ export const requireAuth = async (
   });
 };
 
-/**
- * JWT Authentication middleware that verifies JWT tokens
- * and attaches user information to the request object
- * Includes IP geofencing validation for operational regions
- */
 export async function authenticateToken(
   req: Request,
   res: Response,
@@ -207,6 +204,42 @@ export async function authenticateToken(
       return;
     }
 
+    // Account Activity Anomaly Detection
+    if (decoded.userId) {
+      const ipAddress = getCurrentRequestIp(req);
+      const countryCode = await securityAnomalyService.getCountryFromIp(
+        typeof ipAddress === "string" ? ipAddress : ""
+      );
+
+      // Check for new country login anomaly
+      const countryAnomaly = await securityAnomalyService.detectCountryAnomaly(
+        decoded.userId,
+        typeof ipAddress === "string" ? ipAddress : "",
+        countryCode || "",
+        req.get("user-agent")
+      );
+
+      if (countryAnomaly.isAnomaly && countryAnomaly.requiresBlock) {
+        res.status(403).json({
+          error: "Access denied",
+          message: "Login from this location requires approval. Check your email.",
+        });
+        return;
+      }
+
+      // Check for new IP login anomaly
+      const loginAnomaly = await securityAnomalyService.detectLoginAnomaly(
+        decoded.userId,
+        typeof ipAddress === "string" ? ipAddress : "",
+        req.get("user-agent")
+      );
+
+      if (loginAnomaly.isAnomaly) {
+        // Non-blocking notification for new IPs
+        console.log(`[Security] New IP login detected for user ${decoded.userId}`);
+      }
+    }
+
     req.jwtUser = decoded;
     next();
   } catch (error) {

src/middleware/errorHandler.ts

@@ -95,10 +95,10 @@ const extractLegacyDetails = (err: AppError): Record<string, unknown> => {
  */
 export const createError = (
   code: string,
-  message?: string,
+  message?: string | null,
   details?: Record<string, unknown>,
 ): AppError => {
-  const error: AppError = new Error(message);
+  const error: AppError = new Error(message ?? undefined);
   error.code = code;
   error.statusCode = getHttpStatus(code);
   error.details = details;

src/middleware/rateLimitRedis.ts

@@ -41,7 +41,7 @@ export async function rateLimitMiddleware(req: Request, res: Response, next: Nex
   try {
     await limiter.consume(key);
     next();
-  } catch (rejRes) {
+  } catch (rejRes: any) {
     const retrySecs = Math.round(rejRes.msBeforeNext / 1000) || 1;
     res.set("Retry-After", String(retrySecs));
     res.status(429).json({

src/models/adminStellarKey.ts

@@ -162,19 +162,19 @@ export class AdminStellarKeyModel {
        WHERE public_key = $1 AND is_active = true`,
       [publicKey]
     );
-    return result.rowCount > 0;
-  }
+return result.rowCount !== null && result.rowCount > 0;
+   }
 
-  /**
-   * Delete an admin Stellar key
-   */
-  async delete(publicKey: string): Promise<boolean> {
+   /**
+    * Delete an admin Stellar key
+    */
+   async delete(publicKey: string): Promise<boolean> {
     const result = await queryWrite(
       "DELETE FROM admin_stellar_keys WHERE public_key = $1",
       [publicKey]
     );
-    return result.rowCount > 0;
-  }
+    return result.rowCount !== null && result.rowCount > 0;
+   }
 }
 
 export const adminStellarKeyModel = new AdminStellarKeyModel();

src/queue/accountMergeWorker.ts

@@ -307,7 +307,14 @@ accountMergeWorker.on("failed", (job, error) => {
   );
 
   if (job) {
-    capturePersistentFailure(job).catch((err) =>
+    capturePersistentFailure({
+      originalJobId: job.id,
+      queueName: ACCOUNT_MERGE_QUEUE_NAME,
+      jobName: "accountMerge",
+      jobData: job.data as unknown as Record<string, unknown>,
+      failureReason: error?.message ?? "Unknown error",
+      attemptsMade: job.attemptsMade,
+    }).catch((err) =>
       console.error("[DLQ] Error capturing failure:", err),
     );
   }

src/queue/batchPayoutWorker.ts

@@ -268,8 +268,8 @@ async function processBatch(provider: string): Promise<void> {
   const durationMs = Date.now() - startTime;
 
   // Record metrics
-  const successCount = result.results.filter(r => r.success).length;
-  const failureCount = result.results.filter(r => !r.success).length;
+const successCount = result.results.filter((r: { success: boolean }) => r.success).length;
+   const failureCount = result.results.filter((r: { success: boolean }) => !r.success).length;
 
   batchPayoutTotal.inc({ provider, status: result.success ? "success" : "partial" });
   batchPayoutItemsTotal.inc({ provider, status: "success" }, successCount);