import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Queue } from 'bullmq';
import IORedis from 'ioredis';
import { Prisma } from '@prisma/client';
import { randomUUID } from 'crypto';
import { assertMoneyUnits, moneyToNumber } from '../common/money';
import { PrismaService } from '../prisma/prisma.service';
import type { CreateChannelDto, UpdateChannelDto, CreateChannelGroupDto, CreateChannelGroupItemDto, UpdateChannelGroupDto, CreateRouteRuleDto, CreateReportFieldDto, ReplaceReportFieldsDto, CreateReportMaterialDto, CreateReportTaskDto, ChangeReportTaskStatusesDto, CreateReportExportDto, CreateReceiptImportDto, UpsertConnectionStateDto, ChangeChannelStatusDto, CopyChannelDto, TestChannelDto } from './channels.contracts';
import { GATEWAY_CONNECTION_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_GATEWAY_CONTROL_URL, DEFAULT_CHANNEL_CONNECTION_ID, DEFAULT_CONNECTING_TIMEOUT_MS, DEFAULT_CONNECTING_TIMEOUT_SCAN_MS, DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS, DEFAULT_GATEWAY_RECONCILE_INTERVAL_MS, DEFAULT_GATEWAY_CONTROL_TIMEOUT_MS, DEFAULT_HEARTBEAT_INTERVAL_SECONDS, DEFAULT_HEARTBEAT_MISS_THRESHOLD, HEARTBEAT_AUDIT_INTERVAL_MS, CONNECTING_TIMEOUT_ERROR, DEFAULT_CMPP_VERSION, normalizeTestPhones, normalizeTestContent, calculateBillingUnits, buildChannelTestSubmitCommand, getConfigValue, getStringConfigValue, normalizeConnectionAction, normalizeCmppVersion, normalizeGatewayConnectionStatus, defaultChannelConnectionId, getDesiredConnections, ChannelConnectionSettings, getRuntimeConfigInteger, channelConnectionSettingsChanged, channelGroupAuditSnapshot, normalizeChannelRuntimeConfig, normalizeCmppServiceId, normalizeChannelRateLimit, normalizeExtensionDigits, getPositiveRuntimeInteger, bullmqConnection, getPositiveIntegerEnv, parseReceiptContent, splitReceiptLine, stripReceiptCell, findReceiptStatusIndex, normalizeReceiptStatus, deriveReceiptStatus, ChannelReportDeliveryRow, summarizeChannelReportDelivery, sumReportDelivery, percentage, latestDate, currentShanghaiDayRange, normalizeRetryTimeLimitMinutes, normalizeSpreadsheetSize, normalizeBusinessCarrier, normalizeChannelCarrier, isChannelCarrierCompatible, normalizeRegion, isRegionCompatible, validateGroupItems, normalizeReportType, summarizeReportStatuses, normalizeLinkEvent } from './channels.helpers';


/** R5 channel domain service composed behind ChannelsService. */
export class ChannelConnectionService {
  private readonly logger = new Logger(ChannelConnectionService.name);
  private gatewayConnectionQueue?: Queue;
  private gatewaySubmitQueue?: Queue;
  private redis?: IORedis;
  private connectionTimeoutTimer?: ReturnType<typeof setInterval>;
  private gatewayStartupReconnectTimer?: ReturnType<typeof setTimeout>;
  private gatewayReconcileTimer?: ReturnType<typeof setInterval>;
  constructor(private readonly prisma: PrismaService) {}

  onModuleInit() {
    if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED !== 'true') {
      this.connectionTimeoutTimer = setInterval(() => {
        void this.markTimedOutConnectingChannels().catch((error) => {
          this.logger.error(`Failed to mark timed-out CMPP connections: ${error instanceof Error ? error.message : String(error)}`);
        });
      }, getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_SCAN_MS', DEFAULT_CONNECTING_TIMEOUT_SCAN_MS));
      this.connectionTimeoutTimer.unref?.();
    }
    this.gatewayStartupReconnectTimer = setTimeout(() => {
      void this.reconnectActiveChannelsAfterGatewayRestart();
    }, getPositiveIntegerEnv('GATEWAY_STARTUP_RECONNECT_DELAY_MS', DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS));
    this.gatewayStartupReconnectTimer.unref?.();
    if (process.env.GATEWAY_CONNECTION_RECONCILER_DISABLED !== 'true') {
      this.gatewayReconcileTimer = setInterval(() => {
        void this.reconcileGatewayConnections().catch((error) => {
          this.logger.error(`Failed to reconcile supplier connections: ${error instanceof Error ? error.message : String(error)}`);
        });
      }, getPositiveIntegerEnv('GATEWAY_CONNECTION_RECONCILE_INTERVAL_MS', DEFAULT_GATEWAY_RECONCILE_INTERVAL_MS));
      this.gatewayReconcileTimer.unref?.();
    }
  }

