fix: normalized the region and logged the otp properly

apps/edr-freight-api/src/modules/notifications/broker.util.spec.ts

@@ -0,0 +1,90 @@
+import { Logger } from '@nestjs/common';
+import { ClientProxy } from '@nestjs/microservices';
+import { NEVER, Observable, throwError } from 'rxjs';
+
+import { isBrokerConnected, publishConfirmed } from './broker.util';
+
+/**
+ * `ClientProxy.emit()` returns a cold Observable that, for RMQ, completes without
+ * emitting once `dispatchEvent` settles — and rejects if the publish fails. These
+ * fakes reproduce each of those three shapes.
+ */
+function clientEmitting(source: Observable<unknown>): ClientProxy {
+  return { emit: jest.fn().mockReturnValue(source) } as unknown as ClientProxy;
+}
+
+describe('publishConfirmed', () => {
+  const logger = { error: jest.fn() } as unknown as Logger;
+
+  beforeEach(() => jest.clearAllMocks());
+
+  it('is true when the publish completes (broker confirmed)', async () => {
+    // Completes with no value — the success shape, and the case that throws
+    // EmptyError without a defaultIfEmpty.
+    const client = clientEmitting(new Observable<never>((s) => s.complete()));
+    await expect(publishConfirmed(client, 'send-sms', {}, logger)).resolves.toBe(true);
+  });
+
+  it('is false when the publish never settles, rather than hanging', async () => {
+    // A broker that is down: amqp-connection-manager buffers the publish and the
+    // promise would never resolve. The timeout is what stops one dead broker from
+    // hanging every caller of sendSms/sendEmail.
+    const client = clientEmitting(NEVER);
+    await expect(publishConfirmed(client, 'send-sms', {}, logger, 20)).resolves.toBe(
+      false,
+    );
+    expect(logger.error).toHaveBeenCalled();
+  });
+
+  it('is false when the publish errors', async () => {
+    const client = clientEmitting(throwError(() => new Error('channel closed')));
+    await expect(publishConfirmed(client, 'send-email', {}, logger)).resolves.toBe(
+      false,
+    );
+    expect(logger.error).toHaveBeenCalled();
+  });
+});
+
+describe('isBrokerConnected', () => {
+  /** Stands in for `ClientProxy.unwrap()`, which returns the AmqpConnectionManager. */
+  function clientUnwrapping(manager: unknown): ClientProxy {
+    return { unwrap: () => manager } as unknown as ClientProxy;
+  }
+
+  it('reports the connection manager state', () => {
+    expect(isBrokerConnected(clientUnwrapping({ isConnected: () => true }))).toBe(
+      true,
+    );
+    expect(isBrokerConnected(clientUnwrapping({ isConnected: () => false }))).toBe(
+      false,
+    );
+  });
+
+  it('is false when unwrap throws — the client never connected', () => {
+    // ClientRMQ.unwrap() throws "Not initialized" while its internal client is
+    // null, which is what a failed boot-time connect leaves behind. That is a
+    // real down signal and must not be softened to "unknown".
+    const uninitialised = {
+      unwrap: () => {
+        throw new Error('Not initialized. Please call the "connect" method first.');
+      },
+    } as unknown as ClientProxy;
+    expect(isBrokerConnected(uninitialised)).toBe(false);
+  });
+
+  it('is null — not a guess — when the manager lacks isConnected or it throws', () => {
+    // Guards the health endpoint against reporting "ok" if amqp-connection-manager
+    // or Nest changes shape and the accessor we rely on disappears.
+    expect(isBrokerConnected(clientUnwrapping(null))).toBeNull();
+    expect(isBrokerConnected(clientUnwrapping({}))).toBeNull();
+    expect(
+      isBrokerConnected(
+        clientUnwrapping({
+          isConnected: () => {
+            throw new Error('boom');
+          },
+        }),
+      ),
+    ).toBeNull();
+  });
+});

apps/edr-freight-api/src/modules/notifications/broker.util.ts

@@ -0,0 +1,92 @@
+// 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;
+  }
+}

apps/edr-freight-api/src/modules/notifications/email-client.service.ts

@@ -6,6 +6,7 @@ import {
 } from "@nestjs/common";
 import { ClientProxy } from "@nestjs/microservices";
 import { SendEmailDto } from "./dtos/email.dto";
+import { isBrokerConnected, publishConfirmed } from "./broker.util";
 
 @Injectable()
 export class EmailClientService implements OnApplicationBootstrap {
@@ -33,19 +34,34 @@ export class EmailClientService implements OnApplicationBootstrap {
       this.logger.warn(`RABBITMQ disabled — skipped EMAIL to=${dto.to}`);
       return { queued: false };
     }
-    this.emailClient.emit("send-email", {
-      to: dto.to,
-      subject: dto.subject,
-      text: dto.text,
-      html: dto.html,
-      appKey: "IFHCRS-LICENSE-MANAGEMENT",
-    });
-    // Fire-and-forget enqueue: confirms hand-off to RabbitMQ, NOT delivery.
+    const queued = await publishConfirmed(
+      this.emailClient,
+      "send-email",
+      {
+        to: dto.to,
+        subject: dto.subject,
+        text: dto.text,
+        html: dto.html,
+        appKey: "IFHCRS-LICENSE-MANAGEMENT",
+      },
+      this.logger,
+    );
+    // Publisher-confirmed: RabbitMQ has taken ownership of the message. Still NOT
+    // delivery — the consumer and the SMTP hop are downstream and invisible here.
     this.logger.log(
-      `EMAIL queued to RabbitMQ [${process.env.EMAIL_QUEUE ?? "email_queue"}] pattern='send-email'`,
+      `EMAIL publish to RabbitMQ [${process.env.EMAIL_QUEUE ?? "email_queue"}] pattern='send-email' confirmed=${queued}`,
     );
     // Recipient + content are PII — debug only.
     this.logger.debug(`EMAIL payload to=${dto.to} subject="${dto.subject}"`);
-    return { queued: true };
+    return { queued };
+  }
+
+  /**
+   * Connection state for the health endpoint. `null` means the broker client did
+   * not expose its manager — reported as "unknown" rather than assumed healthy.
+   */
+  get brokerConnected(): boolean | null {
+    if (!this.enabled) return false;
+    return isBrokerConnected(this.emailClient);
   }
 }

