didi-lot2-backend/backend/services/orchestration-layer/agent-v3/src/queue/connection.ts

/**
 * RabbitMQ Connection Manager
 *
 * Lazy initialization with graceful degradation.
 * When RabbitMQ is unavailable, async operations fall back to sync.
 */

import amqp from 'amqplib';
import { getRabbitMQUrl, getRabbitMQConfig, EXCHANGE_NAME } from '../shared/queue/constants';
import { log } from '../shared/logger';

// Use any to avoid amqplib type issues
let connection: any = null;
let channel: any = null;
let confirmChannel: any = null;
let isConnecting = false;
let lastConnectionAttempt = 0;
// 2s between retries (was 5s). Tuned for HA clusters where HAProxy may need
// a few seconds to reroute after failover; shorter backoff = faster recovery.
const CONNECTION_RETRY_DELAY = 2000;

/**
 * Get or create RabbitMQ connection
 */
export async function getConnection(): Promise<any> {
  if (connection) {
    return connection;
  }

  // Prevent concurrent connection attempts
  if (isConnecting) {
    await new Promise((resolve) => setTimeout(resolve, 100));
    return connection;
  }

  // Rate limit connection attempts
  const now = Date.now();
  if (now - lastConnectionAttempt < CONNECTION_RETRY_DELAY) {
    return null;
  }

  isConnecting = true;
  lastConnectionAttempt = now;

  try {
    const url = getRabbitMQUrl();
    const cfg = getRabbitMQConfig();
    const safeTarget = `${cfg.host}:${cfg.port}${cfg.vhost} (user: ${cfg.user})`;
    log.info(`[RabbitMQ] Connecting to ${safeTarget}...`);

    connection = await amqp.connect(url);

    connection.on('error', (err: Error) => {
      log.error(`[RabbitMQ] Connection error to ${cfg.host}${cfg.vhost}:`, err.message);
      connection = null;
      channel = null;
      confirmChannel = null;
    });

    connection.on('close', () => {
      log.warn(`[RabbitMQ] Connection closed to ${cfg.host}${cfg.vhost}`);
      connection = null;
      channel = null;
      confirmChannel = null;
    });

    connection.on('blocked', (reason: string) => {
      log.warn(`[RabbitMQ] Connection blocked by broker: ${reason}`);
    });
    connection.on('unblocked', () => {
      log.info('[RabbitMQ] Connection unblocked');
    });

    log.info(`[RabbitMQ] ✓ Connected to ${safeTarget}`);
    return connection;
  } catch (err) {
    log.warn('[RabbitMQ] Connection failed:', (err as Error).message);
    return null;
  } finally {
    isConnecting = false;
  }
}

/**
 * Get or create channel for consuming messages
 */
export async function getChannel(): Promise<any> {
  if (channel) {
    return channel;
  }

  const conn = await getConnection();
  if (!conn) {
    return null;
  }

  try {
    channel = await conn.createChannel();

    // Ensure exchange exists
    await channel.assertExchange(EXCHANGE_NAME, 'topic', { durable: true });

    channel.on('error', (err: Error) => {
      log.error('[RabbitMQ] Channel error:', err.message);
      channel = null;
    });

    channel.on('close', () => {
      log.warn('[RabbitMQ] Channel closed');
      channel = null;
    });

    return channel;
  } catch (err) {
    log.error('[RabbitMQ] Failed to create channel:', (err as Error).message);
    return null;
  }
}

/**
 * Get or create confirm channel for publishing with confirms
 */
export async function getConfirmChannel(): Promise<any> {
  if (confirmChannel) {
    return confirmChannel;
  }

  const conn = await getConnection();
  if (!conn) {
    return null;
  }

  try {
    confirmChannel = await conn.createConfirmChannel();

    // Ensure exchange exists
    await confirmChannel.assertExchange(EXCHANGE_NAME, 'topic', { durable: true });

    confirmChannel.on('error', (err: Error) => {
      log.error('[RabbitMQ] Confirm channel error:', err.message);
      confirmChannel = null;
    });

    confirmChannel.on('close', () => {
      log.warn('[RabbitMQ] Confirm channel closed');
      confirmChannel = null;
    });

    return confirmChannel;
  } catch (err) {
    log.error('[RabbitMQ] Failed to create confirm channel:', (err as Error).message);
    return null;
  }
}

/**
 * Publish on a confirm channel and WAIT for the broker ack.
 *
 * Without this, publishes on a confirm channel are fire-and-forget: the broker
 * can nack (queue overflow, internal error) or the connection can drop before
 * delivery, and the message is silently lost while the caller reports success.
 * Throws on nack/error so callers can fall back (sync path, error result).
 */
export function publishConfirmed(
  ch: any,
  exchange: string,
  routingKey: string,
  content: Buffer,
  options: Record<string, unknown>,
): Promise<void> {
  return new Promise<void>((resolve, reject) => {
    ch.publish(exchange, routingKey, content, options, (err: unknown) => {
      if (err) reject(err instanceof Error ? err : new Error(String(err)));
      else resolve();
    });
  });
}

/**
 * Check if RabbitMQ is available
 */
export async function isRabbitMQAvailable(): Promise<boolean> {
  const conn = await getConnection();
  return conn !== null;
}

/**
 * Close all connections gracefully
 */
export async function closeConnection(): Promise<void> {
  try {
    if (channel) {
      await channel.close();
      channel = null;
    }
    if (confirmChannel) {
      await confirmChannel.close();
      confirmChannel = null;
    }
    if (connection) {
      await connection.close();
      connection = null;
    }
    log.info('[RabbitMQ] Connections closed');
  } catch (err) {
    log.error('[RabbitMQ] Error closing connections:', (err as Error).message);
  }
}

/**
 * Health check
 */
export async function healthCheck(): Promise<{
  connected: boolean;
  host: string;
  error?: string;
}> {
  const url = getRabbitMQUrl();
  const host = url.replace(/amqp:\/\/[^@]+@/, '').replace(/\/.*$/, '');

  try {
    const conn = await getConnection();
    return {
      connected: conn !== null,
      host,
    };
  } catch (err) {
    return {
      connected: false,
      host,
      error: (err as Error).message,
    };
  }
}