  async onModuleDestroy() {
    if (this.connectionTimeoutTimer) {
      clearInterval(this.connectionTimeoutTimer);
    }
    if (this.gatewayStartupReconnectTimer) {
      clearTimeout(this.gatewayStartupReconnectTimer);
    }
    if (this.gatewayReconcileTimer) {
      clearInterval(this.gatewayReconcileTimer);
    }
    await this.gatewayConnectionQueue?.close();
    await this.gatewaySubmitQueue?.close();
    this.redis?.disconnect();
  }

  listChannelMetrics(channelId: string) {
    return this.prisma.channelHealthMetric.findMany({
      where: { channelId },
      orderBy: { windowStart: 'desc' },
      take: 100,
    });
  }

  listChannelConnections(channelId: string) {
    return this.prisma.cmppConnectionState.findMany({
      where: { channelId },
      orderBy: { updatedAt: 'desc' },
    });
  }

  async listChannelConnectionLogs(channelId: string) {
    const channel = await this.prisma.smsChannel.findUnique({ where: { id: channelId }, select: { id: true } });
    if (!channel) {
      throw new NotFoundException('Channel not found');
    }
    const [connectionStates, logs] = await Promise.all([
      this.prisma.cmppConnectionState.findMany({
        where: { channelId },
        orderBy: { updatedAt: 'desc' },
        take: 50,
      }),
      this.prisma.operationLog.findMany({
        where: {
          OR: [
            { resource: 'sms_channel', resourceId: channelId },
            { resource: 'cmpp_connection', resourceId: { startsWith: `${channelId}:` } },
          ],
        },
        orderBy: { createdAt: 'desc' },
        take: 100,
      }),
    ]);
    return {
      channelId,
      connectionStates,
      logs: logs.map((log) => ({
        id: log.id,
        time: log.createdAt,
        event: normalizeLinkEvent(log.action),
        action: log.action,
        resourceId: log.resourceId,
        detail: log.detail,
      })),
    };
  }

  listTenantConnections(tenantId: string) {
    return this.prisma.cmppConnectionState.findMany({
      where: { tenantId },
      include: { channel: true },
      orderBy: { updatedAt: 'desc' },
    });
  }

