| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854 |
- // apps/box-app-api/src/rabbitmq/rabbitmq-publisher.service.ts
- import {
- Injectable,
- Logger,
- OnModuleDestroy,
- OnModuleInit,
- } from '@nestjs/common';
- import { ConfigService } from '@nestjs/config';
- import { Connection, ConfirmChannel } from 'amqplib';
- import * as amqp from 'amqplib';
- import { UserLoginEventPayload } from '@box/common/events/user-login-event.dto';
- import { AdsClickEventPayload } from '@box/common/events/ads-click-event.dto';
- import { nowEpochMsBigInt } from '@box/common/time/time.util';
- import { RedisService } from '@box/db/redis/redis.service';
- type StatsAdClickRoutingKey = string;
- type StatsVideoClickRoutingKey = string;
- type StatsAdImpressionRoutingKey = string;
- export interface StatsAdClickEventPayload {
- messageId: string;
- uid: string;
- adId: string;
- adType: string;
- clickedAt: bigint;
- ip: string;
- }
- export interface StatsVideoClickEventPayload {
- messageId: string;
- uid: string;
- videoId: string;
- clickedAt: bigint;
- ip: string;
- }
- export interface StatsAdImpressionEventPayload {
- messageId: string;
- uid: string;
- adId: string;
- adType: string;
- impressionAt: bigint;
- visibleDurationMs?: number;
- ip: string;
- }
- // Circuit breaker states
- enum CircuitBreakerState {
- CLOSED = 'CLOSED', // Normal operation
- OPEN = 'OPEN', // Failing, reject requests
- HALF_OPEN = 'HALF_OPEN', // Testing if service recovered
- }
- interface CircuitBreakerConfig {
- failureThreshold: number; // Number of failures to open circuit
- successThreshold: number; // Number of successes to close circuit
- timeout: number; // Time in ms to wait before trying again (half-open)
- }
- @Injectable()
- export class RabbitmqPublisherService implements OnModuleInit, OnModuleDestroy {
- private readonly logger = new Logger(RabbitmqPublisherService.name);
- private connection?: Connection;
- private channel?: ConfirmChannel;
- private exchange!: string;
- private routingKeyLogin!: string;
- private routingKeyAdsClick!: string;
- private statsExchange!: string;
- private routingKeyStatsAdClick!: StatsAdClickRoutingKey;
- private routingKeyStatsVideoClick!: StatsVideoClickRoutingKey;
- private routingKeyStatsAdImpression!: StatsAdImpressionRoutingKey;
- private dlqExchange!: string;
- // Circuit breaker state
- private circuitState: CircuitBreakerState = CircuitBreakerState.CLOSED;
- private failureCount = 0;
- private successCount = 0;
- private nextAttemptTime = 0;
- private readonly circuitConfig: CircuitBreakerConfig = {
- failureThreshold: 5, // Open circuit after 5 failures
- successThreshold: 2, // Close circuit after 2 successes
- timeout: 60000, // Wait 60s before trying again
- };
- // Reconnection state
- private isReconnecting = false;
- private reconnectionScheduled = false;
- // Retry configuration
- private readonly maxRetries = 3;
- private readonly retryDelays = [100, 500, 2000]; // Exponential backoff
- // Message TTL (24 hours for fallback queue)
- private readonly messageTTL = 86400000; // 24 hours in ms
- private readonly idempotencyTTL = 604800; // 7 days in seconds
- constructor(
- private readonly config: ConfigService,
- private readonly redis: RedisService,
- ) {}
- async onModuleInit(): Promise<void> {
- const url = this.config.get<string>('RABBITMQ_URL');
- this.exchange =
- this.config.get<string>('RABBITMQ_LOGIN_EXCHANGE') ?? 'stats.user';
- this.routingKeyLogin =
- this.config.get<string>('RABBITMQ_LOGIN_ROUTING_KEY') ?? 'user.login';
- this.routingKeyAdsClick =
- this.config.get<string>('RABBITMQ_ADS_CLICK_ROUTING_KEY') ?? 'ads.click';
- this.statsExchange =
- this.config.get<string>('RABBITMQ_STATS_EXCHANGE') ?? this.exchange;
- this.routingKeyStatsAdClick =
- this.config.get<string>('RABBITMQ_STATS_AD_CLICK_ROUTING_KEY') ??
- 'stats.ad.click';
- this.routingKeyStatsVideoClick =
- this.config.get<string>('RABBITMQ_STATS_VIDEO_CLICK_ROUTING_KEY') ??
- 'stats.video.click';
- this.routingKeyStatsAdImpression =
- this.config.get<string>('RABBITMQ_STATS_AD_IMPRESSION_ROUTING_KEY') ??
- 'stats.ad.impression';
- this.dlqExchange =
- this.config.get<string>('RABBITMQ_DLQ_EXCHANGE') ?? 'dlq.stats';
- if (!url) {
- this.logger.error(
- 'RABBITMQ_URL is not set. Stats will be stored in Redis fallback queue only.',
- );
- this.circuitState = CircuitBreakerState.OPEN;
- return;
- }
- try {
- this.logger.log(`Connecting to RabbitMQ at ${url} ...`);
- await this.initializeConnection(url);
- this.logger.log('RabbitMQ connection initialized successfully');
- } catch (error) {
- this.logger.error(
- `Failed to initialize RabbitMQ connection: ${error}`,
- error instanceof Error ? error.stack : undefined,
- );
- this.circuitState = CircuitBreakerState.OPEN;
- this.nextAttemptTime = Date.now() + this.circuitConfig.timeout;
- }
- }
- private async initializeConnection(url: string): Promise<void> {
- this.connection = await amqp.connect(url);
- // Handle connection errors
- this.connection.on('error', (err) => {
- this.logger.error('RabbitMQ connection error:', err);
- this.openCircuit();
- });
- this.connection.on('close', () => {
- this.logger.warn('RabbitMQ connection closed');
- this.openCircuit();
- });
- // Use a confirm channel so we know when broker has accepted the message
- this.channel = await this.connection.createConfirmChannel();
- // Handle channel errors
- this.channel.on('error', (err) => {
- this.logger.error('RabbitMQ channel error:', err);
- this.openCircuit();
- });
- this.channel.on('close', () => {
- this.logger.warn('RabbitMQ channel closed');
- this.openCircuit();
- });
- // Assert exchanges with DLQ
- await this.channel.assertExchange(this.exchange, 'topic', {
- durable: true,
- });
- if (this.statsExchange !== this.exchange) {
- await this.channel.assertExchange(this.statsExchange, 'topic', {
- durable: true,
- });
- }
- // Assert Dead Letter Exchange
- await this.channel.assertExchange(this.dlqExchange, 'topic', {
- durable: true,
- });
- // Assert DLQ queue for stats events
- await this.channel.assertQueue('dlq.stats.events', {
- durable: true,
- arguments: {
- 'x-message-ttl': this.messageTTL, // Messages expire after 24 hours
- 'x-max-length': 100000, // Maximum 100k messages in DLQ
- },
- });
- // Bind DLQ queue to DLQ exchange
- // Routing convention: sendToDLQ() publishes with 'dlq.{original-routing-key}' format
- // Examples: dlq.stats.ad.click, dlq.stats.video.click, dlq.stats.ad.impression
- // Pattern 'dlq.#' matches all DLQ messages regardless of their original routing key
- await this.channel.bindQueue('dlq.stats.events', this.dlqExchange, 'dlq.#');
- this.logger.log(
- `RabbitMQ publisher ready. exchange="${this.exchange}", statsExchange="${this.statsExchange}", dlqExchange="${this.dlqExchange}"`,
- );
- }
- async onModuleDestroy(): Promise<void> {
- try {
- await this.channel?.close();
- await this.connection?.close();
- } catch (error: any) {
- this.logger.error('Error while closing RabbitMQ connection', error.stack);
- }
- }
- /**
- * Circuit breaker: Open circuit (stop attempting to send to RabbitMQ)
- */
- private openCircuit(): void {
- if (this.circuitState !== CircuitBreakerState.OPEN) {
- this.logger.warn(
- `Circuit breaker OPENED (failureCount=${this.failureCount}, successCount=${this.successCount}). Will retry after ${this.circuitConfig.timeout}ms`,
- );
- this.circuitState = CircuitBreakerState.OPEN;
- this.failureCount = 0;
- this.successCount = 0;
- this.nextAttemptTime = Date.now() + this.circuitConfig.timeout;
- // Schedule reconnection attempt
- this.scheduleReconnection();
- }
- }
- /**
- * Circuit breaker: Move to half-open state (test if service recovered)
- */
- private async halfOpenCircuit(): Promise<void> {
- this.logger.log(
- `Circuit breaker HALF-OPEN (failureCount=${this.failureCount}, successCount=${this.successCount}). Testing connection...`,
- );
- this.circuitState = CircuitBreakerState.HALF_OPEN;
- this.successCount = 0;
- // Attempt reconnection before allowing publish attempts
- await this.reconnectIfNeeded();
- }
- /**
- * Circuit breaker: Close circuit (resume normal operation)
- */
- private closeCircuit(): void {
- this.logger.log(
- `Circuit breaker CLOSED (failureCount=${this.failureCount}, successCount=${this.successCount}). Resuming normal operation.`,
- );
- this.circuitState = CircuitBreakerState.CLOSED;
- this.failureCount = 0;
- this.successCount = 0;
- }
- /**
- * Record successful publish (for circuit breaker)
- */
- private recordSuccess(): void {
- this.failureCount = 0;
- if (this.circuitState === CircuitBreakerState.HALF_OPEN) {
- this.successCount++;
- if (this.successCount >= this.circuitConfig.successThreshold) {
- this.closeCircuit();
- }
- }
- }
- /**
- * Record failed publish (for circuit breaker)
- */
- private recordFailure(): void {
- this.failureCount++;
- if (this.circuitState === CircuitBreakerState.HALF_OPEN) {
- this.openCircuit();
- } else if (
- this.circuitState === CircuitBreakerState.CLOSED &&
- this.failureCount >= this.circuitConfig.failureThreshold
- ) {
- this.openCircuit();
- }
- }
- /**
- * Check if circuit breaker allows request
- */
- private async canAttempt(): Promise<boolean> {
- if (this.circuitState === CircuitBreakerState.CLOSED) {
- return true;
- }
- if (this.circuitState === CircuitBreakerState.HALF_OPEN) {
- return true;
- }
- // OPEN state: check if timeout elapsed
- if (Date.now() >= this.nextAttemptTime) {
- await this.halfOpenCircuit();
- return true;
- }
- return false;
- }
- /**
- * Schedule a reconnection attempt after circuit timeout
- */
- private scheduleReconnection(): void {
- if (this.reconnectionScheduled) {
- return; // Already scheduled
- }
- this.reconnectionScheduled = true;
- this.logger.debug(
- `Scheduling reconnection attempt in ${this.circuitConfig.timeout}ms`,
- );
- setTimeout(async () => {
- this.reconnectionScheduled = false;
- if (this.circuitState === CircuitBreakerState.OPEN) {
- await this.halfOpenCircuit();
- }
- }, this.circuitConfig.timeout);
- }
- /**
- * Reconnect to RabbitMQ if connection or channel is closed/undefined
- */
- private async reconnectIfNeeded(): Promise<void> {
- // Check if reconnection is needed
- const connectionClosed =
- !this.connection || this.connection.connection?.destroyed;
- const channelClosed = !this.channel;
- if (!connectionClosed && !channelClosed) {
- this.logger.debug(
- 'Connection and channel are healthy, no reconnection needed',
- );
- return;
- }
- // Prevent concurrent reconnection attempts
- if (this.isReconnecting) {
- this.logger.debug('Reconnection already in progress, skipping');
- return;
- }
- this.isReconnecting = true;
- this.logger.log(
- `🔄 Starting RabbitMQ reconnection attempt (circuitState=${this.circuitState})...`,
- );
- try {
- // Get current URL from config
- const url = this.config.get<string>('RABBITMQ_URL');
- if (!url) {
- this.logger.error(
- '❌ Reconnection failed: RABBITMQ_URL is not set. Cannot reconnect to RabbitMQ.',
- );
- this.isReconnecting = false;
- return;
- }
- // Close existing connections if any
- try {
- await this.channel?.close();
- } catch (err) {
- // Ignore errors on close
- }
- try {
- await this.connection?.close();
- } catch (err) {
- // Ignore errors on close
- }
- // Clear references
- this.channel = undefined;
- this.connection = undefined;
- // Reinitialize connection
- this.logger.debug(`🔌 Reconnecting to RabbitMQ at ${url}...`);
- await this.initializeConnection(url);
- this.logger.log(
- `✅ RabbitMQ reconnection successful (hasConnection=${!!this.connection}, hasChannel=${!!this.channel})`,
- );
- this.isReconnecting = false;
- // Close circuit if reconnection succeeded
- if (this.circuitState === CircuitBreakerState.HALF_OPEN) {
- this.logger.log(
- 'Reconnection successful during HALF_OPEN, closing circuit',
- );
- this.closeCircuit();
- }
- } catch (error) {
- this.logger.error(
- `❌ RabbitMQ reconnection failed (circuitState=${this.circuitState}): ${error}`,
- error instanceof Error ? error.stack : undefined,
- );
- this.isReconnecting = false;
- // Keep circuit open on reconnection failure
- if (this.circuitState === CircuitBreakerState.HALF_OPEN) {
- this.logger.warn(
- '⚠️ Reconnection failed during HALF_OPEN, reopening circuit',
- );
- this.openCircuit();
- }
- }
- }
- /**
- * Check publisher-level idempotency: Has this message already been published?
- *
- * This prevents duplicate publishes from the publisher side within a 7-day window.
- * This is NOT end-to-end idempotency - consumers must perform their own duplicate
- * detection on the receiving end based on their business logic.
- *
- * Redis key format: rabbitmq:publish-idempotency:{messageId}
- * TTL: 7 days (604800 seconds)
- *
- * @param messageId - Unique message identifier (UUID)
- * @returns true if message was already published, false otherwise
- *
- * Note: On Redis errors, returns false (prefer duplicates over data loss)
- */
- private async checkIdempotency(messageId: string): Promise<boolean> {
- try {
- const key = `rabbitmq:publish-idempotency:${messageId}`;
- const exists = await this.redis.exists(key);
- return exists > 0;
- } catch (error) {
- this.logger.error(
- `Failed to check publish idempotency for ${messageId}: ${error}`,
- );
- // On Redis error, allow the message (better to have duplicate than lose data)
- return false;
- }
- }
- /**
- * Mark message as published (for publisher-level idempotency)
- *
- * Records that this messageId has been successfully published to RabbitMQ.
- * This prevents duplicate publishes from retry logic or circuit breaker recovery.
- *
- * Consumers still need to implement their own idempotency checks when processing
- * messages, as network issues or broker failures could cause duplicates downstream.
- *
- * Redis key format: rabbitmq:publish-idempotency:{messageId}
- * TTL: 7 days (604800 seconds)
- *
- * @param messageId - Unique message identifier (UUID)
- *
- * Note: Errors are logged but do not fail the publish operation
- */
- private async markAsProcessed(messageId: string): Promise<void> {
- try {
- const key = `rabbitmq:publish-idempotency:${messageId}`;
- await this.redis.set(key, '1', this.idempotencyTTL);
- } catch (error) {
- this.logger.error(
- `Failed to mark ${messageId} as published (idempotency): ${error}`,
- );
- }
- }
- /**
- * Store message in Redis fallback queue
- */
- private async storeInFallbackQueue(
- routingKey: string,
- payload: unknown,
- messageId: string,
- ): Promise<void> {
- try {
- const fallbackKey = `rabbitmq:fallback:${routingKey}:${messageId}`;
- await this.redis.setJson(fallbackKey, payload, 86400); // 24 hours TTL
- this.logger.warn(
- `Stored message ${messageId} in Redis fallback queue: ${fallbackKey}`,
- );
- } catch (error) {
- this.logger.error(
- `CRITICAL: Failed to store message ${messageId} in fallback queue: ${error}`,
- error instanceof Error ? error.stack : undefined,
- );
- }
- }
- /**
- * Send message to Dead Letter Queue
- */
- private async sendToDLQ(
- routingKey: string,
- payload: unknown,
- reason: string,
- ): Promise<void> {
- if (!this.channel) {
- this.logger.error(
- `Cannot send to DLQ: channel not available. Reason: ${reason}`,
- );
- return;
- }
- const dlqRoutingKey = `dlq.${routingKey}`;
- try {
- const payloadBuffer = this.toPayloadBuffer(payload);
- await new Promise<void>((resolve, reject) => {
- this.channel!.publish(
- this.dlqExchange,
- dlqRoutingKey,
- payloadBuffer,
- {
- persistent: true,
- contentType: 'application/json',
- headers: {
- 'x-death-reason': reason,
- 'x-death-timestamp': Date.now(),
- },
- },
- (err) => {
- if (err) {
- reject(err);
- } else {
- resolve();
- }
- },
- );
- });
- this.logger.warn(
- `Sent message to DLQ: exchange="${this.dlqExchange}", routingKey="${dlqRoutingKey}", queue="dlq.stats.events". Reason: ${reason}`,
- );
- } catch (error) {
- this.logger.error(
- `Failed to send message to DLQ (routingKey="${dlqRoutingKey}"): ${error}`,
- error instanceof Error ? error.stack : undefined,
- );
- }
- }
- /**
- * Retry logic with exponential backoff
- */
- private async retryPublish(
- publishFn: () => Promise<void>,
- context: string,
- ): Promise<void> {
- for (let attempt = 0; attempt < this.maxRetries; attempt++) {
- try {
- await publishFn();
- return; // Success
- } catch (error) {
- const isLastAttempt = attempt === this.maxRetries - 1;
- if (isLastAttempt) {
- this.logger.error(
- `Failed to publish after ${this.maxRetries} attempts (${context}): ${error}`,
- );
- throw error;
- }
- const delay = this.retryDelays[attempt];
- this.logger.warn(
- `Publish attempt ${attempt + 1} failed (${context}). Retrying in ${delay}ms...`,
- );
- await new Promise((resolve) => setTimeout(resolve, delay));
- }
- }
- }
- /**
- * Publish a user.login event.
- */
- async publishUserLogin(event: UserLoginEventPayload): Promise<void> {
- if (!this.channel) {
- this.logger.warn(
- 'RabbitMQ channel not ready. Skipping user.login publish.',
- );
- return;
- }
- const payloadBuffer = Buffer.from(JSON.stringify(event));
- return new Promise((resolve, reject) => {
- this.channel!.publish(
- this.exchange,
- this.routingKeyLogin,
- payloadBuffer,
- {
- persistent: true,
- contentType: 'application/json',
- },
- (err) => {
- if (err) {
- this.logger.error(
- `Failed to publish user.login event for uid=${event.uid}: ${err.message}`,
- err.stack,
- );
- return reject(err);
- }
- this.logger.debug(`Published user.login event for uid=${event.uid}`);
- resolve();
- },
- );
- });
- }
- /**
- * Publish an ads.click event.
- */
- async publishAdsClick(event: AdsClickEventPayload): Promise<void> {
- if (!this.channel) {
- this.logger.warn(
- 'RabbitMQ channel not ready. Skipping ads.click publish.',
- );
- return;
- }
- const payloadBuffer = Buffer.from(JSON.stringify(event));
- return new Promise((resolve, reject) => {
- this.channel!.publish(
- this.exchange,
- this.routingKeyAdsClick,
- payloadBuffer,
- {
- persistent: true,
- contentType: 'application/json',
- },
- (err) => {
- if (err) {
- this.logger.error(
- `Failed to publish ads.click event for adsId=${event.adsId}: ${err.message}`,
- err.stack,
- );
- return reject(err);
- }
- this.logger.debug(
- `Published ads.click event for adsId=${event.adsId}`,
- );
- resolve();
- },
- );
- });
- }
- /**
- * Publish stats.ad.click event with full error handling
- */
- async publishStatsAdClick(event: StatsAdClickEventPayload): Promise<void> {
- return this.publishStatsEventWithFallback(
- this.routingKeyStatsAdClick,
- event,
- event.messageId,
- `stats.ad.click adId=${event.adId}`,
- );
- }
- /**
- * Publish stats.video.click event with full error handling
- */
- async publishStatsVideoClick(
- event: StatsVideoClickEventPayload,
- ): Promise<void> {
- return this.publishStatsEventWithFallback(
- this.routingKeyStatsVideoClick,
- event,
- event.messageId,
- `stats.video.click videoId=${event.videoId}`,
- );
- }
- /**
- * Publish stats.ad.impression event with full error handling
- */
- async publishStatsAdImpression(
- event: StatsAdImpressionEventPayload,
- ): Promise<void> {
- return this.publishStatsEventWithFallback(
- this.routingKeyStatsAdImpression,
- event,
- event.messageId,
- `stats.ad.impression adId=${event.adId}`,
- );
- }
- /**
- * PUBLIC API for replaying messages from Redis fallback queue
- * Used by RabbitmqFallbackReplayService to republish failed messages
- *
- * IMPORTANT: This method will NOT store failed replays back to the fallback queue
- * to prevent infinite loops. Failed replays will only go to DLQ for manual inspection.
- *
- * @param routingKey - Original routing key (e.g., 'stats.ad.click')
- * @param payload - Original message payload
- * @param messageId - Original message ID (from payload.messageId)
- */
- async replayFallbackMessage(
- routingKey: string,
- payload: unknown,
- messageId: string,
- ): Promise<void> {
- // Use the same internal publish logic, but with a special context
- // to indicate this is a replay from fallback queue
- return this.publishStatsEventWithFallback(
- routingKey,
- payload,
- messageId,
- `fallback-replay routingKey=${routingKey}`,
- );
- }
- /**
- * Enhanced publish with circuit breaker, retry, fallback queue, DLQ, and idempotency
- */
- private async publishStatsEventWithFallback(
- routingKey: string,
- event: unknown,
- messageId: string,
- context: string,
- ): Promise<void> {
- // 1. Check idempotency
- const alreadyProcessed = await this.checkIdempotency(messageId);
- if (alreadyProcessed) {
- this.logger.debug(`Skipping duplicate message ${messageId} (${context})`);
- return;
- }
- // 2. Check circuit breaker
- if (!(await this.canAttempt())) {
- this.logger.warn(
- `Circuit breaker OPEN. Storing ${messageId} in fallback queue (${context})`,
- );
- await this.storeInFallbackQueue(routingKey, event, messageId);
- return;
- }
- // 3. Attempt to publish with retry logic
- try {
- await this.retryPublish(async () => {
- await this.publishStatsEvent(routingKey, event, context);
- }, context);
- // Success!
- this.recordSuccess();
- await this.markAsProcessed(messageId);
- this.logger.debug(`Successfully published ${messageId} (${context})`);
- } catch (error) {
- // All retries failed
- this.recordFailure();
- this.logger.error(
- `All retry attempts failed for ${messageId} (${context}): ${error}`,
- );
- // 4. Store in fallback queue
- await this.storeInFallbackQueue(routingKey, event, messageId);
- // 5. Send to DLQ for manual inspection
- await this.sendToDLQ(routingKey, event, `Max retries exceeded: ${error}`);
- // Don't throw error - fire-and-forget pattern
- }
- }
- /**
- * Core publish logic (used by retry mechanism)
- */
- private async publishStatsEvent(
- routingKey: string,
- event: unknown,
- context: string,
- ): Promise<void> {
- if (!this.channel) {
- throw new Error('RabbitMQ channel not ready');
- }
- const payloadBuffer = this.toPayloadBuffer(event);
- return new Promise((resolve, reject) => {
- this.channel!.publish(
- this.statsExchange,
- routingKey,
- payloadBuffer,
- {
- persistent: true,
- contentType: 'application/json',
- timestamp: Number(nowEpochMsBigInt()),
- expiration: this.messageTTL.toString(), // Message TTL
- },
- (err) => {
- if (err) {
- this.logger.error(
- `Failed to publish stats event (${context}): ${err.message}`,
- err.stack,
- );
- return reject(err);
- }
- this.logger.debug(
- `Published stats event (${context}) to ${this.statsExchange}/${routingKey}`,
- );
- resolve();
- },
- );
- });
- }
- private toPayloadBuffer(event: unknown): Buffer {
- const json = JSON.stringify(event, (_, value) =>
- typeof value === 'bigint' ? value.toString() : value,
- );
- return Buffer.from(json);
- }
- /**
- * Get circuit breaker status (for monitoring)
- */
- getCircuitStatus(): {
- state: CircuitBreakerState;
- failureCount: number;
- successCount: number;
- hasConnection: boolean;
- hasChannel: boolean;
- isReconnecting: boolean;
- nextAttemptTime: number;
- } {
- return {
- state: this.circuitState,
- failureCount: this.failureCount,
- successCount: this.successCount,
- hasConnection: !!(
- this.connection && !this.connection.connection?.destroyed
- ),
- hasChannel: !!this.channel,
- isReconnecting: this.isReconnecting,
- nextAttemptTime: this.nextAttemptTime,
- };
- }
- }
|