import cluster from 'node:cluster';

import postgres from 'postgres';

import { loadConfig } from './config.js';
import { ProtocolHandler } from './mcp/protocol-handler.js';
import { SessionManager } from './mcp/session-manager.js';
import { createLogger } from './observability/logger.js';
import { metrics } from './observability/metrics.js';
import { createMonitoredDb } from './observability/query-monitor.js';
import { CredentialCache } from './registry/credential-cache.js';
import { ServerRegistry } from './registry/server-registry.js';
import { ToolLoader } from './registry/tool-loader.js';
import { CircuitBreaker } from './resilience/circuit-breaker.js';
import { ConnectionMonitor } from './resilience/connection-monitor.js';
import { createRedisClient, closeRedis } from './redis.js';
import { createApp } from './server.js';
import { FallbackPoller } from './sync/fallback-poller.js';
import { loadAllServers, fetchToolsForServer, fetchCredentialHeaders } from './sync/postgres-loader.js';
import type { DbClient } from './sync/postgres-loader.js';
import { RedisSubscriber } from './sync/redis-subscriber.js';
import { decrypt } from './vault/decrypt.js';
import { encrypt } from './vault/encrypt.js';
import { clearKeyCache } from './vault/derive-key.js';

export async function startWorker(): Promise<void> {
  const config = loadConfig();
  const logger = createLogger(config);

  logger.info('Starting MCP Runtime...');

  // Database connection (raw postgres.js for direct SQL)
  const needsSsl =
    config.databaseUrl.includes('sslmode=require') ||
    config.databaseUrl.includes('sslmode=verify-full') ||
    config.databaseUrl.includes('sslmode=verify-ca') ||
    process.env['DATABASE_SSL'] === 'true';
  const ssl = needsSsl
    ? { rejectUnauthorized: process.env['DATABASE_SSL_REJECT_UNAUTHORIZED'] !== 'false' }
    : false;

  const sql = postgres(config.databaseUrl, {
    max: config.databasePoolMax,
    idle_timeout: 20,
    connect_timeout: 10,
    ssl,
  });

  const rawDb: DbClient = {
    async query<T>(queryStr: string, params?: readonly unknown[]): Promise<{ readonly rows: readonly T[] }> {
      // All queries MUST use $1/$2 parameterization. The queryStr is always a
      // compile-time constant from postgres-loader.ts — never user input.
      // eslint-disable-next-line no-restricted-syntax -- sole entry point; queryStr is always a compile-time constant
      const result = await sql.unsafe(queryStr, (params ?? []) as never[]);
      return { rows: result as unknown as readonly T[] };
    },
  };

  const db = createMonitoredDb({ logger, db: rawDb });

  // Redis connections (separate for subscriber — ioredis requirement)
  const redis = createRedisClient({ url: config.redisUrl });
  const redisSub = createRedisClient({ url: config.redisUrl });

  // Vault functions
  const decryptFn = (ciphertext: string): string =>
    decrypt(ciphertext, config.vaultSecret, config.vaultSalt);
  const encryptFn = (plaintext: string): string =>
    encrypt(plaintext, config.vaultSecret, config.vaultSalt);

  // Registry tiers
  const registry = new ServerRegistry({ logger });
  const toolLoader = new ToolLoader({
    logger,
    fetchTools: (serverId) => fetchToolsForServer(db, serverId),
  });
  const credentialCache = new CredentialCache({
    logger,
    fetchHeaders: (serverId) =>
      fetchCredentialHeaders(db, serverId, decryptFn, { logger, encryptFn, redis }),
    ttlMs: config.credentialTtlMs,
  });

  // Resilience
  const circuitBreaker = new CircuitBreaker({
    config: {
      failureThreshold: config.circuitBreakerThreshold,
      cooldownMs: config.circuitBreakerCooldownMs,
      halfOpenMaxProbes: 2,
    },
    logger,
  });
  const connectionMonitor = new ConnectionMonitor(logger);

  // Session management
  const sessionManager = new SessionManager({
    logger,
    connectionMonitor,
    maxSessions: config.maxSseSessions,
    heartbeatIntervalMs: config.sseHeartbeatIntervalMs,
    idleTimeoutMs: config.sseIdleTimeoutMs,
  });

  const toolExecutorDeps = {
    logger,
    circuitBreaker,
    authInjector: { credentialCache },
    timeoutMs: config.upstreamTimeoutMs,
  };

  // Protocol handler
  const protocolHandler = new ProtocolHandler({
    logger,
    registry,
    toolLoader,
    sessionManager,
    toolExecutorDeps,
    redis,
    db,
  });

  // Ready state
  let isReady = false;

  // Create Express app
  const app = createApp({
    config,
    logger,
    sessionManager,
    protocolHandler,
    registry,
    redis,
    isReady: () => isReady,
    toolLoader,
    toolExecutorDeps,
    db,
  });

  // Start HTTP server
  const httpServer = app.listen(config.port, () => {
    logger.info({ port: config.port }, 'HTTP server listening');
  });

  // Tune socket timeouts for high-concurrency SSE workloads.
  // keepAliveTimeout must exceed any upstream load-balancer timeout (Fly.io proxy = 60s)
  // so the LB doesn't send requests to a connection the server already closed.
  httpServer.keepAliveTimeout = 65_000;
  httpServer.headersTimeout = 70_000; // must be > keepAliveTimeout
  httpServer.requestTimeout = 30_000; // only applies until request is fully received — active SSE streams are unaffected
  httpServer.maxConnections = Infinity; // unlimited — we enforce limits at the session layer

  // Load initial data from Postgres
  const pgLoaderDeps = { db, logger, registry };
  await loadAllServers(pgLoaderDeps);

  // Redis sync
  const redisSubscriber = new RedisSubscriber({
    redis: redisSub,
    logger,
    registry,
    toolLoader,
    credentialCache,
    sessionManager,
    pgLoaderDeps,
  });

  const fallbackPoller = new FallbackPoller({
    logger,
    pgLoaderDeps,
    intervalMs: config.fallbackPollIntervalMs,
  });

  try {
    await redisSub.connect();
    await redisSubscriber.subscribe();
    logger.info('Redis subscriber connected');
  } catch (err) {
    logger.warn({ err }, 'Redis connection failed, starting fallback poller');
    fallbackPoller.start();
  }

  // Start session manager timers
  sessionManager.start();

  isReady = true;
  logger.info('MCP Runtime is ready');

  // Graceful shutdown with double-signal guard
  let isShuttingDown = false;

  const shutdown = async (signal: string): Promise<void> => {
    if (isShuttingDown) return;
    isShuttingDown = true;

    logger.info({ signal }, 'Received shutdown signal');
    isReady = false;

    try {
      await new Promise<void>((resolve) => httpServer.close(() => resolve()));
      sessionManager.stop();
      fallbackPoller.stop();

      await sessionManager.drainAll(config.drainTimeoutMs);
      await redisSubscriber.disconnect();

      await closeRedis();

      // Stop cleanup timers and free cached data
      circuitBreaker.dispose();
      toolLoader.dispose();
      credentialCache.dispose();
      credentialCache.evictAll();
      await sql.end();
      clearKeyCache();
      logger.info('Graceful shutdown complete');
      process.exit(0);
    } catch (err) {
      logger.error({ err }, 'Error during shutdown');
      process.exit(1);
    }
  };

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

  // In cluster mode, listen for shutdown message from primary
  if (cluster.isWorker) {
    process.on('message', (msg) => {
      if (msg === 'shutdown') {
        shutdown('cluster-shutdown').catch(() => process.exit(1));
      }
      if (msg && typeof msg === 'object' && (msg as Record<string, unknown>).type === 'metrics:request') {
        process.send?.({
          type: 'metrics:gauges',
          gauges: { active_sse_connections: metrics.getGauge('active_sse_connections') },
        });
      }
    });
  }
}

// Direct execution guard — allows `node dist/index.js` for dev/test
const isDirectExecution = !cluster.isWorker && (process.argv[1]?.endsWith('index.js') || process.argv[1]?.endsWith('index.ts'));
if (isDirectExecution) {
  startWorker().catch((err) => {
    // eslint-disable-next-line no-console
    console.error('Fatal startup error:', err);
    process.exit(1);
  });
}
