ai/gateway/src/main.ts

import Fastify from 'fastify';
import websocket from '@fastify/websocket';
import cors from '@fastify/cors';
import Redis from 'ioredis';
import { readFileSync } from 'fs';
import { load as loadYaml } from 'js-yaml';
import { UserService } from './db/user-service.js';
import { Authenticator } from './auth/authenticator.js';
import { createBetterAuth } from './auth/better-auth-config.js';
import { AuthService } from './auth/auth-service.js';
import { AuthRoutes } from './routes/auth-routes.js';
import { WebSocketHandler } from './channels/websocket-handler.js';
import { TelegramHandler } from './channels/telegram-handler.js';
import { KubernetesClient } from './k8s/client.js';
import { ContainerManager } from './k8s/container-manager.js';
import { ZMQRelayClient } from './clients/zmq-relay-client.js';
import { IcebergClient } from './clients/iceberg-client.js';
import { OHLCService } from './services/ohlc-service.js';
import { SymbolIndexService } from './services/symbol-index-service.js';
import { SymbolRoutes } from './routes/symbol-routes.js';

// Catch unhandled promise rejections for better debugging
process.on('unhandledRejection', (reason: any, promise) => {
  console.error('=== UNHANDLED PROMISE REJECTION ===');
  console.error('Reason:', reason);
  console.error('Message:', reason?.message);
  console.error('Stack:', reason?.stack);
  console.error('Cause:', reason?.cause);
  console.error('Promise:', promise);
  console.error('===================================');
  process.exit(1);
});
import {
  SessionRegistry,
  EventSubscriber,
  EventRouter,
  DeliveryService,
} from './events/index.js';
import { QdrantClient } from './clients/qdrant-client.js';
import { EmbeddingService, RAGRetriever, DocumentLoader } from './harness/memory/index.js';
import { join } from 'path';
import { fileURLToPath } from 'url';
import { dirname } from 'path';

const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);

