// broker.util.ts

import { Logger } from "@nestjs/common";
import { ClientProxy } from "@nestjs/microservices";
import { defaultIfEmpty, lastValueFrom, timeout } from "rxjs";

/**
 * How long to wait for a publisher confirm before giving up on a message.
 *
 * Load-bearing, not a nicety: when the broker is unreachable
 * amqp-connection-manager buffers the publish and retries it on reconnect, so the
 * underlying promise never settles. Without a bound, one dead broker turns every
 * caller of sendSms/sendEmail into a hung request.
 */
export const PUBLISH_CONFIRM_TIMEOUT_MS = Number(
  process.env.RABBITMQ_PUBLISH_TIMEOUT_MS ?? 5000,
);

/**
 * Publish an event and wait for RabbitMQ to confirm it.
 *
 * `ClientProxy.emit()` returns a *cold* Observable. Called without subscribing —
 * as this codebase did everywhere — nothing forces the publish to be observed, so
 * the caller reports success whether or not the broker ever accepted the message.
 * Awaiting it drives `dispatchEvent`, which resolves only once
 * amqp-connection-manager's ChannelWrapper has a publisher confirm.
 *
 * So `true` here means the broker took ownership of the message. It still says
 * nothing about the consumer, the SMS gateway, or delivery to a handset — those
 * remain outside this process's knowledge.
 */
export async function publishConfirmed(
  client: ClientProxy,
  pattern: string,
  payload: unknown,
  logger: Logger,
  timeoutMs: number = PUBLISH_CONFIRM_TIMEOUT_MS,
): Promise<boolean> {
  try {
    // `emit` completes without emitting a value, so lastValueFrom needs a default
    // or it rejects with EmptyError on the success path.
    await lastValueFrom(
      client
        .emit(pattern, payload)
        .pipe(timeout(timeoutMs), defaultIfEmpty(undefined)),
    );
    return true;
  } catch (error) {
    logger.error(
      `broker.publish.failed pattern='${pattern}' timeoutMs=${timeoutMs}: ${
        error instanceof Error ? error.message : String(error)
      }`,
      error instanceof Error ? error.stack : undefined,
    );
    return false;
  }
}

/**
 * Whether the client's connection manager currently believes it is connected.
 *
 * Uses `ClientProxy.unwrap()` — Nest's public accessor for the underlying
 * transport client, which for `ClientRMQ` is the `AmqpConnectionManager`. Calling
 * `connect()` instead cannot answer this: it resolves against a *disconnected*
 * manager too, so it never distinguishes up from down.
 *
 * Three outcomes, deliberately distinct:
 *  - `false` when the manager reports disconnected, or when `unwrap()` throws
 *    because the client was never initialised (a failed boot-time connect leaves
 *    it null — genuinely down, not unknown);
 *  - `null` when the manager exists but has no `isConnected`, i.e. the library
 *    shape changed under us — the health endpoint reports "unknown" rather than
 *    quietly claiming health;
 *  - `true` only on an explicit positive from the manager.
 */
export function isBrokerConnected(client: ClientProxy): boolean | null {
  let manager: unknown;
  try {
    manager = client.unwrap<unknown>();
  } catch {
    // "Not initialized. Please call the connect method first." — no connection
    // was ever established, which is a real down signal, not an unknown one.
    return false;
  }
  const probe = manager as { isConnected?: () => boolean } | null;
  if (!probe || typeof probe.isConnected !== "function") return null;
  try {
    return probe.isConnected();
  } catch {
    return null;
  }
}