feat: ( payment ) create payment microservice

apps/edr-payment-api/src/modules/outbox/entities/notification-outbox.entity.ts

@@ -0,0 +1,55 @@
+import { Column, Entity, Index } from "typeorm";
+import { BaseEntity } from "@edr/api-common";
+import {
+  PaymentEvent,
+  PaymentEventType,
+  PaymentReferenceType,
+  PaymentService,
+} from "@edr/types";
+
+export type OutboxStatus = "PENDING" | "SENT" | "FAILED";
+
+/**
+ * Transactional outbox: a row is inserted in the SAME transaction that finalizes an intent,
+ * so "payment succeeded" and "a notification is owed" commit or roll back together. The relay
+ * drains PENDING rows and retries until acked (at-least-once delivery; consumers are idempotent).
+ */
+@Entity({ name: "notification_outbox" })
+@Index("idx_notification_outbox_relay", ["status", "nextRetryAt"])
+export class NotificationOutbox extends BaseEntity {
+  @Column({ name: "event_type", type: "varchar", length: 32 })
+  eventType!: PaymentEventType;
+
+  /** Routing discriminator — which app's mark-paid endpoint the relay delivers to. */
+  @Column({ name: "service", type: "varchar", length: 16 })
+  service!: PaymentService;
+
+  @Index("idx_notification_outbox_intent")
+  @Column({ name: "intent_id", type: "uuid" })
+  intentId!: string;
+
+  @Column({ name: "reference_type", type: "varchar", length: 16 })
+  referenceType!: PaymentReferenceType;
+
+  @Column({ name: "reference_id", type: "varchar", length: 64 })
+  referenceId!: string;
+
+  /** The full versioned event envelope delivered verbatim to the consumer. */
+  @Column({ name: "payload", type: "jsonb" })
+  payload!: PaymentEvent;
+
+  @Column({ name: "status", type: "varchar", length: 16, default: "PENDING" })
+  status!: OutboxStatus;
+
+  @Column({ name: "attempts", type: "integer", default: 0 })
+  attempts!: number;
+
+  @Column({ name: "next_retry_at", type: "timestamptz", nullable: true })
+  nextRetryAt?: Date | null;
+
+  @Column({ name: "last_error", type: "text", nullable: true })
+  lastError?: string | null;
+
+  @Column({ name: "sent_at", type: "timestamptz", nullable: true })
+  sentAt?: Date | null;
+}

apps/edr-payment-api/src/modules/outbox/outbox-relay.service.ts

@@ -0,0 +1,119 @@
+import {
+  Inject,
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+  OnModuleInit,
+} from "@nestjs/common";
+import { ConfigService } from "@nestjs/config";
+import { SchedulerRegistry } from "@nestjs/schedule";
+import { NotificationOutbox } from "./entities/notification-outbox.entity";
+import { OutboxRepository } from "./outbox.repository";
+import {
+  PAYMENT_EVENT_PUBLISHER,
+  PaymentEventPublisher,
+} from "./publisher/payment-event-publisher";
+
+const RELAY_INTERVAL_NAME = "outbox-relay";
+/** Retry backoff: base doubles per attempt, capped. */
+const BACKOFF_BASE_MS = 10_000;
+const BACKOFF_CAP_MS = 10 * 60_000;
+
+/**
+ * Drains the transactional outbox: PENDING rows are published (HTTP now, RabbitMQ later),
+ * marked SENT on ack, retried with exponential backoff on failure, and flagged FAILED after
+ * OUTBOX_MAX_ATTEMPTS (an alertable condition — delivery is at-least-once, never dropped
+ * silently). A crash between commit and publish only delays delivery.
+ */
+@Injectable()
+export class OutboxRelayService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(OutboxRelayService.name);
+  private readonly intervalMs: number;
+  private readonly maxAttempts: number;
+  private readonly batchSize: number;
+  private draining = false;
+
+  constructor(
+    config: ConfigService,
+    private readonly outboxRepository: OutboxRepository,
+    private readonly schedulerRegistry: SchedulerRegistry,
+    @Inject(PAYMENT_EVENT_PUBLISHER)
+    private readonly publisher: PaymentEventPublisher,
+  ) {
+    this.intervalMs = config.get<number>("notifier.relayIntervalMs") ?? 5_000;
+    this.maxAttempts = config.get<number>("notifier.maxAttempts") ?? 10;
+    this.batchSize = config.get<number>("notifier.relayBatchSize") ?? 20;
+  }
+
+  onModuleInit(): void {
+    const interval = setInterval(() => void this.drain(), this.intervalMs);
+    this.schedulerRegistry.addInterval(RELAY_INTERVAL_NAME, interval);
+  }
+
+  onModuleDestroy(): void {
+    if (this.schedulerRegistry.doesExist("interval", RELAY_INTERVAL_NAME)) {
+      this.schedulerRegistry.deleteInterval(RELAY_INTERVAL_NAME);
+    }
+  }
+
+  /** One relay pass; re-entrant ticks are skipped so slow deliveries don't overlap. */
+  async drain(): Promise<void> {
+    if (this.draining) return;
+    this.draining = true;
+    try {
+      const due = await this.outboxRepository.findDue(this.batchSize);
+      for (const row of due) {
+        await this.deliver(row);
+      }
+    } catch (err) {
+      this.logger.error(
+        `relay pass failed: ${err instanceof Error ? err.message : String(err)}`,
+      );
+    } finally {
+      this.draining = false;
+    }
+  }
+
+  private async deliver(row: NotificationOutbox): Promise<void> {
+    try {
+      await this.publisher.publish(row.payload);
+      await this.outboxRepository.markSent(row.id);
+    } catch (err) {
+      const message = this.describeError(err);
+      const attempts = row.attempts + 1;
+      const exhausted = attempts >= this.maxAttempts;
+      const backoffMs = Math.min(
+        BACKOFF_BASE_MS * 2 ** row.attempts,
+        BACKOFF_CAP_MS,
+      );
+      await this.outboxRepository.markAttemptFailed(
+        row,
+        message,
+        exhausted ? null : new Date(Date.now() + backoffMs),
+        exhausted,
+      );
+      if (exhausted) {
+        // ALERT: a paid order may not be confirmed in the owning app — needs operator action.
+        this.logger.error(
+          `outbox ${row.id} (${row.eventType} intent=${row.intentId}) FAILED after ${attempts} attempts: ${message}`,
+        );
+      } else {
+        this.logger.warn(
+          `outbox ${row.id} delivery attempt ${attempts} failed (retry in ${backoffMs}ms): ${message}`,
+        );
+      }
+    }
+  }
+
+  /** Connection failures surface as AggregateError with an empty message — dig out the code. */
+  private describeError(err: unknown): string {
+    if (err instanceof Error) {
+      if (err.message) return err.message;
+      const code = (err as { code?: string }).code;
+      if (code) return code;
+      const inner = (err as { errors?: unknown[] }).errors?.[0];
+      if (inner instanceof Error && inner.message) return inner.message;
+    }
+    return String(err);
+  }
+}

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