// Load configuration from YAML files
function loadConfig() {
  const configPath = process.env.CONFIG_PATH || '/config/config.yaml';
  const secretsPath = process.env.SECRETS_PATH || '/config/secrets.yaml';

  let configData: any = {};
  let secretsData: any = {};

  try {
    const configFile = readFileSync(configPath, 'utf8');
    configData = loadYaml(configFile) || {};
    console.log(`Loaded configuration from ${configPath}`);
  } catch (error: any) {
    console.warn(`Could not load config from ${configPath}: ${error.message}, using defaults`);
  }

  try {
    const secretsFile = readFileSync(secretsPath, 'utf8');
    secretsData = loadYaml(secretsFile) || {};
    console.log(`Loaded secrets from ${secretsPath}`);
  } catch (error: any) {
    console.warn(`Could not load secrets from ${secretsPath}: ${error.message}`);
  }

  return {
    port: configData.server?.port || parseInt(process.env.PORT || '3000'),
    host: configData.server?.host || process.env.HOST || '0.0.0.0',
    logLevel: configData.server?.log_level || process.env.LOG_LEVEL || 'info',
    corsOrigin: configData.server?.cors_origin || process.env.CORS_ORIGIN || '*',
    baseUrl: configData.server?.base_url || process.env.BASE_URL || 'http://localhost:3000',
    trustedOrigins: configData.server?.trusted_origins || [
      process.env.BASE_URL || 'http://localhost:3000',
      'http://localhost:5173',
      'http://localhost:8080',
    ],
    databaseUrl: configData.database?.url || process.env.DATABASE_URL || 'postgresql://localhost/dexorder',

    // Authentication configuration
    authSecret: secretsData.auth?.secret || process.env.AUTH_SECRET || 'change-me-in-production',

    // LLM provider API keys and model configuration
    providerConfig: {
      anthropicApiKey: secretsData.llm_providers?.anthropic_api_key || process.env.ANTHROPIC_API_KEY,
      openaiApiKey: secretsData.llm_providers?.openai_api_key || process.env.OPENAI_API_KEY,
      googleApiKey: secretsData.llm_providers?.google_api_key || process.env.GOOGLE_API_KEY,
      openrouterApiKey: secretsData.llm_providers?.openrouter_api_key || process.env.OPENROUTER_API_KEY,
      defaultModel: {
        provider: configData.defaults?.model_provider || 'anthropic',
        model: configData.defaults?.model || 'claude-sonnet-4-6',
      },
      licenseModels: {
        free: {
          default: configData.license_models?.free?.default || 'claude-haiku-4-5-20251001',
          cost_optimized: configData.license_models?.free?.cost_optimized || 'claude-haiku-4-5-20251001',
          complex: configData.license_models?.free?.complex || 'claude-haiku-4-5-20251001',
          allowed_models: configData.license_models?.free?.allowed_models || ['claude-haiku-4-5-20251001'],
        },
        pro: {
          default: configData.license_models?.pro?.default || 'claude-sonnet-4-6',
          cost_optimized: configData.license_models?.pro?.cost_optimized || 'claude-haiku-4-5-20251001',
          complex: configData.license_models?.pro?.complex || 'claude-sonnet-4-6',
          blocked_models: configData.license_models?.pro?.blocked_models || ['claude-opus-4-6'],
        },
        enterprise: {
          default: configData.license_models?.enterprise?.default || 'claude-sonnet-4-6',
          cost_optimized: configData.license_models?.enterprise?.cost_optimized || 'claude-haiku-4-5-20251001',
          complex: configData.license_models?.enterprise?.complex || 'claude-opus-4-6',
        },
      },
    },

    telegramBotToken: secretsData.telegram?.bot_token || process.env.TELEGRAM_BOT_TOKEN || '',

    // Email service configuration
    emailServiceKey: secretsData.email?.service_key || process.env.EMAIL_SERVICE_KEY,
    emailFromAddress: configData.email?.from_address || process.env.EMAIL_FROM_ADDRESS || 'noreply@dexorder.com',

    // Push notification service configuration
    pushServiceKey: secretsData.push?.service_key || process.env.PUSH_SERVICE_KEY,

    // Event router configuration
    eventRouterBind: configData.events?.router_bind || process.env.EVENT_ROUTER_BIND || 'tcp://*:5571',

    // Redis configuration (for harness memory layer)
    redisUrl: configData.redis?.url || process.env.REDIS_URL || 'redis://localhost:6379',

    // Qdrant configuration (for RAG)
    qdrant: {
      url: configData.qdrant?.url || process.env.QDRANT_URL || 'http://localhost:6333',
      apiKey: secretsData.qdrant?.api_key || process.env.QDRANT_API_KEY,
      collectionName: configData.qdrant?.collection || process.env.QDRANT_COLLECTION || 'gateway_memory',
    },

    // Iceberg configuration (for durable storage)
    iceberg: {
      catalogUri: configData.iceberg?.catalog_uri || process.env.ICEBERG_CATALOG_URI || 'http://iceberg-catalog:8181',
      namespace: configData.iceberg?.namespace || process.env.ICEBERG_NAMESPACE || 'gateway',
      ohlcCatalogUri: configData.iceberg?.ohlc_catalog_uri || process.env.ICEBERG_OHLC_CATALOG_URI,
      ohlcNamespace: configData.iceberg?.ohlc_namespace || process.env.ICEBERG_OHLC_NAMESPACE || 'trading',
      s3Endpoint: configData.iceberg?.s3_endpoint || process.env.S3_ENDPOINT,
      s3AccessKey: secretsData.iceberg?.s3_access_key || process.env.S3_ACCESS_KEY,
      s3SecretKey: secretsData.iceberg?.s3_secret_key || process.env.S3_SECRET_KEY,
    },

    // Relay configuration (for historical data)
    relay: {
      requestEndpoint: configData.relay?.request_endpoint || process.env.RELAY_REQUEST_ENDPOINT || 'tcp://relay:5559',
      notificationEndpoint: configData.relay?.notification_endpoint || process.env.RELAY_NOTIFICATION_ENDPOINT || 'tcp://relay:5558',
    },

    // Embedding configuration (for RAG)
    embedding: {
      provider: (configData.embedding?.provider || process.env.EMBEDDING_PROVIDER || 'ollama') as 'ollama' | 'openai' | 'anthropic' | 'local' | 'voyage' | 'cohere' | 'none',
      model: configData.embedding?.model || process.env.EMBEDDING_MODEL,
      apiKey: secretsData.embedding?.api_key || process.env.EMBEDDING_API_KEY || secretsData.llm_providers?.openai_api_key || process.env.OPENAI_API_KEY,
      ollamaUrl: configData.embedding?.ollama_url || process.env.OLLAMA_URL || 'http://localhost:11434',
    },

    // Kubernetes configuration
    kubernetes: {
      namespace: configData.kubernetes?.namespace || process.env.KUBERNETES_NAMESPACE || 'dexorder-agents',
      inCluster: configData.kubernetes?.in_cluster ?? (process.env.KUBERNETES_IN_CLUSTER === 'true'),
      context: configData.kubernetes?.context || process.env.KUBERNETES_CONTEXT,
      agentImage: configData.kubernetes?.agent_image || process.env.AGENT_IMAGE || 'ghcr.io/dexorder/agent:latest',
      sidecarImage: configData.kubernetes?.sidecar_image || process.env.SIDECAR_IMAGE || 'ghcr.io/dexorder/lifecycle-sidecar:latest',
      storageClass: configData.kubernetes?.storage_class || process.env.AGENT_STORAGE_CLASS || 'standard',
      imagePullPolicy: configData.kubernetes?.image_pull_policy || process.env.IMAGE_PULL_POLICY || 'Always',
    },
  };
}

