mirror of
https://github.com/Tria-plc/edr-platform.git
synced 2026-08-26 18:42:49 +00:00
93 lines
3.4 KiB
TypeScript
93 lines
3.4 KiB
TypeScript
// 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;
|
|
}
|
|
}
|