apps/edr-freight-api/src/modules/notifications/sms-client.service.ts

@@ -6,6 +6,7 @@ import {
 } from "@nestjs/common";
 import { ClientProxy } from "@nestjs/microservices";
 import { BulkMessagesDto, SingleMessageDto } from "./dtos/sms.dto";
+import { isBrokerConnected, publishConfirmed } from "./broker.util";
 
 @Injectable()
 export class SmsClientService implements OnApplicationBootstrap {
@@ -14,7 +15,7 @@ export class SmsClientService implements OnApplicationBootstrap {
   constructor(
     @Inject("SMS_SERVICE")
     private smsClient: ClientProxy,
-  ) {}
+  ) { }
 
   private readonly enabled = process.env.RABBITMQ_ENABLED !== "false";
 
@@ -26,7 +27,7 @@ export class SmsClientService implements OnApplicationBootstrap {
         this.logger.log("connected to SMS service");
       })
       .catch((err) => {
-        console.error("Error happened at SMS service", err);
+        this.logger.error("Error happened at SMS service", err);
       });
   }
 
@@ -35,34 +36,61 @@ export class SmsClientService implements OnApplicationBootstrap {
       this.logger.warn(`RABBITMQ disabled — skipped SMS`);
       return { queued: false };
     }
-    this.smsClient.emit("send-sms", {
-      to: dto.to,
-      text: dto.message,
-      appKey: "IFHCRS-LICENSE-MANAGEMENT",
-    });
-    // Fire-and-forget enqueue: confirms hand-off to RabbitMQ, NOT delivery.
+    const queued = await publishConfirmed(
+      this.smsClient,
+      "send-sms",
+      {
+        to: dto.to,
+        text: dto.message,
+        appKey: "IFHCRS-LICENSE-MANAGEMENT",
+      },
+      this.logger,
+    );
+    // Publisher-confirmed: RabbitMQ has taken ownership of the message. Still NOT
+    // delivery — the consumer, the SMS gateway and the carrier are all downstream
+    // of this and invisible from here.
     this.logger.log(
-      `SMS queued to RabbitMQ [${process.env.SMS_QUEUE ?? "sms_queue"}] pattern='send-sms'`,
+      `SMS publish to RabbitMQ [${process.env.SMS_QUEUE ?? "sms_queue"}] pattern='send-sms' confirmed=${queued}`,
     );
     // Recipient + content are PII — debug only.
     this.logger.debug(`SMS payload to=${dto.to} text="${dto.message}"`);
-    return { queued: true };
+    return { queued };
   }
 
   async sendBulkMessages(dto: BulkMessagesDto): Promise<{ queued: boolean }> {
     if (!this.enabled) {
-      this.logger.warn(`RABBITMQ disabled — skipped BULK SMS (${dto.messages?.length ?? 0} messages)`);
+      this.logger.warn(
+        `RABBITMQ disabled — skipped BULK SMS (${dto.messages?.length ?? 0} messages)`,
+      );
       return { queued: false };
     }
-    const messages = (dto.messages ?? []).map((m) => ({ to: m.to, text: m.message, from: m.from }));
-    this.smsClient.emit("ozeking-bulk-sms", {
-      messages,
-      appKey: "IFHCRS-LICENSE-MANAGEMENT",
-    });
+    const messages = (dto.messages ?? []).map((m) => ({
+      to: m.to,
+      text: m.message,
+      from: m.from,
+    }));
+    const queued = await publishConfirmed(
+      this.smsClient,
+      "ozeking-bulk-sms",
+      {
+        messages,
+        appKey: "IFHCRS-LICENSE-MANAGEMENT",
+      },
+      this.logger,
+    );
     this.logger.log(
-      `BULK SMS queued to RabbitMQ [${process.env.SMS_QUEUE ?? "sms_queue"}] pattern='ozeking-bulk-sms' count=${messages.length}`,
+      `BULK SMS publish to RabbitMQ [${process.env.SMS_QUEUE ?? "sms_queue"}] pattern='ozeking-bulk-sms' count=${messages.length} confirmed=${queued}`,
     );
     this.logger.debug(`BULK SMS payload messages=${JSON.stringify(messages)}`);
-    return { queued: true };
+    return { queued };
+  }
+
+  /**
+   * Connection state for the health endpoint. `null` means the broker client did
+   * not expose its manager — reported as "unknown" rather than assumed healthy.
+   */
+  get brokerConnected(): boolean | null {
+    if (!this.enabled) return false;
+    return isBrokerConnected(this.smsClient);
   }
 }