  async upsertConnectionState(data: UpsertConnectionStateDto) {
    const rawStatus = data.status;
    const status = normalizeGatewayConnectionStatus(rawStatus);
    if (data.applicationId) {
      const application = await this.prisma.smsApplication.findUnique({ where: { id: data.applicationId }, select: { tenantId: true } });
      if (!application) {
        throw new BadRequestException('applicationId does not reference an existing application');
      }
      if (data.tenantId && data.tenantId !== application.tenantId) {
        throw new BadRequestException('applicationId does not belong to tenantId');
      }
      data.tenantId = application.tenantId;
    }
    const payload = {
      tenantId: data.tenantId,
      applicationId: data.applicationId,
      status,
      desiredConnections: data.desiredConnections ?? 1,
      currentConnections: data.currentConnections ?? (status === 'connected' ? 1 : 0),
      lastConnectedAt: data.lastConnectedAt ? new Date(data.lastConnectedAt) : undefined,
      lastDisconnectedAt: data.lastDisconnectedAt ? new Date(data.lastDisconnectedAt) : undefined,
      lastHeartbeatAt: data.lastHeartbeatAt ? new Date(data.lastHeartbeatAt) : undefined,
      reconnectCount: data.reconnectCount ?? 0,
      lastReconnectAttemptAt: data.lastReconnectAttemptAt ? new Date(data.lastReconnectAttemptAt) : undefined,
      nextReconnectAt: data.nextReconnectAt ? new Date(data.nextReconnectAt) : status === 'connected' ? null : undefined,
      lastErrorCategory: status === 'connected' ? null : data.lastErrorCategory,
      lastError: status === 'connected' ? null : data.lastError,
    };
    const existing = await this.prisma.cmppConnectionState.findFirst({
      where: {
        applicationId: data.applicationId ?? null,
        channelId: data.channelId,
        connectionId: data.connectionId,
      },
    });
    let state;
    if (existing) {
      state = await this.prisma.cmppConnectionState.update({ where: { id: existing.id }, data: payload });
    } else {
      try {
        state = await this.prisma.cmppConnectionState.create({
          data: {
            channelId: data.channelId,
            connectionId: data.connectionId,
            ...payload,
          },
        });
      } catch (error) {
        if ((error as { code?: string }).code !== 'P2002') {
          throw error;
        }
        const concurrent = await this.prisma.cmppConnectionState.findFirst({
          where: {
            applicationId: data.applicationId ?? null,
            channelId: data.channelId,
            connectionId: data.connectionId,
          },
        });
        if (!concurrent) {
          throw error;
        }
        state = await this.prisma.cmppConnectionState.update({ where: { id: concurrent.id }, data: payload });
      }
    }
    const action = normalizeConnectionAction(
      ['heartbeat', 'active_test'].includes(rawStatus.toLowerCase()) ? rawStatus : status,
    );
    const heartbeatObservedAt = data.lastHeartbeatAt ? new Date(data.lastHeartbeatAt) : new Date();
    const shouldWriteAudit = action !== 'heartbeat'
      || !existing?.lastHeartbeatAt
      || heartbeatObservedAt.getTime() - existing.lastHeartbeatAt.getTime() >= HEARTBEAT_AUDIT_INTERVAL_MS;
    if (shouldWriteAudit) {
      await this.prisma.operationLog.create({
        data: {
          tenantId: data.tenantId,
          action: `cmpp_connection.${action}`,
          resource: 'cmpp_connection',
          resourceId: `${data.channelId}:${data.connectionId}`,
          detail: {
            status,
            applicationId: state.applicationId,
            desiredConnections: state.desiredConnections,
            currentConnections: state.currentConnections,
            lastError: state.lastError,
          } as Prisma.InputJsonValue,
        },
      });
    }
    return state;
  }

  async markTimedOutConnectingChannels(now = new Date()) {
    const timeoutMs = getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_MS', DEFAULT_CONNECTING_TIMEOUT_MS);
    const cutoff = new Date(now.getTime() - timeoutMs);
    const lastError = `${CONNECTING_TIMEOUT_ERROR} after ${Math.round(timeoutMs / 1000)} seconds`;
    const states = await this.prisma.cmppConnectionState.findMany({
      where: {
        status: 'connecting',
        updatedAt: { lte: cutoff },
      },
      select: {
        id: true,
        tenantId: true,
        applicationId: true,
        channelId: true,
        connectionId: true,
        desiredConnections: true,
        currentConnections: true,
        updatedAt: true,
      },
      take: 100,
    });
    let failed = 0;
    for (const state of states) {
      const result = await this.prisma.cmppConnectionState.updateMany({
        where: {
          id: state.id,
          status: 'connecting',
          updatedAt: { lte: cutoff },
        },
        data: {
          status: 'failed',
          currentConnections: 0,
          lastDisconnectedAt: now,
          nextReconnectAt: now,
          lastErrorCategory: 'timeout',
          lastError,
        },
      });
      if (result.count === 0) {
        continue;
      }
      failed += result.count;
      await this.prisma.operationLog.create({
        data: {
          tenantId: state.tenantId,
          action: 'cmpp_connection.failed',
          resource: 'cmpp_connection',
          resourceId: `${state.channelId}:${state.connectionId}`,
          detail: {
            reason: 'connect_timeout',
            applicationId: state.applicationId,
            status: 'failed',
            previousStatus: 'connecting',
            desiredConnections: state.desiredConnections,
            currentConnectionsBefore: state.currentConnections,
            currentConnections: 0,
            timeoutMs,
            lastError,
          } as Prisma.InputJsonValue,
        },
      });
    }
    return { checked: states.length, failed };
  }

