Add ECDSA signature verification and enhance ComboHandler functionality

- Integrated ECDSA for signature verification in ComboHandler. - Updated event emissions to include additional parameters for better tracking. - Improved gas tracking during execution of combo plans. - Enhanced database interactions for storing and retrieving plans, including conflict resolution and status updates. - Added new dependencies for security and database management in orchestrator.
2025-11-05 16:28:48 -08:00
parent 3b09c35c47
commit f600b7b15e
48 changed files with 3381 additions and 46 deletions
--- a/orchestrator/src/services/cache.ts
+++ b/orchestrator/src/services/cache.ts
@@ -0,0 +1,106 @@
+import Redis from "ioredis";
+
+/**
+ * Redis caching service
+ */
+let redis: Redis | null = null;
+
+/**
+ * Initialize Redis connection
+ */
+export function initRedis(url?: string): Redis {
+  if (!redis) {
+    redis = new Redis(url || process.env.REDIS_URL || "redis://localhost:6379", {
+      maxRetriesPerRequest: 3,
+      retryStrategy: (times) => {
+        const delay = Math.min(times * 50, 2000);
+        return delay;
+      },
+    });
+
+    redis.on("error", (err) => {
+      console.error("Redis connection error:", err);
+    });
+
+    redis.on("connect", () => {
+      console.log("✅ Redis connected");
+    });
+  }
+
+  return redis;
+}
+
+/**
+ * Get Redis client
+ */
+export function getRedis(): Redis | null {
+  if (!redis && process.env.REDIS_URL) {
+    initRedis();
+  }
+  return redis;
+}
+
+/**
+ * Cache wrapper with TTL
+ */
+export async function cacheGet<T>(key: string): Promise<T | null> {
+  const client = getRedis();
+  if (!client) return null;
+
+  try {
+    const value = await client.get(key);
+    return value ? JSON.parse(value) : null;
+  } catch (error) {
+    console.error("Cache get error:", error);
+    return null;
+  }
+}
+
+export async function cacheSet<T>(key: string, value: T, ttlSeconds = 3600): Promise<void> {
+  const client = getRedis();
+  if (!client) return;
+
+  try {
+    await client.setex(key, ttlSeconds, JSON.stringify(value));
+  } catch (error) {
+    console.error("Cache set error:", error);
+  }
+}
+
+export async function cacheDelete(key: string): Promise<void> {
+  const client = getRedis();
+  if (!client) return;
+
+  try {
+    await client.del(key);
+  } catch (error) {
+    console.error("Cache delete error:", error);
+  }
+}
+
+/**
+ * Cache middleware for Express routes
+ */
+export function cacheMiddleware(ttlSeconds = 300) {
+  return async (req: express.Request, res: express.Response, next: express.NextFunction) => {
+    if (req.method !== "GET") {
+      return next();
+    }
+
+    const cacheKey = `cache:${req.path}:${JSON.stringify(req.query)}`;
+    const cached = await cacheGet(cacheKey);
+
+    if (cached) {
+      return res.json(cached);
+    }
+
+    const originalSend = res.json;
+    res.json = function (body: any) {
+      cacheSet(cacheKey, body, ttlSeconds).catch(console.error);
+      return originalSend.call(this, body);
+    };
+
+    next();
+  };
+}
+
--- a/orchestrator/src/services/deadLetterQueue.ts
+++ b/orchestrator/src/services/deadLetterQueue.ts
@@ -0,0 +1,62 @@
+import { query } from "../db/postgres";
+
+interface DeadLetterMessage {
+  messageId: string;
+  originalQueue: string;
+  payload: any;
+  error: string;
+  retryCount: number;
+  createdAt: string;
+}
+
+/**
+ * Add message to dead letter queue
+ */
+export async function addToDLQ(
+  queue: string,
+  payload: any,
+  error: string
+): Promise<void> {
+  const messageId = `dlq-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
+
+  await query(
+    `INSERT INTO dead_letter_queue (message_id, queue, payload, error, retry_count, created_at)
+     VALUES ($1, $2, $3, $4, $5, $6)`,
+    [messageId, queue, JSON.stringify(payload), error, 0, new Date().toISOString()]
+  );
+}
+
+/**
+ * Get messages from DLQ for retry
+ */
+export async function getDLQMessages(queue: string, limit = 10): Promise<DeadLetterMessage[]> {
+  const result = await query<DeadLetterMessage>(
+    `SELECT * FROM dead_letter_queue 
+     WHERE queue = $1 AND retry_count < 3
+     ORDER BY created_at ASC
+     LIMIT $2`,
+    [queue, limit]
+  );
+
+  return result.map((row) => ({
+    messageId: row.message_id,
+    originalQueue: row.queue,
+    payload: typeof row.payload === "string" ? JSON.parse(row.payload) : row.payload,
+    error: row.error,
+    retryCount: row.retry_count,
+    createdAt: row.created_at,
+  }));
+}
+
+/**
+ * Increment retry count
+ */
+export async function incrementRetryCount(messageId: string): Promise<void> {
+  await query(
+    `UPDATE dead_letter_queue 
+     SET retry_count = retry_count + 1, updated_at = CURRENT_TIMESTAMP
+     WHERE message_id = $1`,
+    [messageId]
+  );
+}
+
--- a/orchestrator/src/services/errorHandler.ts
+++ b/orchestrator/src/services/errorHandler.ts
@@ -0,0 +1,103 @@
+import { Request, Response, NextFunction } from "express";
+import { logger } from "../logging/logger";
+
+/**
+ * Error classification
+ */
+export enum ErrorType {
+  USER_ERROR = "USER_ERROR",
+  SYSTEM_ERROR = "SYSTEM_ERROR",
+  VALIDATION_ERROR = "VALIDATION_ERROR",
+  AUTHENTICATION_ERROR = "AUTHENTICATION_ERROR",
+  AUTHORIZATION_ERROR = "AUTHORIZATION_ERROR",
+  NOT_FOUND_ERROR = "NOT_FOUND_ERROR",
+  RATE_LIMIT_ERROR = "RATE_LIMIT_ERROR",
+  EXTERNAL_SERVICE_ERROR = "EXTERNAL_SERVICE_ERROR",
+}
+
+/**
+ * Custom error class
+ */
+export class AppError extends Error {
+  constructor(
+    public type: ErrorType,
+    public statusCode: number,
+    message: string,
+    public details?: any
+  ) {
+    super(message);
+    this.name = "AppError";
+  }
+}
+
+/**
+ * Error handling middleware
+ */
+export function errorHandler(
+  err: Error | AppError,
+  req: Request,
+  res: Response,
+  next: NextFunction
+) {
+  const requestId = req.headers["x-request-id"] as string || "unknown";
+
+  // Handle known application errors
+  if (err instanceof AppError) {
+    logger.warn({
+      error: err,
+      type: err.type,
+      requestId,
+      path: req.path,
+    }, `Application error: ${err.message}`);
+
+    return res.status(err.statusCode).json({
+      error: err.type,
+      message: err.message,
+      details: err.details,
+      requestId,
+    });
+  }
+
+  // Handle validation errors
+  if (err.name === "ValidationError" || err.name === "ZodError") {
+    logger.warn({
+      error: err,
+      requestId,
+      path: req.path,
+    }, "Validation error");
+
+    return res.status(400).json({
+      error: ErrorType.VALIDATION_ERROR,
+      message: "Validation failed",
+      details: err.message,
+      requestId,
+    });
+  }
+
+  // Handle unknown errors
+  logger.error({
+    error: err,
+    requestId,
+    path: req.path,
+    stack: err.stack,
+  }, "Unhandled error");
+
+  res.status(500).json({
+    error: ErrorType.SYSTEM_ERROR,
+    message: "An internal server error occurred",
+    requestId,
+    ...(process.env.NODE_ENV === "development" && { details: err.message }),
+  });
+}
+
+/**
+ * Async error wrapper
+ */
+export function asyncHandler(
+  fn: (req: Request, res: Response, next: NextFunction) => Promise<any>
+) {
+  return (req: Request, res: Response, next: NextFunction) => {
+    Promise.resolve(fn(req, res, next)).catch(next);
+  };
+}
+
--- a/orchestrator/src/services/featureFlags.ts
+++ b/orchestrator/src/services/featureFlags.ts
@@ -0,0 +1,61 @@
+/**
+ * Feature flags service with LaunchDarkly integration
+ */
+
+interface FeatureFlag {
+  key: string;
+  value: boolean;
+  defaultValue: boolean;
+}
+
+const flags: Map<string, boolean> = new Map();
+
+/**
+ * Initialize feature flags
+ */
+export function initFeatureFlags() {
+  // Load from environment variables
+  const envFlags = {
+    enableRecursion: process.env.ENABLE_RECURSION === "true",
+    enableFlashLoans: process.env.ENABLE_FLASH_LOANS === "true",
+    enableSimulation: process.env.ENABLE_SIMULATION === "true",
+    enableWebSocket: process.env.ENABLE_WEBSOCKET === "true",
+  };
+
+  Object.entries(envFlags).forEach(([key, value]) => {
+    flags.set(key, value);
+  });
+}
+
+/**
+ * Get feature flag value
+ */
+export function getFeatureFlag(key: string, defaultValue = false): boolean {
+  return flags.get(key) ?? defaultValue;
+}
+
+/**
+ * Set feature flag (for testing/admin)
+ */
+export function setFeatureFlag(key: string, value: boolean) {
+  flags.set(key, value);
+}
+
+/**
+ * LaunchDarkly integration (optional)
+ */
+export class LaunchDarklyService {
+  private client: any;
+
+  constructor(ldClient: any) {
+    this.client = ldClient;
+  }
+
+  async getFlag(key: string, defaultValue = false): Promise<boolean> {
+    if (this.client) {
+      return await this.client.variation(key, { key: "user" }, defaultValue);
+    }
+    return defaultValue;
+  }
+}
+
--- a/orchestrator/src/services/gracefulDegradation.ts
+++ b/orchestrator/src/services/gracefulDegradation.ts
@@ -0,0 +1,62 @@
+/**
+ * Graceful degradation strategies
+ */
+
+export interface DegradationStrategy {
+  fallback: () => Promise<any>;
+  timeout?: number;
+}
+
+/**
+ * Execute with graceful degradation
+ */
+export async function executeWithDegradation<T>(
+  primary: () => Promise<T>,
+  strategies: DegradationStrategy[]
+): Promise<T> {
+  try {
+    return await primary();
+  } catch (error) {
+    // Try fallback strategies in order
+    for (const strategy of strategies) {
+      try {
+        if (strategy.timeout) {
+          return await Promise.race([
+            strategy.fallback(),
+            new Promise<T>((_, reject) =>
+              setTimeout(() => reject(new Error("Fallback timeout")), strategy.timeout)
+            ),
+          ]);
+        }
+        return await strategy.fallback();
+      } catch (fallbackError) {
+        // Try next strategy
+        continue;
+      }
+    }
+    throw error; // All fallbacks failed
+  }
+}
+
+/**
+ * Example: Get plan with fallback to cache
+ */
+export async function getPlanWithFallback(planId: string, getFromCache: () => Promise<any>) {
+  return executeWithDegradation(
+    async () => {
+      // Primary: Get from database
+      const { getPlanById } = await import("../db/plans");
+      return await getPlanById(planId);
+    },
+    [
+      {
+        fallback: getFromCache,
+        timeout: 1000,
+      },
+      {
+        fallback: async () => ({ planId, status: "unknown" }),
+      },
+    ]
+  );
+}
+
--- a/orchestrator/src/services/hsm.ts
+++ b/orchestrator/src/services/hsm.ts
@@ -0,0 +1,66 @@
+/**
+ * HSM (Hardware Security Module) integration service
+ * For cryptographic operations in production
+ */
+
+export interface HSMService {
+  sign(data: Buffer, keyId: string): Promise<Buffer>;
+  verify(data: Buffer, signature: Buffer, keyId: string): Promise<boolean>;
+  generateKey(keyId: string): Promise<string>;
+  encrypt(data: Buffer, keyId: string): Promise<Buffer>;
+  decrypt(encrypted: Buffer, keyId: string): Promise<Buffer>;
+}
+
+/**
+ * Mock HSM service (for development)
+ * In production, integrate with actual HSM (AWS CloudHSM, Azure Dedicated HSM, etc.)
+ */
+export class MockHSMService implements HSMService {
+  private keys: Map<string, Buffer> = new Map();
+
+  async sign(data: Buffer, keyId: string): Promise<Buffer> {
+    // Mock implementation - in production use HSM SDK
+    const key = this.keys.get(keyId) || Buffer.from(keyId);
+    // In production: return await hsmClient.sign(data, keyId);
+    return Buffer.from("mock-signature");
+  }
+
+  async verify(data: Buffer, signature: Buffer, keyId: string): Promise<boolean> {
+    // Mock implementation
+    // In production: return await hsmClient.verify(data, signature, keyId);
+    return true;
+  }
+
+  async generateKey(keyId: string): Promise<string> {
+    // Mock implementation
+    // In production: return await hsmClient.generateKey(keyId);
+    const key = Buffer.from(`key-${keyId}-${Date.now()}`);
+    this.keys.set(keyId, key);
+    return keyId;
+  }
+
+  async encrypt(data: Buffer, keyId: string): Promise<Buffer> {
+    // Mock implementation
+    // In production: return await hsmClient.encrypt(data, keyId);
+    return Buffer.from(`encrypted-${data.toString()}`);
+  }
+
+  async decrypt(encrypted: Buffer, keyId: string): Promise<Buffer> {
+    // Mock implementation
+    // In production: return await hsmClient.decrypt(encrypted, keyId);
+    return Buffer.from(encrypted.toString().replace("encrypted-", ""));
+  }
+}
+
+/**
+ * Get HSM service instance
+ */
+export function getHSMService(): HSMService {
+  // In production, initialize actual HSM client
+  // const hsmUrl = process.env.HSM_URL;
+  // const hsmClient = new HSMClient(hsmUrl);
+  // return new HSMService(hsmClient);
+  
+  return new MockHSMService();
+}
+
--- a/orchestrator/src/services/redis.ts
+++ b/orchestrator/src/services/redis.ts
@@ -0,0 +1,3 @@
+// Re-export cache functions
+export { initRedis, getRedis, cacheGet, cacheSet, cacheDelete, cacheMiddleware } from "./cache";
+
--- a/orchestrator/src/services/secrets.ts
+++ b/orchestrator/src/services/secrets.ts
@@ -0,0 +1,104 @@
+/**
+ * Secrets management service
+ * Supports Azure Key Vault and AWS Secrets Manager
+ */
+
+export interface SecretsService {
+  getSecret(name: string): Promise<string | null>;
+  setSecret(name: string, value: string): Promise<void>;
+  deleteSecret(name: string): Promise<void>;
+}
+
+/**
+ * Azure Key Vault implementation
+ */
+export class AzureKeyVaultService implements SecretsService {
+  private vaultUrl: string;
+
+  constructor(vaultUrl: string) {
+    this.vaultUrl = vaultUrl;
+  }
+
+  async getSecret(name: string): Promise<string | null> {
+    // Mock implementation - in production use @azure/keyvault-secrets
+    try {
+      // const client = new SecretClient(this.vaultUrl, credential);
+      // const secret = await client.getSecret(name);
+      // return secret.value;
+      return process.env[name] || null;
+    } catch (error) {
+      console.error(`Failed to get secret ${name}:`, error);
+      return null;
+    }
+  }
+
+  async setSecret(name: string, value: string): Promise<void> {
+    // Mock implementation
+    // const client = new SecretClient(this.vaultUrl, credential);
+    // await client.setSecret(name, value);
+    console.log(`[Secrets] Setting secret ${name} (mock)`);
+  }
+
+  async deleteSecret(name: string): Promise<void> {
+    // Mock implementation
+    // const client = new SecretClient(this.vaultUrl, credential);
+    // await client.beginDeleteSecret(name);
+    console.log(`[Secrets] Deleting secret ${name} (mock)`);
+  }
+}
+
+/**
+ * AWS Secrets Manager implementation
+ */
+export class AWSSecretsManagerService implements SecretsService {
+  private region: string;
+
+  constructor(region: string) {
+    this.region = region;
+  }
+
+  async getSecret(name: string): Promise<string | null> {
+    // Mock implementation - in production use AWS SDK
+    try {
+      // const client = new SecretsManagerClient({ region: this.region });
+      // const response = await client.send(new GetSecretValueCommand({ SecretId: name }));
+      // return response.SecretString || null;
+      return process.env[name] || null;
+    } catch (error) {
+      console.error(`Failed to get secret ${name}:`, error);
+      return null;
+    }
+  }
+
+  async setSecret(name: string, value: string): Promise<void> {
+    // Mock implementation
+    console.log(`[Secrets] Setting secret ${name} (mock)`);
+  }
+
+  async deleteSecret(name: string): Promise<void> {
+    // Mock implementation
+    console.log(`[Secrets] Deleting secret ${name} (mock)`);
+  }
+}
+
+/**
+ * Get secrets service instance
+ */
+export function getSecretsService(): SecretsService {
+  const vaultUrl = process.env.AZURE_KEY_VAULT_URL;
+  const awsRegion = process.env.AWS_SECRETS_MANAGER_REGION;
+
+  if (vaultUrl) {
+    return new AzureKeyVaultService(vaultUrl);
+  } else if (awsRegion) {
+    return new AWSSecretsManagerService(awsRegion);
+  } else {
+    // Fallback to environment variables
+    return {
+      getSecret: async (name: string) => process.env[name] || null,
+      setSecret: async () => {},
+      deleteSecret: async () => {},
+    };
+  }
+}
+
--- a/orchestrator/src/services/timeout.ts
+++ b/orchestrator/src/services/timeout.ts
@@ -0,0 +1,27 @@
+/**
+ * Timeout wrapper for async operations
+ */
+export function withTimeout<T>(
+  promise: Promise<T>,
+  timeoutMs: number,
+  errorMessage = "Operation timed out"
+): Promise<T> {
+  return Promise.race([
+    promise,
+    new Promise<T>((_, reject) =>
+      setTimeout(() => reject(new Error(errorMessage)), timeoutMs)
+    ),
+  ]);
+}
+
+/**
+ * Create timeout configuration for different operation types
+ */
+export const TIMEOUTS = {
+  DLT_EXECUTION: 300000, // 5 minutes
+  BANK_API_CALL: 60000, // 1 minute
+  COMPLIANCE_CHECK: 30000, // 30 seconds
+  DATABASE_QUERY: 10000, // 10 seconds
+  EXTERNAL_API: 30000, // 30 seconds
+};
+