@@ -0,0 +1,20 @@
+import { Module } from "@nestjs/common";
+import { HttpModule } from "@nestjs/axios";
+import { TypeOrmModule } from "@nestjs/typeorm";
+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";
+
+@Module({
+  imports: [TypeOrmModule.forFeature([NotificationOutbox]), HttpModule],
+  providers: [
+    OutboxRepository,
+    OutboxRelayService,
+    // Swap to RabbitPaymentEventPublisher here when the broker lands — nothing else changes.
+    { provide: PAYMENT_EVENT_PUBLISHER, useClass: HttpPaymentEventPublisher },
+  ],
+  exports: [OutboxRepository],
+})
+export class OutboxModule {}

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

@@ -0,0 +1,58 @@
+import { Injectable } from "@nestjs/common";
+import { InjectRepository } from "@nestjs/typeorm";
+import { Repository } from "typeorm";
+import { BaseRepository } from "@edr/api-common";
+import { NotificationOutbox } from "./entities/notification-outbox.entity";
+
+@Injectable()
+export class OutboxRepository extends BaseRepository<NotificationOutbox> {
+  constructor(
+    @InjectRepository(NotificationOutbox)
+    repository: Repository<NotificationOutbox>,
+  ) {
+    super(repository);
+  }
+
+  /**
+   * PENDING rows whose retry time has come, oldest first. The relay runs as a single
+   * non-overlapping loop per instance; with multiple service instances this should move to a
+   * SELECT … FOR UPDATE SKIP LOCKED claim.
+   */
+  async findDue(limit: number): Promise<NotificationOutbox[]> {
+    return this.repository
+      .createQueryBuilder("outbox")
+      .where(`outbox.status = 'PENDING'`)
+      .andWhere(
+        "(outbox.next_retry_at IS NULL OR outbox.next_retry_at <= now())",
+      )
+      .orderBy("outbox.created_at", "ASC")
+      .take(limit)
+      .getMany();
+  }
+
+  async markSent(id: string): Promise<void> {
+    await this.update(id, {
+      status: "SENT",
+      sentAt: new Date(),
+      lastError: null,
+    });
+  }
+
+  async markAttemptFailed(
+    row: NotificationOutbox,
+    error: string,
+    nextRetryAt: Date | null,
+    exhausted: boolean,
+  ): Promise<void> {
+    await this.update(row.id, {
+      attempts: row.attempts + 1,
+      lastError: error,
+      nextRetryAt,
+      status: exhausted ? "FAILED" : "PENDING",
+    });
+  }
+
+  async countBacklog(): Promise<number> {
+    return this.repository.count({ where: { status: "PENDING" } });
+  }
+}

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

