refactor: ( payment ) use rabbitmq for webhooks event

apps/edr-payment-api/package.json

@@ -20,6 +20,7 @@
     "@edr/api-common": "workspace:*",
     "@edr/payment-providers": "workspace:*",
     "@edr/types": "workspace:*",
+    "@golevelup/nestjs-rabbitmq": "^5.5.0",
     "@nestjs/axios": "^4.0.1",
     "@nestjs/common": "^11.0.0",
     "@nestjs/config": "^4.0.0",

apps/edr-payment-api/src/app.module.ts

@@ -5,6 +5,7 @@ import { TypeOrmModule, TypeOrmModuleOptions } from "@nestjs/typeorm";
 import appConfig from "./config/app.config";
 import databaseConfig from "./config/database.config";
 import notifierConfig from "./config/notifier.config";
+import rabbitmqConfig from "./config/rabbitmq.config";
 import telebirrConfig from "./config/telebirr.config";
 import waafiConfig from "./config/waafi.config";
 import cbeConfig from "./config/cbe.config";
@@ -26,6 +27,7 @@ import { WebhooksModule } from "./modules/webhooks/webhooks.module";
         appConfig,
         databaseConfig,
         notifierConfig,
+        rabbitmqConfig,
         telebirrConfig,
         waafiConfig,
         cbeConfig,

apps/edr-payment-api/src/config/rabbitmq.config.ts

@@ -0,0 +1,23 @@
+import { registerAs } from "@nestjs/config";
+
+/**
+ * RabbitMQ connection for the outbox event publisher (payment -> apps).
+ * `publisherTransport` selects which PaymentEventPublisher the outbox relay uses:
+ *   "rabbitmq" -> RabbitMqPaymentEventPublisher (publish to the payment.events exchange) [DEFAULT]
+ *   "http"     -> HttpPaymentEventPublisher (POST to each app's mark-paid endpoint) [rollback]
+ * RabbitMQ is the default; set PUBLISHER_TRANSPORT=http explicitly to fall back to HTTP.
+ */
+export default registerAs("rabbitmq", () => ({
+  url: process.env.PAYMENT_RABBITMQ_URL ?? "amqp://localhost:5672/payment",
+  publisherTransport: (process.env.PUBLISHER_TRANSPORT ?? "rabbitmq").toLowerCase(),
+  /** Publish is failed (so the relay retries) if the broker does not confirm within this. */
+  publishTimeoutMs: parseInt(process.env.PAYMENT_PUBLISH_TIMEOUT_MS ?? "10000", 10),
+}));
+
+/**
+ * True when the outbox relay should publish to RabbitMQ rather than POST over HTTP.
+ * RabbitMQ is the default; only an explicit PUBLISHER_TRANSPORT=http opts out.
+ */
+export function isRabbitPublisher(): boolean {
+  return (process.env.PUBLISHER_TRANSPORT ?? "rabbitmq").toLowerCase() !== "http";
+}

apps/edr-payment-api/src/modules/outbox/outbox.module.ts

@@ -1,19 +1,56 @@
 import { Module } from "@nestjs/common";
 import { HttpModule } from "@nestjs/axios";
+import { ConfigService } from "@nestjs/config";
 import { TypeOrmModule } from "@nestjs/typeorm";
+import { RabbitMQModule } from "@golevelup/nestjs-rabbitmq";
+import { PAYMENT_EVENTS_DLX, PAYMENT_EVENTS_EXCHANGE } from "@edr/types";
+import { isRabbitPublisher } from "../../config/rabbitmq.config";
 import { NotificationOutbox } from "./entities/notification-outbox.entity";
 import { OutboxRelayService } from "./outbox-relay.service";
 import { OutboxRepository } from "./outbox.repository";
 import { HttpPaymentEventPublisher } from "./publisher/http-payment-event-publisher";
 import { PAYMENT_EVENT_PUBLISHER } from "./publisher/payment-event-publisher";