  async requestChannelConnection(
    channel: {
      id: string;
      code: string;
      name: string;
      gatewayHost: string;
      gatewayPort: number;
      account: string;
      passwordCipher: string;
      srcId: string;
      cmppVersion: string;
      rateLimitPerSecond: number;
      config?: Prisma.JsonValue | null;
    },
    reason: 'channel_created' | 'channel_enabled' | 'channel_updated' | 'gateway_restarted' | 'automatic_reconnect',
    operatorId?: string,
  ) {
    const desiredConnections = getDesiredConnections(channel.config);
    const connectionId = defaultChannelConnectionId(channel.id);
    const existing = await this.prisma.cmppConnectionState.findFirst({
      where: {
        applicationId: null,
        channelId: channel.id,
        connectionId,
      },
    });
    const data = {
      applicationId: null,
      status: 'connecting',
      desiredConnections,
      currentConnections: 0,
      lastError: null,
      lastReconnectAttemptAt: new Date(),
      nextReconnectAt: new Date(Date.now() + getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_MS', DEFAULT_CONNECTING_TIMEOUT_MS)),
    };
    let state;
    if (existing) {
      state = await this.prisma.cmppConnectionState.update({ where: { id: existing.id }, data });
    } else {
      try {
        state = await this.prisma.cmppConnectionState.create({
          data: {
            channelId: channel.id,
            connectionId,
            ...data,
          },
        });
      } catch (error) {
        if ((error as { code?: string }).code !== 'P2002') {
          throw error;
        }
        const concurrent = await this.prisma.cmppConnectionState.findFirst({
          where: { applicationId: null, channelId: channel.id, connectionId },
        });
        if (!concurrent) {
          throw error;
        }
        state = await this.prisma.cmppConnectionState.update({ where: { id: concurrent.id }, data });
      }
    }
    await this.prisma.operationLog.create({
      data: {
        userId: operatorId,
        action: 'cmpp_connection.connect_requested',
        resource: 'cmpp_connection',
        resourceId: `${channel.id}:${connectionId}`,
        detail: {
          reason,
          status: state.status,
          desiredConnections: state.desiredConnections,
          currentConnections: state.currentConnections,
        } as Prisma.InputJsonValue,
      },
    });
    const command = {
      schemaVersion: 'v1',
      messageType: 'ConnectChannel',
      traceId: randomUUID(),
      channelId: channel.id,
      connectionId,
      createdAt: new Date().toISOString(),
      reason,
      desiredConnections,
      channel: {
        code: channel.code,
        name: channel.name,
        gatewayHost: channel.gatewayHost,
        gatewayPort: channel.gatewayPort,
        account: channel.account,
        passwordCipher: channel.passwordCipher,
        srcId: channel.srcId,
        cmppVersion: channel.cmppVersion,
        rateLimitPerSecond: channel.rateLimitPerSecond,
        windowSize: getPositiveRuntimeInteger(getConfigValue(channel.config, 'windowSize'), 16, 'windowSize'),
        connectionWarmupSeconds: Number(getConfigValue(channel.config, 'connectionWarmupSeconds') ?? 30),
        connectionDrainTimeoutSeconds: getPositiveRuntimeInteger(getConfigValue(channel.config, 'connectionDrainTimeoutSeconds'), 60, 'connectionDrainTimeoutSeconds'),
        submitResponseTimeoutSeconds: getPositiveRuntimeInteger(getConfigValue(channel.config, 'submitResponseTimeoutSeconds'), 60, 'submitResponseTimeoutSeconds'),
        connectionFailureCooldownSeconds: getPositiveRuntimeInteger(getConfigValue(channel.config, 'connectionFailureCooldownSeconds'), 30, 'connectionFailureCooldownSeconds'),
        heartbeatIntervalSeconds: getPositiveRuntimeInteger(
          getConfigValue(channel.config, 'heartbeatIntervalSeconds'),
          DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
          'heartbeatIntervalSeconds',
        ),
        heartbeatMissThreshold: getPositiveRuntimeInteger(
          getConfigValue(channel.config, 'heartbeatMissThreshold'),
          DEFAULT_HEARTBEAT_MISS_THRESHOLD,
          'heartbeatMissThreshold',
        ),
      },
    };
    const queuedJob = await this.getGatewayConnectionQueue().add('connect-channel', command, {
      jobId: `gateway-connect-${channel.id}-${command.traceId}`,
      removeOnComplete: 1000,
      removeOnFail: 1000,
    }).catch((error) => {
      this.logger.warn(`Gateway connect marker enqueue failed; continuing with direct control request: ${error instanceof Error ? error.message : String(error)}`);
      return undefined;
    });
    try {
      await this.notifyGatewayConnect(command);
    } finally {
      if (queuedJob) {
        await queuedJob.remove().catch((error) => {
          this.logger.warn(`Failed to remove delivered Gateway connect marker ${queuedJob.id}: ${error instanceof Error ? error.message : String(error)}`);
        });
      }
    }
    return state;
  }

  async reconnectActiveChannelsAfterGatewayRestart() {
    const channels = await this.prisma.smsChannel.findMany({ where: { status: 'active' } });
    const results = await Promise.allSettled(
      channels.map((channel) => this.requestChannelConnection(channel, 'gateway_restarted')),
    );
    results.forEach((result, index) => {
      if (result.status === 'rejected') {
        const channel = channels[index];
        this.logger.error(
          `Failed to restore active CMPP channel ${channel?.code ?? channel?.id ?? index}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`,
        );
      }
    });
  }

  async reconcileGatewayConnections(now = new Date()) {
    const channels = await this.prisma.smsChannel.findMany({
      where: { status: { in: ['active', 'disabled', 'deleted'] } },
      include: {
        connectionStates: {
          where: { applicationId: null },
        },
      },
      take: 200,
    });
    let reconnectRequested = 0;
    let disconnectRequested = 0;
    for (const channel of channels) {
      const state = channel.connectionStates.find((item) => item.connectionId === defaultChannelConnectionId(channel.id));
      if (channel.status !== 'active') {
        if (state && (state.currentConnections > 0 || ['connected', 'connecting', 'reconnecting'].includes(state.status))) {
          await this.withGatewayReconcileLock(channel.id, async () => {
            await this.requestChannelDisconnection(channel, 'inactive_channel_reconcile');
            disconnectRequested++;
          });
        }
        continue;
      }
      const desiredConnections = getDesiredConnections(channel.config);
      const heartbeatIntervalSeconds = getPositiveRuntimeInteger(
        getConfigValue(channel.config, 'heartbeatIntervalSeconds'),
        DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
        'heartbeatIntervalSeconds',
      );
      const heartbeatMissThreshold = getPositiveRuntimeInteger(
        getConfigValue(channel.config, 'heartbeatMissThreshold'),
        DEFAULT_HEARTBEAT_MISS_THRESHOLD,
        'heartbeatMissThreshold',
      );
      const heartbeatCutoff = new Date(now.getTime() - heartbeatIntervalSeconds * (heartbeatMissThreshold + 1) * 1000);
      const connectedAndFresh = state?.status === 'connected'
        && state.currentConnections >= desiredConnections
        && Boolean(state.lastHeartbeatAt && state.lastHeartbeatAt > heartbeatCutoff);
      const retryDue = !state?.nextReconnectAt || state.nextReconnectAt <= now;
      if (!connectedAndFresh && retryDue) {
        await this.withGatewayReconcileLock(channel.id, async () => {
          await this.requestChannelConnection(channel, 'automatic_reconnect');
          reconnectRequested++;
        });
      }
    }
    return { scanned: channels.length, reconnectRequested, disconnectRequested };
  }

  async withGatewayReconcileLock(channelId: string, action: () => Promise<void>) {
    const redis = this.getRedis();
    const key = `cmpp:gateway:reconcile:${channelId}`;
    const token = randomUUID();
    const acquired = await redis.set(key, token, 'PX', DEFAULT_CONNECTING_TIMEOUT_MS, 'NX');
    if (acquired !== 'OK') {
      return;
    }
    try {
      await action();
    } finally {
      await redis.eval(
        'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end',
        1,
        key,
        token,
      );
    }
  }

  async requestChannelDisconnection(
    channel: { id: string },
    reason: 'channel_disabled' | 'channel_deleted' | 'inactive_channel_reconcile',
    operatorId?: string,
  ) {
    const connectionId = defaultChannelConnectionId(channel.id);
    const now = new Date();
    await this.prisma.cmppConnectionState.updateMany({
      where: {
        applicationId: null,
        channelId: channel.id,
        connectionId,
      },
      data: {
        status: 'disconnected',
        currentConnections: 0,
        lastDisconnectedAt: now,
        nextReconnectAt: null,
        lastErrorCategory: null,
        lastError: null,
      },
    });
    await this.prisma.operationLog.create({
      data: {
        userId: operatorId,
        action: 'cmpp_connection.disconnect_requested',
        resource: 'cmpp_connection',
        resourceId: `${channel.id}:${connectionId}`,
        detail: { reason } as Prisma.InputJsonValue,
      },
    });
    const command = {
      schemaVersion: 'v1',
      messageType: 'DisconnectChannel',
      traceId: randomUUID(),
      channelId: channel.id,
      connectionId,
      createdAt: now.toISOString(),
      reason,
    };
    const queuedJob = await this.getGatewayConnectionQueue().add('disconnect-channel', command, {
      jobId: `gateway-disconnect-${channel.id}-${command.traceId}`,
      removeOnComplete: 1000,
      removeOnFail: 1000,
    }).catch((error) => {
      this.logger.warn(`Gateway disconnect marker enqueue failed; continuing with direct control request: ${error instanceof Error ? error.message : String(error)}`);
      return undefined;
    });
    try {
      await this.notifyGatewayDisconnect(command);
    } finally {
      if (queuedJob) {
        await queuedJob.remove().catch((error) => {
          this.logger.warn(`Failed to remove delivered Gateway disconnect marker ${queuedJob.id}: ${error instanceof Error ? error.message : String(error)}`);
        });
      }
    }
  }

  getGatewayConnectionQueue() {
    this.gatewayConnectionQueue ??= new Queue(GATEWAY_CONNECTION_QUEUE, { connection: bullmqConnection() });
    return this.gatewayConnectionQueue;
  }

  getGatewaySubmitQueue() {
    this.gatewaySubmitQueue ??= new Queue(GATEWAY_SUBMIT_QUEUE, { connection: bullmqConnection() });
    return this.gatewaySubmitQueue;
  }

  getRedis() {
    if (!this.redis) {
      this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', {
        maxRetriesPerRequest: null,
      });
    }
    return this.redis;
  }

  async publishGatewaySubmitCommand(command: unknown) {
    return this.getRedis().xadd(
      process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM,
      '*',
      'messageType',
      'SubmitCommand',
      'data',
      JSON.stringify(command),
    );
  }

  async notifyGatewayConnect(command: Record<string, unknown>) {
    const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? DEFAULT_GATEWAY_CONTROL_URL).replace(/\/+$/, '');
    let response: { ok: boolean; status: number; text: () => Promise<string> };
    try {
      response = await fetch(`${baseUrl}/connections/connect`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(command),
        signal: AbortSignal.timeout(getPositiveIntegerEnv('GATEWAY_CONTROL_TIMEOUT_MS', DEFAULT_GATEWAY_CONTROL_TIMEOUT_MS)),
      });
    } catch (error) {
      throw new BadRequestException(`Gateway connect request failed: ${error instanceof Error ? error.message : String(error)}`);
    }
    if (!response.ok) {
      const responseText = await response.text();
      throw new BadRequestException(`Gateway connect request failed: ${response.status} ${responseText}`);
    }
  }

  async notifyGatewayDisconnect(command: Record<string, unknown>) {
    const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? DEFAULT_GATEWAY_CONTROL_URL).replace(/\/+$/, '');
    let response: { ok: boolean; status: number; text: () => Promise<string> };
    try {
      response = await fetch(`${baseUrl}/connections/disconnect`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(command),
        signal: AbortSignal.timeout(getPositiveIntegerEnv('GATEWAY_CONTROL_TIMEOUT_MS', DEFAULT_GATEWAY_CONTROL_TIMEOUT_MS)),
      });
    } catch (error) {
      throw new BadRequestException(`Gateway disconnect request failed: ${error instanceof Error ? error.message : String(error)}`);
    }
    if (!response.ok) {
      const responseText = await response.text();
      throw new BadRequestException(`Gateway disconnect request failed: ${response.status} ${responseText}`);
    }
  }
}