outbox pattern

Author: eiberham

README.md

@@ -14,12 +14,14 @@
 6. [Caching (Redis)](#caching-redis)
 7. [User Preferences](#user-preferences)
 8. [AI Recipe Assistant](#ai-recipe-assistant)
-9. [Project setup](#project-setup)
-10. [Environment variables](#environment-variables)
-11. [Compile and run the project](#compile-and-run-the-project)
-12. [Run tests](#run-tests)
-13. [Deployment](#deployment)
-14. [License](#license)
+9. [Analytics](#bar_chart-analytics)
+10.[Database Transactional Outbox Pattern](#database-transactional-outbox-pattern)
+11. [Project setup](#project-setup)
+12. [Environment variables](#environment-variables)
+13. [Compile and run the project](#compile-and-run-the-project)
+14. [Run tests](#run-tests)
+15. [Deployment](#deployment)
+16. [License](#license)
 
 
 ## SuperChef
@@ -265,6 +267,52 @@ Sample response:
 ]
 ```
 
+## Database Transactional Outbox Pattern
+
+When a user is created in the system two things happen:
+
+1. A new record is created in the users table respectively.
+2. A welcome email is dispatched to the user's inbox.
+
+This alone, could be a problem and lead to data inconsistencies, for example, it could be the case that the user is created successfully in our database but the message broker for whatever reason goes down precisely at that moment and the message is not delivered.
+
+How do we mitigate this ?
+
+By adding a transactional outbox pattern, this is how it works:
+
+1. There's a table holding the outbox events:
+
+  ```typescript
+  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")
+  }
+  ```
+
+  Creating the user is done by means of a transaction, this transaction involves updating both the user table and the outbox_event table.
+
+2. There's a cron job checking the outbox_event table every `5` seconds. 
+
+- If there's a `PENDING` event it will try to send it to the message broker so it can be processed. 
+- If the processing fails and the third attempt hasn't been reached the status remains `PENDING` and the attempts count bumps up.
+- If it's the third attempt and the processing fails the status is updated to `FAILED`.
+
+You can check the entire flow in the image below.
+
+<p align="center">
+  <img src="./outbox.png" alt="superchef" />
+</p>
+
+
 ## :credit_card: Stripe Integration
 
 Superchef integrates with Stripe for subscription management, allowing users to subscribe to a basic plan and access enhanced features.

apps/rabbitmq/src/controllers/rabbitmq.controller.ts

@@ -1,7 +1,7 @@
 import { Controller, Logger, UseFilters } from '@nestjs/common';
 import { EventPattern, Ctx, RmqContext } from '@nestjs/microservices';
 import type { ChannelModel, ConsumeMessage } from 'amqplib';
-import { UserRegisteredPayload } from '../domain/email.interface';
+import { Email, UserRegisteredPayload } from '../domain/email.interface';
 import { SendMailUsecase } from '../application/send-mail.usecase';
 import { RmqErrorFilter } from '../infrastructure/filters/rmq.error.filter';
 
@@ -18,15 +18,17 @@ export class RabbitmqController {
     const message = context.getMessage() as ConsumeMessage;
 
     let payload: UserRegisteredPayload;
+    let data : Email
     try {
       payload = JSON.parse(message.content.toString()) as UserRegisteredPayload;
+      data = JSON.parse(String(payload.data)) as Email
     } catch (err) {
       this.logger.error('Failed to parse message content', err as Error);
       channel.nack(message);
       return;
     }
 
-    await this.email.send(payload.data);
+    await this.email.send(data);
 
     channel.ack(message);
     this.logger.log('Email sent successfully');

apps/superchef/src/app.module.ts

@@ -17,9 +17,12 @@ import { CacheModule } from '@nestjs/cache-manager';
 import { KafkaModule } from './kafka.module';
 import { AnalyticsModule } from './analytics/analytics.module';
 import { HealthModule } from './health/health.module';
+import { ScheduleModule } from '@nestjs/schedule';
+import { OutboxModule } from './outbox/outbox.module';
 
 @Module({
   imports: [
+    ScheduleModule.forRoot(),
     KafkaModule,
     CacheModule.register({ isGlobal: true }),
     StripeModule.forRootAsync(),
@@ -44,6 +47,7 @@ import { HealthModule } from './health/health.module';
     CheckoutModule,
     AnalyticsModule,
     HealthModule,
+    OutboxModule,
   ],
   controllers: [AppController],
   providers: [

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

@@ -0,0 +1,28 @@
+import { Prisma } from "generated/prisma/client";
+
+export interface OutboxEvent {
+  id: string;
+  topic: string;
+  payload: string;
+  status: OutboxStatus;
+  error: string | null;
+  attempts: number;
+  createdAt: Date;
+  updatedAt: Date | null;
+}
+
+export enum OutboxStatus {
+  PENDING = 'PENDING',
+  PROCESSING = 'PROCESSING',
+  PROCESSED = 'PROCESSED',
+  FAILED = 'FAILED'
+}
+
+export type CreateOutboxEventData = Omit<OutboxEvent, 'id' | 'createdAt' | 'updatedAt' | '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,67 @@
+import { Injectable, Inject, Logger, OnModuleInit } from '@nestjs/common';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { ClientProxy } from '@nestjs/microservices';
+import { firstValueFrom, timeout, catchError, of } from 'rxjs';
+import { PrismaOutboxRepository } from './outbox.service';
+import { OutboxStatus } from '../domain/outbox.interface';
+
+@Injectable()
+export class OutboxProcessor implements OnModuleInit {
+    private isProcessing = false;
+    private readonly logger = new Logger(OutboxProcessor.name);
+
+    constructor(
+        private readonly outbox: PrismaOutboxRepository,
+        @Inject('EMAIL_SERVICE') private client: ClientProxy,
+    ) {}
+
+    async onModuleInit() {
+        await this.client.connect();
+    }
+
+    @Cron(CronExpression.EVERY_30_SECONDS)
+    async handle() {
+        if (this.isProcessing) return // Prevent overlapping if the previous one hasn't finished
+
+        this.isProcessing = true
+
+        try {
+            this.logger.debug('Processing outbox events...');
+            const messages = await this.outbox.findManyBy({ status: OutboxStatus.PENDING });
+            if (!messages?.length) {
+                this.logger.debug('No pending outbox events found.');
+                return;
+            }
+            for (const message of messages) {
+                try {
+                    const {topic, payload } = message
+
+                    // Use firstValueFrom to wait for the emit to complete
+                    // emit() completes when the message is sent to the broker
+                    await firstValueFrom(
+                        this.client.emit(topic, payload).pipe(
+                            timeout(5000),
+                            catchError((err) => {
+                                this.logger.error(`Emit failed for ${message.id}:`, err);
+                                throw err;
+                            })
+                        )
+                    );
+
+                    this.logger.log(`Successfully emitted event ${message.id} to topic: ${topic}`);
+                    await this.outbox.update(message.id, { status: OutboxStatus.PROCESSED })
+                } catch (error) {
+                    const errorMessage = error?.message || String(error) || 'Unknown error';
+                    this.logger.error(`Failed to process outbox event ${message.id}:`, JSON.stringify(errorMessage));
+                    await this.outbox.update(message.id, {
+                        error: errorMessage,
+                        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, OutboxEvent, 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) : Promise<OutboxEvent> {
+    return this.prisma.outboxEvent.create({
+      data
+    })
+  }
+
+  async update(id: string, data: UpdateOutboxEventData) : Promise<OutboxEvent> {
+    return this.prisma.outboxEvent.update({
+      where: { id },
+      data
+    })
+  }
+
+  async findManyBy<T extends Prisma.OutboxEventWhereInput>(query: T) : Promise<OutboxEvent[] | null> {
+    return this.prisma.outboxEvent.findMany({
+      where: query,
+      take: 20,
+      orderBy: { createdAt: 'asc' }
+    })
+  }
+
+}

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

@@ -0,0 +1,34 @@
+import { Module } from '@nestjs/common';
+import { PrismaOutboxRepository } from './infrastructure/outbox.service';
+import { PrismaService } from '../prisma/prisma.service';
+import { ClientsModule, Transport } from '@nestjs/microservices';
+import { OutboxProcessor } from './infrastructure/outbox.processor';
+
+@Module({
+  imports: [
+    ClientsModule.register([
+      {
+        name: 'EMAIL_SERVICE',
+        transport: Transport.RMQ,
+        options: {
+          urls: [process.env.RABBITMQ_URL || 'amqp://user:password@localhost:5672'],
+          queue: 'rabbitmq',
+          queueOptions: {
+            durable: false,
+            arguments: {
+              'prefetch-count': 1,
+              'x-message-ttl': 60000,
+              'x-max-priority': 10,
+              'x-dead-letter-exchange': 'dead_letter_exchange',
+              'x-dead-letter-routing-key': 'dead_letter',
+            },
+          },
+          noAck: true,
+        },
+      },
+    ]),
+  ],
+  providers: [PrismaOutboxRepository, PrismaService, OutboxProcessor],
+  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();
   }

apps/superchef/src/users/application/create-user.usecase.ts

@@ -35,18 +35,6 @@ export class CreateUserUsecase {
       user = await this.user.create(payload);
       this.logger.log(`User created: ${user.email}`);
 
-      const emailPayload = {
-        name: data.name,
-        to: data.email,
-        subject: 'Welcome to superchef!',
-        body: `
-                Thank you for registering at superchef. 
-                We are excited to have you on board! 
-                `,
-      };
-
-      this.client.emit('user_registered', emailPayload);
-
       return user;
     } catch (error) {
       this.logger.error(`Error creating user: ${data.email}`, error.stack);