+import { RabbitMqPaymentEventPublisher } from "./publisher/rabbitmq-payment-event-publisher";
+
+const rabbitImports = isRabbitPublisher()
+  ? [
+      RabbitMQModule.forRootAsync({
+        inject: [ConfigService],
+        useFactory: (config: ConfigService) => ({
+          uri: config.get<string>("rabbitmq.url") as string,
+          exchanges: [
+            {
+              name: PAYMENT_EVENTS_EXCHANGE,
+              type: "topic",
+              options: { durable: true },
+            },
+            {
+              name: PAYMENT_EVENTS_DLX,
+              type: "topic",
+              options: { durable: true },
+            },
+          ],
+          connectionInitOptions: { wait: false },
+        }),
+      }),
+    ]
+  : [];
 
 @Module({
-  imports: [TypeOrmModule.forFeature([NotificationOutbox]), HttpModule],
+  imports: [
+    TypeOrmModule.forFeature([NotificationOutbox]),
+    HttpModule,
+    ...rabbitImports,
+  ],
   providers: [
     OutboxRepository,
     OutboxRelayService,
-    // Swap to RabbitPaymentEventPublisher here when the broker lands — nothing else changes.
-    { provide: PAYMENT_EVENT_PUBLISHER, useClass: HttpPaymentEventPublisher },
+    {
+      provide: PAYMENT_EVENT_PUBLISHER,
+      useClass: isRabbitPublisher()
+        ? RabbitMqPaymentEventPublisher
+        : HttpPaymentEventPublisher,
+    },
   ],
   exports: [OutboxRepository],
 })

apps/edr-payment-api/src/modules/outbox/publisher/rabbitmq-payment-event-publisher.ts

@@ -0,0 +1,61 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { ConfigService } from "@nestjs/config";
+import { AmqpConnection } from "@golevelup/nestjs-rabbitmq";
+import {
+  PAYMENT_EVENTS_EXCHANGE,
+  PaymentEvent,
+  paymentRoutingKey,
+} from "@edr/types";
+import { PaymentEventPublisher } from "./payment-event-publisher";
+
+@Injectable()
+export class RabbitMqPaymentEventPublisher implements PaymentEventPublisher {
+  private readonly logger = new Logger(RabbitMqPaymentEventPublisher.name);
+  private readonly publishTimeoutMs: number;
+
+  constructor(
+    private readonly amqp: AmqpConnection,
+    config: ConfigService,
+  ) {
+    this.publishTimeoutMs =
+      config.get<number>("rabbitmq.publishTimeoutMs") ?? 10_000;
+  }
+
+  async publish(event: PaymentEvent): Promise<void> {
+    const routingKey = paymentRoutingKey(event.service, event.eventType);
+
+    const confirmed = await this.withTimeout(
+      this.amqp.publish(PAYMENT_EVENTS_EXCHANGE, routingKey, event, {
+        persistent: true,
+        messageId: event.eventId,
+        contentType: "application/json",
+        type: event.eventType,
+        headers: { "x-event-version": event.version },
+      }),
+    );
+
+    if (!confirmed) {
+      throw new Error(
+        `broker did not confirm publish of ${event.eventType} (${event.eventId})`,
+      );
+    }
+
+    this.logger.log(
+      `published ${event.eventType} (${event.eventId}) -> ${routingKey}`,
+    );
+  }
+
+  private withTimeout(p: Promise<boolean>): Promise<boolean> {
+    let timer: NodeJS.Timeout;
+    const timeout = new Promise<boolean>((_, reject) => {
+      timer = setTimeout(
+        () =>
+          reject(
+            new Error(`publish confirm timed out after ${this.publishTimeoutMs}ms`),
+          ),
+        this.publishTimeoutMs,
+      );
+    });
+    return Promise.race([p, timeout]).finally(() => clearTimeout(timer));
+  }
+}