const config = loadConfig();

const app = Fastify({
  logger: {
    level: config.logLevel,
    transport: {
      target: 'pino-pretty',
      options: {
        colorize: true,
        translateTime: 'HH:MM:ss Z',
        ignore: 'pid,hostname',
      },
    },
  },
});

// Validate at least one LLM provider is configured
const hasAnyProvider = Object.values(config.providerConfig).some(key => !!key);
if (!hasAnyProvider) {
  app.log.error('At least one LLM provider API key is required (ANTHROPIC_API_KEY, OPENAI_API_KEY, GOOGLE_API_KEY, or OPENROUTER_API_KEY)');
  process.exit(1);
}

// Register plugins
await app.register(cors, {
  origin: config.corsOrigin,
});

await app.register(websocket, {
  options: {
    maxPayload: 1024 * 1024, // 1MB
  },
});

// Initialize services
const userService = new UserService(config.databaseUrl);

// Initialize Better Auth
let betterAuth;
try {
  app.log.info({ databaseUrl: config.databaseUrl.replace(/:[^:@]+@/, ':***@') }, 'Initializing Better Auth');
  betterAuth = await createBetterAuth({
    databaseUrl: config.databaseUrl,
    secret: config.authSecret,
    baseUrl: config.baseUrl,
    trustedOrigins: config.trustedOrigins,
    logger: app.log,
  });
  app.log.info('Better Auth initialized successfully');
} catch (error: any) {
  app.log.error({ error, message: error.message, stack: error.stack }, 'Failed to initialize Better Auth');
  throw error;
}

// Initialize Auth Service
const authService = new AuthService({
  auth: betterAuth,
  pool: userService.getPool(),
  logger: app.log,
});

// Connect UserService with AuthService for JWT verification
userService.setAuthService(authService);

// Initialize Redis client (for harness memory layer)
const redis = new Redis(config.redisUrl, {
  maxRetriesPerRequest: 3,
  connectTimeout: 10000, // 10 seconds
  retryStrategy: (times) => {
    if (times > 5) {
      app.log.error('Redis connection failed after 5 retries');
      return null; // Stop retrying
    }
    const delay = Math.min(times * 50, 2000);
    return delay;
  },
  lazyConnect: true,
});

// Initialize Qdrant client (for RAG)
const qdrantClient = new QdrantClient(config.qdrant, app.log);

// Initialize Iceberg client (for durable storage)
// const icebergClient = new IcebergClient(config.iceberg, app.log);

// Initialize ZMQ Relay client (for historical data)
// Note: onMetadataUpdate callback will be set after symbolIndexService is initialized
const zmqRelayClient = new ZMQRelayClient({
  relayRequestEndpoint: config.relay.requestEndpoint,
  relayNotificationEndpoint: config.relay.notificationEndpoint,
}, app.log);

app.log.info({
  redis: config.redisUrl,
  qdrant: config.qdrant.url,
  iceberg: config.iceberg.catalogUri,
  relay: config.relay.requestEndpoint,
  embeddingProvider: config.embedding.provider,
}, 'Harness storage clients configured');

// Initialize Kubernetes client and container manager
const k8sClient = new KubernetesClient({
  namespace: config.kubernetes.namespace,
  inCluster: config.kubernetes.inCluster,
  context: config.kubernetes.context,
  logger: app.log,
});

const containerManager = new ContainerManager({
  k8sClient,
  agentImage: config.kubernetes.agentImage,
  sidecarImage: config.kubernetes.sidecarImage,
  storageClass: config.kubernetes.storageClass,
  imagePullPolicy: config.kubernetes.imagePullPolicy,
  namespace: config.kubernetes.namespace,
  logger: app.log,
});
app.log.debug('Container manager initialized');

const authenticator = new Authenticator({
  userService,
  containerManager,
  logger: app.log,
});
app.log.debug('Authenticator initialized');