@@ -0,0 +1,66 @@
+import { randomUUID } from "node:crypto";
+import {
+  PaymentEvent,
+  PaymentFailedEvent,
+  PaymentSucceededEvent,
+} from "@edr/types";
+import { PaymentIntent } from "../intents/entities/payment-intent.entity";
+import { NotificationOutbox } from "./entities/notification-outbox.entity";
+
+/**
+ * Build a ready-to-insert outbox row for a terminal intent. Pure (no DI) so the intents
+ * state machine can insert it inside its own DB transaction without a module cycle.
+ * The row id is generated here because the event envelope embeds it as `eventId`.
+ */
+export function buildOutboxRow(
+  intent: PaymentIntent,
+  terminal:
+    | { eventType: "payment.succeeded"; providerTxnId?: string; paidAt: Date }
+    | {
+        eventType: "payment.failed";
+        failureCode?: string;
+        failureMessage?: string;
+      },
+): Partial<NotificationOutbox> {
+  const id = randomUUID();
+  const base = {
+    version: 1 as const,
+    eventId: id,
+    occurredAt: new Date().toISOString(),
+    service: intent.service,
+    intentId: intent.id,
+    referenceType: intent.referenceType,
+    referenceId: intent.referenceId,
+    merchantOrderId: intent.merchantOrderId,
+    provider: intent.provider,
+    amountMinor: intent.amountMinor,
+    currency: intent.currency,
+  };
+
+  const event: PaymentEvent =
+    terminal.eventType === "payment.succeeded"
+      ? ({
+          ...base,
+          eventType: "payment.succeeded",
+          providerTxnId: terminal.providerTxnId,
+          paidAt: terminal.paidAt.toISOString(),
+        } satisfies PaymentSucceededEvent)
+      : ({
+          ...base,
+          eventType: "payment.failed",
+          failureCode: terminal.failureCode,
+          failureMessage: terminal.failureMessage,
+        } satisfies PaymentFailedEvent);
+
+  return {
+    id,
+    eventType: event.eventType,
+    service: intent.service,
+    intentId: intent.id,
+    referenceType: intent.referenceType,
+    referenceId: intent.referenceId,
+    payload: event,
+    status: "PENDING",
+    attempts: 0,
+  };
+}

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

@@ -0,0 +1,53 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { ConfigService } from "@nestjs/config";
+import { HttpService } from "@nestjs/axios";
+import { firstValueFrom } from "rxjs";
+import { PaymentEvent, PaymentService } from "@edr/types";
+import { PaymentEventPublisher } from "./payment-event-publisher";
+
+/**
+ * Delivers events by POSTing to the owning app's idempotent mark-paid endpoint, routed by
+ * the `service` discriminator. Authenticated with the shared service token (the same secret
+ * the apps use to call /payments/initiate).
+ */
+@Injectable()
+export class HttpPaymentEventPublisher implements PaymentEventPublisher {
+  private readonly logger = new Logger(HttpPaymentEventPublisher.name);
+  private readonly routes: Record<PaymentService, string>;
+  private readonly timeoutMs: number;
+  private readonly serviceToken: string;
+
+  constructor(
+    config: ConfigService,
+    private readonly http: HttpService,
+  ) {
+    this.routes = {
+      [PaymentService.PASSENGER]:
+        config.get<string>("notifier.passengerUrl") ?? "",
+      [PaymentService.FREIGHT]: config.get<string>("notifier.freightUrl") ?? "",
+    };
+    this.timeoutMs = config.get<number>("notifier.httpTimeoutMs") ?? 10_000;
+    this.serviceToken = config.get<string>("app.serviceAuthToken") ?? "";
+  }
+
+  async publish(event: PaymentEvent): Promise<void> {
+    const url = this.routes[event.service];
+    if (!url) {
+      throw new Error(
+        `No mark-paid URL configured for service ${event.service}`,
+      );
+    }
+
+    const response = await firstValueFrom(
+      this.http.post(url, event, {
+        timeout: this.timeoutMs,
+        headers: this.serviceToken
+          ? { "x-service-token": this.serviceToken }
+          : {},
+      }),
+    );
+    this.logger.log(
+      `delivered ${event.eventType} (${event.eventId}) to ${event.service} — HTTP ${response.status}`,
+    );
+  }
+}

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

@@ -0,0 +1,13 @@
+import { PaymentEvent } from "@edr/types";
+
+/**
+ * Publisher port (architecture.md §12): how a payment event leaves this service.
+ * HTTP implementation now; a RabbitMQ implementation later is a DI swap only — the outbox
+ * and relay stay exactly as they are.
+ */
+export interface PaymentEventPublisher {
+  /** Deliver one event; throw on failure so the relay can retry with backoff. */
+  publish(event: PaymentEvent): Promise<void>;
+}
+
+export const PAYMENT_EVENT_PUBLISHER = Symbol("PAYMENT_EVENT_PUBLISHER");