outbox pattern

Author: eiberham

apps/superchef/src/outbox/domain/outbox.interface.ts

@@ -0,0 +1,29 @@
+import { Prisma } from "generated/prisma/client";
+
+export interface OutboxEvent {
+  id: string;
+  topic: string;
+  payload: string;
+  status: OutboxStatus;
+  error: string | null;
+  attempts: number;
+  processedAt: Date | null;
+  createdAt: Date;
+  updatedAt: Date | null;
+}
+
+export enum OutboxStatus {
+  PENDING = 'PENDING',
+  PROCESSING = 'PROCESSING',
+  PROCESSED = 'PROCESSED',
+  FAILED = 'FAILED'
+}
+
+export type CreateOutboxEventData = Omit<OutboxEvent, 'id' | 'createdAt' | 'updatedAt' | 'processedAt' | 'attempts' | 'error'>;
+export type UpdateOutboxEventData = Partial<Omit<OutboxEvent, 'id' | 'createdAt' | 'updatedAt'>>;
+
+export interface OutboxRepository {
+    findManyBy<T extends Prisma.OutboxEventWhereInput>(query: T): Promise<OutboxEvent[] | null>;
+    create(user: CreateOutboxEventData): Promise<OutboxEvent>;
+    update(id: string, user: UpdateOutboxEventData): Promise<OutboxEvent>;
+}

apps/superchef/src/outbox/infrastructure/outbox.processor.ts

@@ -0,0 +1,42 @@
+import { Injectable, Inject } from '@nestjs/common';
+import { Cron } from '@nestjs/schedule';
+import { ClientProxy } from '@nestjs/microservices';
+import { lastValueFrom } from 'rxjs';
+import { PrismaOutboxRepository } from './outbox.service';
+import { OutboxStatus } from '../domain/outbox.interface';
+
+@Injectable()
+export class OutboxProcessor {
+    private isProcessing = false;
+
+    constructor(
+        private readonly outbox: PrismaOutboxRepository,
+        @Inject('EMAIL_SERVICE') private client: ClientProxy,
+    ) {}
+
+    @Cron('*/5 * * * * *')
+    async handle() {
+        if (this.isProcessing) return // Prevent overlapping if the previous one hasn't finished
+
+        this.isProcessing = true
+
+        try {
+            const messages = await this.outbox.findManyBy({ status: OutboxStatus.PENDING });
+            for (const message of messages) {
+                try {
+                    const {topic, payload } = message
+                    await lastValueFrom(this.client.emit(topic, payload))
+                    await this.outbox.update(message.id, { status: OutboxStatus.PROCESSED, processedAt: new Date() })
+                } catch (error) {
+                    await this.outbox.update(message.id, {
+                        error: error.message,
+                        attempts: message.attempts + 1,
+                        status: message.attempts >= 3 ? OutboxStatus.FAILED : OutboxStatus.PENDING
+                    })
+                }
+            }
+        } finally {
+            this.isProcessing = false
+        }
+    }
+}

apps/superchef/src/outbox/infrastructure/outbox.service.ts

@@ -0,0 +1,34 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService } from '../../prisma/prisma.service';
+import { CreateOutboxEventData, OutboxRepository, UpdateOutboxEventData} from '../domain/outbox.interface';
+import { Prisma } from 'generated/prisma/client';
+
+@Injectable()
+export class PrismaOutboxRepository implements OutboxRepository {
+  
+  constructor(
+    private prisma: PrismaService, 
+  ) {}
+
+  async create(data: CreateOutboxEventData) {
+    return this.prisma.outboxEvent.create({
+      data
+    })
+  }
+
+  async update(id: string, data: UpdateOutboxEventData) {
+    return this.prisma.outboxEvent.update({
+      where: { id },
+      data
+    })
+  }
+
+  async findManyBy<T extends Prisma.OutboxEventWhereInput>(query: T) {
+    return this.prisma.outboxEvent.findMany({
+      where: query,
+      take: 20,
+      orderBy: { createdAt: 'asc' }
+    })
+  }
+
+}

apps/superchef/src/outbox/outbox.module.ts

@@ -0,0 +1,25 @@
+import { Module } from '@nestjs/common';
+import { PrismaOutboxRepository } from './infrastructure/outbox.service';
+import { PrismaService } from '../prisma/prisma.service';
+import { ClientsModule, Transport } from '@nestjs/microservices';
+
+@Module({
+  imports: [
+    ClientsModule.register([
+      {
+        name: 'EMAIL_SERVICE',
+        transport: Transport.RMQ,
+        options: {
+          urls: [process.env.RABBITMQ_URL || 'amqp://localhost:5672'],
+          queue: 'email_queue',
+          queueOptions: {
+            durable: true,
+          },
+        },
+      },
+    ]),
+  ],
+  providers: [PrismaOutboxRepository, PrismaService],
+  exports: [PrismaOutboxRepository],
+})
+export class OutboxModule {}

apps/superchef/src/prisma/prisma.service.ts

@@ -88,6 +88,10 @@ export class PrismaService implements OnModuleInit {
     return this.prisma.refreshToken;
   }
 
+  get outboxEvent() {
+    return this.prisma.outboxEvent;
+  }
+
   async onModuleInit() {
     await this.prisma.$connect();
   }

compose.yml

@@ -1,4 +1,5 @@
 services:
+  
   postgres:
     image: postgres:15-alpine
     container_name: superchef_db

package-lock.json

@@ -37,6 +37,7 @@
     "@nestjs/microservices": "^11.0.0",
     "@nestjs/platform-express": "^11.0.1",
     "@nestjs/platform-socket.io": "^11.1.12",
+    "@nestjs/schedule": "^6.1.0",
     "@nestjs/swagger": "^11.2.5",
     "@nestjs/terminus": "^11.0.0",
     "@nestjs/throttler": "^6.3.0",

package.json

@@ -0,0 +1,19 @@
+-- CreateEnum
+CREATE TYPE "OutboxStatus" AS ENUM ('PENDING', 'PROCESSING', 'PROCESSED', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "outbox_event" (
+    "id" TEXT NOT NULL,
+    "topic" TEXT NOT NULL,
+    "payload" JSONB NOT NULL,
+    "status" "OutboxStatus" NOT NULL DEFAULT 'PENDING',
+    "error" TEXT,
+    "attempts" INTEGER NOT NULL DEFAULT 0,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "outbox_event_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "outbox_event_status_created_at_idx" ON "outbox_event"("status", "created_at");

prisma/migrations/20260201214154_init/migration.sql

@@ -131,3 +131,24 @@ model RefreshToken {
   @@unique([userId, deviceId])
   @@map("refresh_token")
 }
+
+enum OutboxStatus {
+  PENDING
+  PROCESSING
+  PROCESSED
+  FAILED
+}
+
+model OutboxEvent {
+  id          String   @id @default(uuid())
+  topic       String
+  payload     Json
+  status      OutboxStatus   @default(PENDING)
+  error       String?
+  attempts    Int       @default(0)
+  createdAt   DateTime @default(now()) @map("created_at")
+  updatedAt   DateTime @default(now()) @map("updated_at")
+
+  @@index([status, createdAt])
+  @@map("outbox_event")
+}