// Initialize event system
const sessionRegistry = new SessionRegistry();
app.log.debug('Session registry initialized');

const deliveryService = new DeliveryService({
  telegramBotToken: config.telegramBotToken,
  emailServiceKey: config.emailServiceKey,
  emailFromAddress: config.emailFromAddress,
  pushServiceKey: config.pushServiceKey,
  logger: app.log,
});
app.log.debug('Delivery service initialized');

const eventSubscriber = new EventSubscriber(sessionRegistry, app.log);
app.log.debug('Event subscriber initialized');

const eventRouter = new EventRouter({
  sessions: sessionRegistry,
  delivery: deliveryService,
  logger: app.log,
  bindEndpoint: config.eventRouterBind,
});
app.log.debug('Event router initialized');

// Initialize OHLC service (optional - only if relay is available)
let ohlcService: OHLCService | undefined;
try {
  const icebergClient = new IcebergClient(config.iceberg, app.log);
  ohlcService = new OHLCService({
    icebergClient,
    relayClient: zmqRelayClient,
    logger: app.log,
  });
  app.log.info('OHLC service initialized');
} catch (error) {
  app.log.warn({ error }, 'Failed to initialize OHLC service - historical data will not be available');
}

// Initialize Symbol Index Service (deferred to after server starts)
let symbolIndexService: SymbolIndexService | undefined;

// Initialize channel handlers
const websocketHandler = new WebSocketHandler({
  authenticator,
  containerManager,
  providerConfig: config.providerConfig,
  sessionRegistry,
  eventSubscriber,
  ohlcService,  // Optional
  symbolIndexService,  // Optional
});
app.log.debug('WebSocket handler initialized');

const telegramHandler = new TelegramHandler({
  authenticator,
  providerConfig: config.providerConfig,
  telegramBotToken: config.telegramBotToken,
});
app.log.debug('Telegram handler initialized');

// Initialize auth routes
app.log.debug('Initializing auth routes...');
const authRoutes = new AuthRoutes({
  authService,
  betterAuth,
});

// Register routes
app.log.debug('Registering auth routes...');
try {
  authRoutes.register(app);
  app.log.debug('Auth routes registered successfully');
} catch (error: any) {
  app.log.error({ error, message: error.message, stack: error.stack }, 'Failed to register auth routes');
  throw error;
}

app.log.debug('Registering websocket handler...');
websocketHandler.register(app);
app.log.debug('Registering telegram handler...');
telegramHandler.register(app);

// Register symbol routes (service may not be ready yet, but routes will handle this)
app.log.debug('Registering symbol routes...');
const getSymbolService = () => symbolIndexService;
const symbolRoutes = new SymbolRoutes({ getSymbolIndexService: getSymbolService });
symbolRoutes.register(app);

app.log.debug('All routes registered');

// Health check
app.get('/health', async () => {
  const health: any = {
    status: 'ok',
    timestamp: new Date().toISOString(),
    activeSessions: sessionRegistry.size(),
    eventSubscriptions: eventSubscriber.getSubscriptionCount(),
    processedEvents: eventRouter.getProcessedEventCount(),
  };

  // Add RAG stats if available
  if (app.hasDecorator('ragRetriever')) {
    try {
      const ragStats = await (app as any).ragRetriever.getStats();
      health.rag = {
        vectorCount: ragStats.vectorCount,
        indexedCount: ragStats.indexedCount,
      };
    } catch (error) {
      // Ignore errors in health check
    }
  }

  return health;
});

// Admin endpoints
app.post('/admin/reload-knowledge', async (_request, reply) => {
  if (!app.hasDecorator('documentLoader')) {
    return reply.code(503).send({
      error: 'Document loader not initialized',
    });
  }

  try {
    app.log.info('Manual knowledge reload requested');
    const stats = await (app as any).documentLoader.loadAll();

    return {
      success: true,
      stats,
      timestamp: new Date().toISOString(),
    };
  } catch (error: any) {
    app.log.error({ error }, 'Failed to reload knowledge');
    return reply.code(500).send({
      error: 'Failed to reload knowledge',
      message: error.message,
    });
  }
});

app.get('/admin/knowledge-stats', async (_request, reply) => {
  if (!app.hasDecorator('documentLoader')) {
    return reply.code(503).send({
      error: 'Document loader not initialized',
    });
  }

  try {
    const loaderStats = (app as any).documentLoader.getStats();
    const ragStats = await (app as any).ragRetriever.getStats();

    return {
      loader: loaderStats,
      rag: {
        vectorCount: ragStats.vectorCount,
        indexedCount: ragStats.indexedCount,
        collectionSize: ragStats.collectionSize,
      },
      timestamp: new Date().toISOString(),
    };
  } catch (error: any) {
    app.log.error({ error }, 'Failed to get knowledge stats');
    return reply.code(500).send({
      error: 'Failed to get knowledge stats',
      message: error.message,
    });
  }
});

// Graceful shutdown
const shutdown = async () => {
  app.log.info('Shutting down gracefully...');
  try {
    // Stop event system first
    await eventSubscriber.stop();
    await eventRouter.stop();

    // Close ZMQ relay client
    if (zmqRelayClient.isConnected()) {
      await zmqRelayClient.close();
    }

    // Disconnect Redis
    redis.disconnect();

    await userService.close();
    await app.close();
    app.log.info('Shutdown complete');
    process.exit(0);
  } catch (error) {
    app.log.error({ error }, 'Error during shutdown');
    process.exit(1);
  }
};

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

// Start server
try {
  app.log.debug('Starting server initialization...');

  // Connect to Redis
  app.log.debug('Connecting to Redis...');
  await redis.connect();
  app.log.info('Redis connected');

  // Connect to ZMQ Relay
  app.log.debug('Connecting to ZMQ Relay...');
  try {
    await zmqRelayClient.connect();
    app.log.info('ZMQ Relay connected');
  } catch (error) {
    app.log.warn({ error }, 'ZMQ Relay connection failed - historical data will not be available');
  }

  // Initialize Qdrant collection
  app.log.debug('Initializing Qdrant...');
  try {
    await qdrantClient.initialize();
    app.log.info('Qdrant collection initialized');
  } catch (error) {
    app.log.warn({ error }, 'Qdrant initialization failed - RAG will not be available');
  }

  // Initialize RAG system and load global knowledge
  app.log.debug('Initializing RAG system...');
  try {
    // Initialize embedding service
    const embeddingService = new EmbeddingService(config.embedding, app.log);
    const vectorDimension = embeddingService.getDimensions();

    // Initialize RAG retriever
    const ragRetriever = new RAGRetriever(config.qdrant, app.log, vectorDimension);
    await ragRetriever.initialize();

    // Initialize document loader
    const knowledgeDir = join(__dirname, '..', 'knowledge');
    const documentLoader = new DocumentLoader(
      { knowledgeDir },
      embeddingService,
      ragRetriever,
      app.log
    );

    // Load all knowledge documents
    const loadStats = await documentLoader.loadAll();
    app.log.info(loadStats, 'Global knowledge loaded into RAG');

    // Store references for admin endpoints
    app.decorate('documentLoader', documentLoader);
    app.decorate('ragRetriever', ragRetriever);
  } catch (error) {
    app.log.warn({ error }, 'Failed to load global knowledge - RAG will use existing data');
  }

  // Start event system
  app.log.debug('Starting event subscriber...');
  await eventSubscriber.start();
  app.log.debug('Starting event router...');
  await eventRouter.start();
  app.log.debug('Event system started');

  app.log.debug('Starting Fastify server...');
  await app.listen({
    port: config.port,
    host: config.host,
  });

  app.log.info(
    {
      port: config.port,
      host: config.host,
      eventRouterBind: config.eventRouterBind,
      redis: config.redisUrl,
      qdrant: config.qdrant.url,
    },
    'Gateway server started'
  );

  // Initialize Symbol Index Service (after server is running)
  // This is done asynchronously to not block server startup
  (async () => {
    try {
      const icebergClient = new IcebergClient(config.iceberg, app.log);
      const indexService = new SymbolIndexService({
        icebergClient,
        logger: app.log,
      });
      await indexService.initialize();
      symbolIndexService = indexService;

      // Update websocket handler's config so it can use the service
      (websocketHandler as any).config.symbolIndexService = indexService;

      // Configure ZMQ relay to reload symbol metadata on updates
      (zmqRelayClient as any).config.onMetadataUpdate = async () => {
        app.log.info('Reloading symbol metadata from Iceberg');
        await indexService.initialize();
        app.log.info({ stats: indexService.getStats() }, 'Symbol metadata reloaded');
      };

      app.log.info({ stats: symbolIndexService.getStats() }, 'Symbol index service initialized');
    } catch (error) {
      app.log.warn({ error }, 'Failed to initialize symbol index service - symbol search will not be available');
    }
  })();
} catch (error) {
  app.log.error({ error }, 'Failed to start server');
  process.exit(1);
}