feat: add notification microservices and transactional outbox

src/adapter-contracts.spec.ts

@@ -353,6 +353,11 @@ describe('controller and repository contracts', () => {
             expect(operations).toHaveLength(before);
             continue;
           }
+          if (`${Repository.name}.${method}` === 'FollowsRepository.runInTransaction') {
+            // Session lifecycle and fallback semantics need a session-aware model mock and
+            // are covered in follows.repository.spec.ts.
+            continue;
+          }
           try {
             await instance[method](...args);
           } catch (error) {

src/app.module.ts

@@ -31,6 +31,7 @@ import { MarketplaceModule } from './modules/marketplace/marketplace.module';
 import { MusicWorldModule } from './modules/music-world/music-world.module';
 import { NotificationsModule } from './modules/notifications/notifications.module';
 import { OutboxModule } from './modules/outbox/outbox.module';
+import { OrdersModule } from './modules/orders/orders.module';
 import { PostsModule } from './modules/posts/posts.module';
 import { ReportsModule } from './modules/reports/reports.module';
 import { SavesModule } from './modules/saves/saves.module';

src/app.service.spec.ts

@@ -14,7 +14,7 @@ describe('AppService readiness', () => {
       getActiveRequests: jest.fn().mockReturnValue(0),
     };
     return {
-      service: new AppService(connection as any, redis as any, storage as any, shutdown as any),
+      service: new AppService(connection as any, redis as any, storage as any, shutdown as any, { isReady: () => true } as any),
       shutdown,
     };
   };

src/app.service.ts

@@ -4,6 +4,7 @@ import { Connection } from 'mongoose';
 import { ManagedStorageService } from './infrastructure/storage/managed-storage.service';
 import { RedisService } from './infrastructure/redis/redis.service';
 import { ShutdownCoordinatorService } from './infrastructure/reliability/shutdown-coordinator.service';
+import { NotificationServiceClient } from './modules/notifications/notification-service.client';
 
 @Injectable()
 export class AppService {
@@ -12,6 +13,7 @@ export class AppService {
     private readonly redis: RedisService,
     private readonly storage: ManagedStorageService,
     private readonly shutdownCoordinator: ShutdownCoordinatorService,
+    private readonly notificationClient: NotificationServiceClient,
   ) {}
 
   getHealth(): { status: string; service: string } {
@@ -51,6 +53,10 @@ export class AppService {
       checks.redis = { status: 'up' };
     }
 
+    checks.rabbitmq = this.notificationClient.isReady()
+      ? { status: 'up' }
+      : { status: 'down', error: 'RabbitMQ connection is unavailable' };
+
     try {
       const health = await this.withTimeout(this.storage.getHealth(), 5000, 'Storage check timeout');
       const local = health.local as { readable?: boolean; writable?: boolean } | undefined;

src/config/validation.schema.ts

@@ -86,6 +86,37 @@ export const validationSchema = Joi.object({
   QUEUE_DEFAULT_BACKOFF_MS: Joi.number().min(100).max(600000).default(1000),
   QUEUE_REMOVE_ON_COMPLETE: Joi.boolean().truthy('true').falsy('false').default(true),
   QUEUE_WORKER_CONCURRENCY: Joi.number().min(1).max(100).default(5),
+  NOTIFICATION_SERVICE_ENABLED: Joi.boolean().truthy('true').falsy('false').default(false),
+  OUTBOX_DISPATCHER_ENABLED: Joi.boolean().truthy('true').falsy('false').default(false),
+  OUTBOX_POLL_INTERVAL_MS: Joi.number().integer().min(100).max(60000).default(1000),
+  OUTBOX_BATCH_SIZE: Joi.number().integer().min(1).max(500).default(25),
+  OUTBOX_MAX_ATTEMPTS: Joi.number().integer().min(1).max(100).default(8),
+  OUTBOX_LEASE_TIMEOUT_MS: Joi.number().integer().min(1000).max(3600000).default(60000),
+  OUTBOX_RETRY_BASE_MS: Joi.number().integer().min(100).max(600000).default(1000),
+  OUTBOX_RETRY_MAX_MS: Joi.number().integer().min(100).max(3600000).default(60000),
+  RABBITMQ_URL: Joi.string().uri().allow('').optional(),
+  RABBITMQ_HOST: Joi.string().default('127.0.0.1'),
+  RABBITMQ_PORT: Joi.number().integer().min(1).max(65535).default(5672),
+  RABBITMQ_USERNAME: Joi.string().allow('').optional(),
+  RABBITMQ_PASSWORD: Joi.string().allow('').optional(),
+  RABBITMQ_VHOST: Joi.string().default('/'),
+  RABBITMQ_TLS_ENABLED: Joi.boolean().truthy('true').falsy('false').default(false),
+  RABBITMQ_TLS_REQUIRED: Joi.boolean().truthy('true').falsy('false').default(true),
+  RABBITMQ_CA_CERT_BASE64: Joi.string().allow('').optional(),
+  RABBITMQ_CA_CERT_PATH: Joi.string().allow('').optional(),
+  RABBITMQ_CLIENT_CERT_BASE64: Joi.string().allow('').optional(),
+  RABBITMQ_CLIENT_CERT_PATH: Joi.string().allow('').optional(),
+  RABBITMQ_CLIENT_KEY_BASE64: Joi.string().allow('').optional(),
+  RABBITMQ_CLIENT_KEY_PATH: Joi.string().allow('').optional(),
+  RABBITMQ_TLS_REJECT_UNAUTHORIZED: Joi.boolean().truthy('true').falsy('false').default(true),
+  RABBITMQ_SERVERNAME: Joi.string().allow('').optional(),
+  RABBITMQ_HEARTBEAT_SECONDS: Joi.number().integer().min(1).max(120).default(10),
+  RABBITMQ_RECONNECT_SECONDS: Joi.number().integer().min(1).max(300).default(5),
+  RABBITMQ_CONNECTION_TIMEOUT_MS: Joi.number().integer().min(100).max(120000).default(10000),
+  NOTIFICATION_RPC_TIMEOUT_MS: Joi.number().integer().min(100).max(30000).default(3000),
+  NOTIFICATION_RPC_RETRIES: Joi.number().integer().min(0).max(5).default(1),
+  NOTIFICATION_CIRCUIT_FAILURE_THRESHOLD: Joi.number().integer().min(1).max(20).default(3),
+  NOTIFICATION_CIRCUIT_RESET_MS: Joi.number().integer().min(1000).max(300000).default(15000),
   STORAGE_PROVIDER: Joi.string().valid('local', 's3').default('local'),
   MEDIA_ACCESS_MODE: Joi.string().valid('direct', 'signed').default('direct'),
   SIGNED_URL_EXPIRES_SECONDS: Joi.number().integer().min(1).max(604800).default(86400),
@@ -156,5 +187,26 @@ export const validationSchema = Joi.object({
       custom: 'HTTP_SERVER_REQUEST_TIMEOUT_MS must be at least HTTP_HEADERS_TIMEOUT_MS',
     });
   }
+  if (Number(value.OUTBOX_RETRY_MAX_MS) < Number(value.OUTBOX_RETRY_BASE_MS)) {
+    return helpers.message({ custom: 'OUTBOX_RETRY_MAX_MS must be at least OUTBOX_RETRY_BASE_MS' });
+  }
+  if (value.NOTIFICATION_SERVICE_ENABLED !== value.OUTBOX_DISPATCHER_ENABLED) {
+    return helpers.message({
+      custom: 'Phase-1 Gateway activation requires NOTIFICATION_SERVICE_ENABLED and OUTBOX_DISPATCHER_ENABLED to have the same value',
+    });
+  }
+  if (value.NODE_ENV === 'production' && value.NOTIFICATION_SERVICE_ENABLED === true) {
+    const rawUrl = String(value.RABBITMQ_URL ?? '');
+    let url: URL | undefined;
+    try { if (rawUrl) url = new URL(rawUrl); } catch { return helpers.message({ custom: 'RABBITMQ_URL is invalid' }); }
+    const username = url?.username ? decodeURIComponent(url.username) : String(value.RABBITMQ_USERNAME ?? '');
+    const password = url?.password ? decodeURIComponent(url.password) : String(value.RABBITMQ_PASSWORD ?? '');
+    if (!username || !password || username === 'guest' || password === 'guest') {
+      return helpers.message({ custom: 'Production notification routing requires non-guest RabbitMQ credentials' });
+    }
+    if (value.RABBITMQ_TLS_REQUIRED === true && url?.protocol !== 'amqps:' && value.RABBITMQ_TLS_ENABLED !== true) {
+      return helpers.message({ custom: 'Production notification routing requires RabbitMQ TLS/amqps' });
+    }
+  }
   return value;
-}, 'storage access mode validation');
+}, 'cross-field configuration validation');

src/modules/follows/follows.repository.spec.ts

@@ -0,0 +1,33 @@
+import { FollowsRepository } from './follows.repository';
+
+describe('FollowsRepository transaction boundary', () => {
+  function setup(withTransaction: jest.Mock) {
+    const session = { withTransaction, endSession: jest.fn().mockResolvedValue(undefined) };
+    const followModel = { db: { startSession: jest.fn().mockResolvedValue(session) } };
+    const repository = new FollowsRepository(followModel as never, {} as never);
+    return { repository, session };
+  }
+
+  it('returns only after the transaction callback commits and always closes the session', async () => {
+    const ctx = setup(jest.fn(async (callback: () => Promise<void>) => callback()));
+    const operation = jest.fn().mockResolvedValue('committed');
+    await expect(ctx.repository.runInTransaction(operation, true)).resolves.toBe('committed');
+    expect(operation).toHaveBeenCalledWith(ctx.session);
+    expect(ctx.session.endSession).toHaveBeenCalled();
+  });
+
+  it('propagates transaction failures in microservices mode without an unsafe fallback', async () => {
+    const ctx = setup(jest.fn(async (callback: () => Promise<void>) => callback()));
+    const operation = jest.fn().mockRejectedValue(new Error('outbox insert failed'));
+    await expect(ctx.repository.runInTransaction(operation, true)).rejects.toThrow('outbox insert failed');
+    expect(operation).toHaveBeenCalledTimes(1);
+    expect(ctx.session.endSession).toHaveBeenCalled();
+  });
+
+  it('allows only the documented legacy fallback when standalone Mongo rejects transactions', async () => {
+    const ctx = setup(jest.fn().mockRejectedValue(new Error('Transaction numbers are only allowed on a replica set member or mongos')));
+    const operation = jest.fn().mockResolvedValue('legacy');
+    await expect(ctx.repository.runInTransaction(operation, false)).resolves.toBe('legacy');
+    expect(operation).toHaveBeenCalledWith(undefined);
+  });
+});

src/modules/follows/follows.repository.ts

@@ -12,12 +12,17 @@ export class FollowsRepository {
     private readonly followRequestModel: Model<FollowRequestDocument>,
   ) {}
 
-  async findOne(followerId: string, followingId: string): Promise<FollowDocument | null> {
+  async findOne(
+    followerId: string,
+    followingId: string,
+    session?: ClientSession,
+  ): Promise<FollowDocument | null> {
     return this.followModel
       .findOne({
         followerId: new Types.ObjectId(followerId),
         followingId: new Types.ObjectId(followingId),
       })
+      .session(session ?? null)
       .exec();
   }
 
@@ -109,6 +114,27 @@ export class FollowsRepository {
       .exec();
   }
 
+  async runInTransaction<T>(
+    operation: (session?: ClientSession) => Promise<T>,
+    required: boolean,
+  ): Promise<T> {
+    const session = await this.followModel.db.startSession();
+    try {
+      let result: T | undefined;
+      try {
+        await session.withTransaction(async () => {
+          result = await operation(session);
+        });
+        return result as T;
+      } catch (error) {
+        if (required || !this.isTransactionUnsupported(error)) throw error;
+        return operation(undefined);
+      }
+    } finally {
+      await session.endSession();
+    }
+  }
+
   async deletePendingRequest(requesterId: string, targetUserId: string): Promise<void> {
     await this.followRequestModel
       .deleteOne({
@@ -162,16 +188,33 @@ export class FollowsRepository {
     requestId: string,
     targetUserId: string,
     status: 'approved' | 'rejected',
+    session?: ClientSession,
   ): Promise<FollowRequestDocument | null> {
     return this.followRequestModel
       .findOneAndUpdate(
         { _id: new Types.ObjectId(requestId), targetUserId: new Types.ObjectId(targetUserId), status: 'pending' },
         { status },
-        { new: true },
+        { new: true, session },
       )
       .exec();
   }
 
+  async findRequestByIdForTarget(
+    requestId: string,
+    targetUserId: string,
+    session?: ClientSession,
+  ): Promise<FollowRequestDocument | null> {
+    return this.followRequestModel
+      .findOne({ _id: new Types.ObjectId(requestId), targetUserId: new Types.ObjectId(targetUserId) })
+      .session(session ?? null)
+      .exec();
+  }
+
+  private isTransactionUnsupported(error: unknown): boolean {
+    const message = error instanceof Error ? error.message : String(error);
+    return /Transaction numbers are only allowed|replica set member or mongos/i.test(message);
+  }
+
   async findPendingRequestsForTarget(
     targetUserId: string,
     skip: number,

src/modules/follows/follows.service.spec.ts

@@ -23,6 +23,8 @@ describe('FollowsService', () => {
       findPendingRequestsForTarget: jest.fn().mockResolvedValue([]),
       countPendingRequestsForTarget: jest.fn().mockResolvedValue(0),
       updateRequestStatus: jest.fn().mockResolvedValue(null),
+      findRequestByIdForTarget: jest.fn().mockResolvedValue(null),
+      runInTransaction: jest.fn().mockImplementation(async (operation: (session?: unknown) => Promise<unknown>) => operation(undefined)),
       ...overrides.followsRepository,
     };
     const usersRepository = {
@@ -38,6 +40,8 @@ describe('FollowsService', () => {
       })),
       setFollowingCount: jest.fn().mockResolvedValue(undefined),
       setFollowersCount: jest.fn().mockResolvedValue(undefined),
+      incrementFollowingCount: jest.fn().mockResolvedValue(undefined),
+      incrementFollowersCount: jest.fn().mockResolvedValue(undefined),
       findManyByIds: jest.fn().mockResolvedValue([]),
       findSuggestionCandidates: jest.fn().mockResolvedValue([]),
       ...overrides.usersRepository,
@@ -45,6 +49,8 @@ describe('FollowsService', () => {
     const outboxService = {
       enqueueFollowNotification: jest.fn().mockResolvedValue(undefined),
       enqueueFollowRequestApprovedNotification: jest.fn().mockResolvedValue(undefined),
+      createFollowRequestApprovedEvent: jest.fn().mockResolvedValue(`follow.request.approved:${requestId}`),
+      enqueuePersistedEvent: jest.fn().mockResolvedValue(undefined),
       ...overrides.outboxService,
     };
     const feedVersionService = {
@@ -413,12 +419,13 @@ describe('FollowsService', () => {
       following: true,
       requesterId: targetUserId,
     });
-    expect(ctx.followsRepository.create).toHaveBeenCalledWith(targetUserId, currentUserId);
+    expect(ctx.followsRepository.create).toHaveBeenCalledWith(targetUserId, currentUserId, undefined);
     expect(ctx.feedVersionService.bumpUserVersion).toHaveBeenCalledWith(targetUserId);
-    expect(ctx.outboxService.enqueueFollowRequestApprovedNotification).toHaveBeenCalledWith(
+    expect(ctx.outboxService.createFollowRequestApprovedEvent).toHaveBeenCalledWith(
       currentUserId,
       targetUserId,
       requestId,
+      undefined,
     );
 
     const existing = setup({
@@ -429,7 +436,7 @@ describe('FollowsService', () => {
     });
     await existing.service.approveRequest(currentUserId, requestId);
     expect(existing.followsRepository.create).not.toHaveBeenCalled();
-    expect(existing.outboxService.enqueueFollowRequestApprovedNotification).toHaveBeenCalledTimes(1);
+    expect(existing.outboxService.createFollowRequestApprovedEvent).toHaveBeenCalledTimes(1);
   });
 
   it('validates approve and reject requests and handles missing records', async () => {
@@ -439,7 +446,7 @@ describe('FollowsService', () => {
     await expect(setup().service.approveRequest(currentUserId, requestId)).rejects.toThrow(
       'Follow request not found',
     );
-    expect(setup().outboxService.enqueueFollowRequestApprovedNotification).not.toHaveBeenCalled();
+    expect(setup().outboxService.createFollowRequestApprovedEvent).not.toHaveBeenCalled();
     await expect(setup().service.rejectRequest(currentUserId, 'bad')).rejects.toThrow(
       'Invalid follow request id',
     );
@@ -458,6 +465,29 @@ describe('FollowsService', () => {
     });
   });
 
+  it('keeps duplicate approvals idempotent and propagates an outbox transaction failure', async () => {
+    const duplicate = setup({
+      followsRepository: {
+        updateRequestStatus: jest.fn().mockResolvedValue(null),
+        findRequestByIdForTarget: jest.fn().mockResolvedValue({
+          status: 'approved', requesterId: { toString: () => targetUserId },
+        }),
+      },
+    });
+    await expect(duplicate.service.approveRequest(currentUserId, requestId)).resolves.toEqual({
+      approved: true, following: true, requesterId: targetUserId,
+    });
+    expect(duplicate.followsRepository.create).not.toHaveBeenCalled();
+    expect(duplicate.outboxService.createFollowRequestApprovedEvent).not.toHaveBeenCalled();
+
+    const failed = setup({
+      followsRepository: { updateRequestStatus: jest.fn().mockResolvedValue({ requesterId: { toString: () => targetUserId } }) },
+      outboxService: { createFollowRequestApprovedEvent: jest.fn().mockRejectedValue(new Error('outbox insert failed')) },
+    });
+    await expect(failed.service.approveRequest(currentUserId, requestId)).rejects.toThrow('outbox insert failed');
+    expect(failed.feedVersionService.bumpUserVersion).not.toHaveBeenCalled();
+  });
+
   it('ranks and explains personalized suggestions with stable pagination', async () => {
     jest.spyOn(Math, 'random').mockReturnValue(0.25);
     const current = {

src/modules/follows/follows.service.ts

@@ -251,24 +251,61 @@ export class FollowsService {
       throw new BadRequestException('Invalid follow request id');
     }
 
-    const request = await this.followsRepository.updateRequestStatus(requestId, currentUserId, 'approved');
-    if (!request) {
-      throw new NotFoundException('Follow request not found');
+    const microservicesEnabled = process.env.NOTIFICATION_SERVICE_ENABLED === 'true'
+      || process.env.OUTBOX_DISPATCHER_ENABLED === 'true';
+    const transaction = await this.followsRepository.runInTransaction(async (session) => {
+      const request = await this.followsRepository.updateRequestStatus(
+        requestId,
+        currentUserId,
+        'approved',
+        session,
+      );
+      if (!request) {
+        const existingRequest = await this.followsRepository.findRequestByIdForTarget(
+          requestId,
+          currentUserId,
+          session,
+        );
+        if (existingRequest?.status === 'approved') {
+          return {
+            requesterId: existingRequest.requesterId.toString(),
+            followCreated: false,
+            eventId: null,
+          };
+        }
+        throw new NotFoundException('Follow request not found');
+      }
+
+      const requesterId = request.requesterId.toString();
+      const existing = await this.followsRepository.findOne(requesterId, currentUserId, session);
+      let followCreated = false;
+      if (!existing) {
+        await this.followsRepository.create(requesterId, currentUserId, session);
+        // MongoDB transactions do not support parallel operations on one session.
+        await this.usersRepository.incrementFollowingCount(requesterId, 1, session);
+        await this.usersRepository.incrementFollowersCount(currentUserId, 1, session);
+        followCreated = true;
+      }
+
+      const eventId = await this.outboxService.createFollowRequestApprovedEvent(
+        currentUserId,
+        requesterId,
+        requestId,
+        session,
+      );
+      return { requesterId, followCreated, eventId };
+    }, microservicesEnabled);
+
+    if (transaction.followCreated) await this.feedVersionService.bumpUserVersion(transaction.requesterId);
+    if (!microservicesEnabled && transaction.eventId) {
+      try {
+        await this.outboxService.enqueuePersistedEvent(transaction.eventId);
+      } catch (error) {
+        this.logger.warn(`Legacy outbox enqueue failed; event remains pending: ${error instanceof Error ? error.message : 'unknown error'}`);
+      }
     }
 
-    const requesterId = request.requesterId.toString();
-    const existing = await this.followsRepository.findOne(requesterId, currentUserId);
-    if (!existing) {
-      await this.followsRepository.create(requesterId, currentUserId);
-      await this.syncFollowCounts(requesterId, currentUserId);
-      await this.feedVersionService.bumpUserVersion(requesterId);
-    }
-
-    await this.enqueueFollowRequestApprovedNotification(
-      currentUserId,
-      requesterId,
-      requestId,
-    );
+    const requesterId = transaction.requesterId;
 
     return { approved: true, following: true, requesterId };
   }
@@ -467,26 +504,6 @@ export class FollowsService {
     }
   }
 
-  private async enqueueFollowRequestApprovedNotification(
-    actorId: string,
-    recipientId: string,
-    requestId: string,
-  ): Promise<void> {
-    try {
-      await this.outboxService.enqueueFollowRequestApprovedNotification(
-        actorId,
-        recipientId,
-        requestId,
-      );
-    } catch (error) {
-      this.logger.warn(
-        `Follow request approval notification failed for actor=${actorId} recipient=${recipientId}: ${
-          error instanceof Error ? error.message : 'unknown error'
-        }`,
-      );
-    }
-  }
-
   private buildFollowActionResponse(
     message: string,
     isFollowing: boolean,

src/modules/notifications/dto/create-notification.dto.ts

@@ -2,6 +2,11 @@ import { IsEnum, IsMongoId, IsObject, IsOptional, IsString, MaxLength } from 'cl
 import { NOTIFICATION_TYPES, NotificationType } from '../schemas/notification.schema';
 
 export class CreateNotificationDto {
+  @IsOptional()
+  @IsString()
+  @MaxLength(128)
+  eventId?: string;
+
   @IsMongoId()
   recipientId!: string;
 

src/modules/notifications/notification-contract.spec.ts

@@ -0,0 +1,92 @@
+import { NOTIFICATION_RPC } from '@app/contracts';
+import { FOLLOW_REQUEST_APPROVED_EVENT, FollowRequestApprovedEvent } from '@app/events';
+import { ConfigService } from '@nestjs/config';
+import { of } from 'rxjs';
+import { NotificationConsumer } from '../../../apps/notification-service/src/notification.consumer';
+import { NotificationServiceClient } from './notification-service.client';
+import { NotificationsGateway } from './notifications.gateway';
+
+const remoteConfig = () => ({
+  get: jest.fn((key: string, fallback?: string) => ({
+    NOTIFICATION_SERVICE_ENABLED: 'true',
+    RABBITMQ_URL: 'amqp://rabbit',
+    NOTIFICATION_RPC_TIMEOUT_MS: '1000',
+    NOTIFICATION_RPC_RETRIES: '0',
+  })[key] ?? fallback),
+}) as unknown as ConfigService;
+
+describe('Gateway to notification-service contract', () => {
+  it('keeps the gateway result identical while using the shared RMQ pattern', async () => {
+    const expected = { items: [{ _id: 'notification-1' }], total: 1, unreadCount: 1 };
+    const service = new NotificationServiceClient(
+      remoteConfig(),
+      { getMine: jest.fn() } as never,
+      { emitCreated: jest.fn(), emitUnreadCount: jest.fn() } as never,
+    );
+    const transport = { send: jest.fn().mockReturnValue(of({ result: expected })), close: jest.fn() };
+    (service as unknown as { client: unknown }).client = transport;
+
+    await expect(service.getMine('user-1', { page: 1 } as never, 'request-1')).resolves.toBe(expected);
+    expect(transport.send).toHaveBeenCalledWith(
+      NOTIFICATION_RPC.list,
+      expect.objectContaining({ data: { recipientId: 'user-1', query: { page: 1 } } }),
+    );
+    await service.onModuleDestroy();
+  });
+
+  it('uses only the local service and never sends to RabbitMQ when the flag is false', async () => {
+    const expected = { items: [], total: 0, unreadCount: 0 };
+    const local = { getMine: jest.fn().mockResolvedValue(expected) };
+    const service = new NotificationServiceClient(
+      { get: jest.fn((_key: string, fallback?: string) => fallback) } as unknown as ConfigService,
+      local as never,
+      { emitCreated: jest.fn(), emitUnreadCount: jest.fn() } as never,
+    );
+    const transport = { send: jest.fn(), connect: jest.fn(), close: jest.fn() };
+    (service as unknown as { client: unknown }).client = transport;
+
+    await expect(service.getMine('user-1', {} as never)).resolves.toBe(expected);
+    expect(local.getMine).toHaveBeenCalledTimes(1);
+    expect(transport.send).not.toHaveBeenCalled();
+    expect(transport.connect).not.toHaveBeenCalled();
+    await service.onModuleDestroy();
+  });
+
+  it('delivers follow approval from the consumer to each compatible Socket.IO event once', async () => {
+    jest.spyOn(console, 'log').mockImplementation(() => undefined);
+    const event: FollowRequestApprovedEvent = {
+      eventId: 'event-socket-1',
+      eventType: FOLLOW_REQUEST_APPROVED_EVENT,
+      occurredAt: '2026-08-03T10:00:00.000Z',
+      payload: { requestId: 'request-1', actorId: 'actor-1', recipientId: 'recipient-1' },
+    };
+    const notification = { _id: 'notification-1', type: 'follow_request_approved' };
+    const consumer = new NotificationConsumer({
+      createFollowRequestApprovedNotification: jest.fn().mockResolvedValue(notification),
+      getUnreadCounts: jest.fn().mockResolvedValue({ total: 1, followRequests: 1 }),
+    } as never);
+    const channel = { ack: jest.fn(), nack: jest.fn() };
+    const response = await consumer.followRequestApproved(event, {
+      getChannelRef: () => channel,
+      getMessage: () => ({ properties: { headers: {} } }),
+    } as never);
+
+    const gateway = new NotificationsGateway({} as never, {} as never, {} as never);
+    const roomEmitter = { emit: jest.fn() };
+    (gateway as unknown as { server: unknown }).server = {
+      to: jest.fn().mockReturnValue(roomEmitter),
+    };
+    const client = new NotificationServiceClient(remoteConfig(), {} as never, gateway);
+    const transport = { send: jest.fn().mockReturnValue(of(response)), close: jest.fn() };
+    (client as unknown as { client: unknown }).client = transport;
+
+    await client.publishFollowRequestApproved(event);
+
+    expect(roomEmitter.emit).toHaveBeenCalledWith('notification_created', notification);
+    expect(roomEmitter.emit).toHaveBeenCalledWith('notification:new', notification);
+    expect(roomEmitter.emit.mock.calls.filter(([name]) => name === 'notification_created')).toHaveLength(1);
+    expect(roomEmitter.emit.mock.calls.filter(([name]) => name === 'notification:new')).toHaveLength(1);
+    await client.onModuleDestroy();
+    jest.restoreAllMocks();
+  });
+});

src/modules/notifications/notification-idempotency.spec.ts

@@ -0,0 +1,38 @@
+import { NotificationsService } from './notifications.service';
+
+describe('Notification event idempotency', () => {
+  const dto = {
+    eventId: 'event-duplicate',
+    actorId: '507f1f77bcf86cd799439011',
+    recipientId: '507f191e810c19729de860ea',
+    requestId: '507f1f77bcf86cd799439012',
+  };
+
+  it('returns the existing record when the same event is delivered twice', async () => {
+    const existing = { _id: 'notification-1' };
+    const repository = {
+      findByEventId: jest.fn().mockResolvedValue(existing),
+      create: jest.fn(),
+    };
+    const gateway = { emitCreated: jest.fn() };
+    const service = new NotificationsService(repository as never, gateway as never);
+
+    await expect(service.createFollowRequestApprovedNotification(dto)).resolves.toBe(existing);
+    await expect(service.createFollowRequestApprovedNotification(dto)).resolves.toBe(existing);
+    expect(repository.findByEventId).toHaveBeenCalledTimes(2);
+    expect(repository.create).not.toHaveBeenCalled();
+    expect(gateway.emitCreated).not.toHaveBeenCalled();
+  });
+
+  it('recovers from a concurrent duplicate-key insert', async () => {
+    const existing = { _id: 'notification-1' };
+    const repository = {
+      findByEventId: jest.fn().mockResolvedValueOnce(null).mockResolvedValueOnce(existing),
+      create: jest.fn().mockRejectedValue({ code: 11000 }),
+    };
+    const service = new NotificationsService(repository as never, { emitCreated: jest.fn() } as never);
+
+    await expect(service.createFollowRequestApprovedNotification(dto)).resolves.toBe(existing);
+    expect(repository.create).toHaveBeenCalledTimes(1);
+  });
+});

src/modules/notifications/notification-service.client.ts

@@ -0,0 +1,121 @@
+import { resolveRabbitMqConfig } from '../../../libs/common/src';
+import { NOTIFICATION_RMQ_DLQ, NOTIFICATION_RMQ_QUEUE, NOTIFICATION_RPC, NotificationRpcResponse } from '../../../libs/contracts/src';
+import { FollowRequestApprovedEvent } from '../../../libs/events/src';
+import { Injectable, OnApplicationBootstrap, OnModuleDestroy, ServiceUnavailableException } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { ClientProxy, ClientProxyFactory, RmqRecordBuilder, Transport } from '@nestjs/microservices';
+import { randomUUID } from 'crypto';
+import { firstValueFrom, retry, timeout, timer } from 'rxjs';
+import { NotificationCategory, NotificationQueryDto } from './dto/notification-query.dto';
+import { NotificationsGateway } from './notifications.gateway';
+import { NotificationsService } from './notifications.service';
+
+@Injectable()
+export class NotificationServiceClient implements OnApplicationBootstrap, OnModuleDestroy {
+  private client?: ClientProxy;
+  private rabbitReady = false;
+  private consecutiveFailures = 0;
+  private circuitOpenUntil = 0;
+
+  constructor(private readonly config: ConfigService, private readonly local: NotificationsService, private readonly gateway: NotificationsGateway) {
+    if (this.isRemoteEnabled()) this.getClient();
+  }
+
+  isRemoteEnabled(): boolean {
+    return String(this.config.get('NOTIFICATION_SERVICE_ENABLED') ?? 'false').toLowerCase() === 'true';
+  }
+
+  isReady(): boolean {
+    return !this.isRemoteEnabled() || this.rabbitReady;
+  }
+
+  async onApplicationBootstrap(): Promise<void> {
+    if (!this.isRemoteEnabled()) return;
+    try { await this.getClient().connect(); }
+    catch { this.rabbitReady = false; }
+  }
+
+  async getMine(recipientId: string, query: NotificationQueryDto, requestId?: string) {
+    if (!this.isRemoteEnabled()) return this.local.getMine(recipientId, query);
+    return this.call(NOTIFICATION_RPC.list, { recipientId, query }, requestId);
+  }
+
+  async getUnreadCount(recipientId: string, category?: NotificationCategory, requestId?: string) {
+    if (!this.isRemoteEnabled()) return this.local.getUnreadCountByCategory(recipientId, category);
+    return this.call(NOTIFICATION_RPC.unreadCount, { recipientId, category }, requestId);
+  }
+
+  async markRead(recipientId: string, notificationId: string, requestId?: string) {
+    if (!this.isRemoteEnabled()) return this.local.markRead(recipientId, notificationId);
+    return this.call(NOTIFICATION_RPC.markRead, { recipientId, notificationId }, requestId);
+  }
+
+  async markAllRead(recipientId: string, requestId?: string) {
+    if (!this.isRemoteEnabled()) return this.local.markAllRead(recipientId);
+    return this.call(NOTIFICATION_RPC.markAllRead, { recipientId }, requestId);
+  }
+
+  async publishFollowRequestApproved(event: FollowRequestApprovedEvent): Promise<void> {
+    if (!this.isRemoteEnabled()) {
+      await this.local.createFollowRequestApprovedNotification({ ...event.payload, eventId: event.eventId });
+      return;
+    }
+    await this.call(event.eventType, event, event.payload.requestId, event.eventId);
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.client) await this.client.close();
+  }
+
+  private getClient(): ClientProxy {
+    if (this.client) return this.client;
+    const rabbit = resolveRabbitMqConfig(
+      { get: (key) => {
+        const value = this.config.get(key);
+        return value === undefined || value === null ? undefined : String(value);
+      } },
+      'api-gateway',
+      true,
+    );
+    this.client = ClientProxyFactory.create({
+      transport: Transport.RMQ,
+      options: {
+        urls: [rabbit.url], socketOptions: rabbit.socketOptions,
+        queue: NOTIFICATION_RMQ_QUEUE,
+        queueOptions: { durable: true, arguments: { 'x-dead-letter-exchange': '', 'x-dead-letter-routing-key': NOTIFICATION_RMQ_DLQ } },
+      },
+    });
+    this.client.status.subscribe((status) => {
+      this.rabbitReady = status === 'connected' || status === 'unblocked';
+    });
+    return this.client;
+  }
+
+  private async call<T>(pattern: string, payload: unknown, requestId?: string, correlationId?: string): Promise<T> {
+    requestId ??= randomUUID();
+    correlationId ??= randomUUID();
+    if (Date.now() < this.circuitOpenUntil) throw new ServiceUnavailableException('Notification service circuit is open');
+    const record = new RmqRecordBuilder(payload).setOptions({ messageId: correlationId, persistent: true, headers: { correlationId, requestId } }).build();
+    try {
+      const response = await firstValueFrom(this.getClient().send<NotificationRpcResponse<T>>(pattern, record).pipe(
+        timeout(Number(this.config.get<string>('NOTIFICATION_RPC_TIMEOUT_MS') ?? 3000)),
+        retry({ count: Number(this.config.get<string>('NOTIFICATION_RPC_RETRIES') ?? 1), delay: (_error, count) => timer(100 * count) }),
+      ));
+      this.consecutiveFailures = 0;
+      if (response.realtime) {
+        const realtime = response.realtime;
+        if (realtime.notification) this.gateway.emitCreated(realtime.recipientId, realtime.notification, realtime.unreadCount, realtime.unreadCounts);
+        else this.gateway.emitUnreadCount(realtime.recipientId, realtime.unreadCount, realtime.unreadCounts);
+      }
+      return response.result;
+    } catch (error) {
+      this.consecutiveFailures += 1;
+      const threshold = Number(this.config.get<string>('NOTIFICATION_CIRCUIT_FAILURE_THRESHOLD') ?? 3);
+      if (this.consecutiveFailures >= threshold) {
+        this.circuitOpenUntil = Date.now() + Number(this.config.get<string>('NOTIFICATION_CIRCUIT_RESET_MS') ?? 15_000);
+        this.consecutiveFailures = 0;
+      }
+      throw new ServiceUnavailableException('Notification service is unavailable', { cause: error });
+    }
+  }
+}

src/modules/notifications/notifications.controller.ts

@@ -1,4 +1,5 @@
-import { Controller, Get, Param, Patch, Query, UseGuards } from '@nestjs/common';
+import { Controller, Get, Param, Patch, Query, Req, UseGuards } from '@nestjs/common';
+import { Request } from 'express';
 import { ApiBearerAuth, ApiTags } from '@nestjs/swagger';
 import { CurrentUser } from '../../common/decorators/current-user.decorator';
 import { SuperAdminPermissions } from '../../common/decorators/superadmin-permissions.decorator';
@@ -12,12 +13,16 @@ import {
   NotificationUnreadCountQueryDto,
 } from './dto/notification-query.dto';
 import { NotificationsService } from './notifications.service';
+import { NotificationServiceClient } from './notification-service.client';
 
 @ApiTags('Notifications')
 @ApiBearerAuth()
 @Controller('notifications')
 export class NotificationsController {
-  constructor(private readonly notificationsService: NotificationsService) {}
+  constructor(
+    private readonly notificationsService: NotificationsService,
+    private readonly notificationClient: NotificationServiceClient,
+  ) {}
 
   @ApiBearerAuth()
   @UseGuards(SuperAdminJwtAuthGuard, SuperAdminPermissionsGuard)
@@ -29,8 +34,12 @@ export class NotificationsController {
 
   @UseGuards(JwtAuthGuard)
   @Get()
-  async getMine(@CurrentUser() user: JwtPayload, @Query() query: NotificationQueryDto) {
-    return this.notificationsService.getMine(user.sub, query);
+  async getMine(
+    @CurrentUser() user: JwtPayload,
+    @Query() query: NotificationQueryDto,
+    @Req() request: Request,
+  ) {
+    return this.notificationClient.getMine(user.sub, query, this.requestId(request));
   }
 
   @UseGuards(JwtAuthGuard)
@@ -38,8 +47,9 @@ export class NotificationsController {
   async getUnreadCount(
     @CurrentUser() user: JwtPayload,
     @Query() query: NotificationUnreadCountQueryDto,
+    @Req() request: Request,
   ) {
-    return this.notificationsService.getUnreadCountByCategory(user.sub, query.category);
+    return this.notificationClient.getUnreadCount(user.sub, query.category, this.requestId(request));
   }
 
   @UseGuards(JwtAuthGuard)
@@ -50,13 +60,22 @@ export class NotificationsController {
 
   @UseGuards(JwtAuthGuard)
   @Patch('read-all')
-  async markAllRead(@CurrentUser() user: JwtPayload) {
-    return this.notificationsService.markAllRead(user.sub);
+  async markAllRead(@CurrentUser() user: JwtPayload, @Req() request: Request) {
+    return this.notificationClient.markAllRead(user.sub, this.requestId(request));
   }
 
   @UseGuards(JwtAuthGuard)
   @Patch(':id/read')
-  async markRead(@CurrentUser() user: JwtPayload, @Param('id') notificationId: string) {
-    return this.notificationsService.markRead(user.sub, notificationId);
+  async markRead(
+    @CurrentUser() user: JwtPayload,
+    @Param('id') notificationId: string,
+    @Req() request: Request,
+  ) {
+    return this.notificationClient.markRead(user.sub, notificationId, this.requestId(request));
+  }
+
+  private requestId(request: Request): string | undefined {
+    const value = request.headers['x-request-id'];
+    return typeof value === 'string' ? value : undefined;
   }
 }

src/modules/notifications/notifications.module.ts

@@ -8,6 +8,7 @@ import { NotificationsService } from './notifications.service';
 import { NotificationsRepository } from './notifications.repository';
 import { Notification, NotificationSchema } from './schemas/notification.schema';
 import { UsersModule } from '../users/users.module';
+import { NotificationServiceClient } from './notification-service.client';
 
 @Module({
   imports: [
@@ -17,7 +18,12 @@ import { UsersModule } from '../users/users.module';
     MongooseModule.forFeature([{ name: Notification.name, schema: NotificationSchema }]),
   ],
   controllers: [NotificationsController],
-  providers: [NotificationsService, NotificationsRepository, NotificationsGateway],
-  exports: [NotificationsService],
+  providers: [
+    NotificationsService,
+    NotificationsRepository,
+    NotificationsGateway,
+    NotificationServiceClient,
+  ],
+  exports: [NotificationsService, NotificationsGateway, NotificationServiceClient],
 })
 export class NotificationsModule {}

src/modules/notifications/notifications.repository.ts

@@ -78,6 +78,13 @@ export class NotificationsRepository {
       .exec();
   }
 
+  async findByEventId(eventId: string): Promise<NotificationDocument | null> {
+    return this.notificationModel
+      .findOne({ eventId })
+      .populate({ path: 'actorId', select: 'name username stageName avatar isVerified isDisabled' })
+      .exec();
+  }
+
   async countUnreadByFilter(
     recipientId: string,
     filter: FilterQuery<NotificationDocument> = {},

src/modules/notifications/notifications.service.ts

@@ -45,11 +45,19 @@ export class NotificationsService {
       return null;
     }
 
+    if (dto.eventId) {
+      const existing = await this.notificationsRepository.findByEventId(dto.eventId);
+      if (existing) return existing;
+    }
+
     const resourceType = (dto.resourceType ?? this.resolveResourceType(dto.type)).trim();
     const deepLink = (dto.deepLink ?? this.buildDeepLink(dto.type, dto.referenceId, resourceType)).trim();
     const title = (dto.title ?? this.buildTitle(dto.type)).trim();
     const previewText = (dto.previewText ?? '').trim();
-    const notification = await this.notificationsRepository.create({
+    let notification;
+    try {
+      notification = await this.notificationsRepository.create({
+      eventId: dto.eventId,
       recipientId: new Types.ObjectId(dto.recipientId),
       actorId: new Types.ObjectId(dto.actorId),
       type: dto.type,
@@ -61,7 +69,14 @@ export class NotificationsService {
       metadata: dto.metadata ?? {},
       read: false,
       readAt: null,
-    });
+      });
+    } catch (error) {
+      if (dto.eventId && this.isDuplicateKeyError(error)) {
+        const existing = await this.notificationsRepository.findByEventId(dto.eventId);
+        if (existing) return existing;
+      }
+      throw error;
+    }
 
     const unreadCounts = await this.getUnreadCounts(dto.recipientId);
     const unreadCount = unreadCounts.total;
@@ -194,11 +209,13 @@ export class NotificationsService {
     actorId: string;
     recipientId: string;
     requestId: string;
+    eventId?: string;
   }) {
     const actorId = String(input.actorId);
     const recipientId = String(input.recipientId);
     const requestId = String(input.requestId);
     return this.create({
+      eventId: input.eventId,
       actorId,
       recipientId,
       type: 'follow_request_approved',
@@ -480,4 +497,9 @@ export class NotificationsService {
       type: { $in: NOTIFICATION_CATEGORY_TYPES[category] },
     });
   }
+
+  private isDuplicateKeyError(error: unknown): boolean {
+    return typeof error === 'object' && error !== null && 'code' in error
+      && (error as { code?: number }).code === 11000;
+  }
 }

src/modules/notifications/schemas/notification.schema.ts

@@ -28,13 +28,16 @@ export type NotificationType = (typeof NOTIFICATION_TYPES)[number];
 
 @Schema({ timestamps: true, versionKey: false })
 export class Notification {
+  @Prop({ type: String, trim: true })
+  eventId?: string;
+
   @Prop({ type: Types.ObjectId, ref: User.name, required: true, index: true })
   recipientId!: Types.ObjectId;
 
   @Prop({ type: Types.ObjectId, ref: User.name, required: true, index: true })
   actorId!: Types.ObjectId;
 
-  @Prop({ required: true, enum: NOTIFICATION_TYPES })
+  @Prop({ type: String, required: true, enum: NOTIFICATION_TYPES })
   type!: NotificationType;
 
   @Prop({ type: Types.ObjectId })
@@ -69,6 +72,10 @@ NotificationSchema.index({ recipientId: 1, read: 1, type: 1, createdAt: -1 });
 NotificationSchema.index({ recipientId: 1, type: 1, createdAt: -1 });
 NotificationSchema.index({ recipientId: 1, resourceType: 1, createdAt: -1 });
 NotificationSchema.index({ referenceId: 1 });
+NotificationSchema.index(
+  { eventId: 1 },
+  { unique: true, partialFilterExpression: { eventId: { $type: 'string' } } },
+);
 NotificationSchema.index(
   { recipientId: 1, type: 1, referenceId: 1 },
   {

src/modules/outbox/outbox-dispatcher.service.spec.ts

@@ -0,0 +1,80 @@
+import { OutboxDispatcherService } from './outbox-dispatcher.service';
+
+describe('OutboxDispatcherService', () => {
+  const event = (attempts = 0) => ({
+    _id: 'mongo-id', eventId: 'follow.request.approved:request-1',
+    eventType: 'follow_request_approved_notification', status: 'processing', attempts,
+    createdAt: new Date('2026-08-03T12:00:00.000Z'),
+    payload: { actorId: 'actor', recipientId: 'recipient', requestId: 'request-1' },
+  });
+
+  function setup(values: Record<string, string> = {}) {
+    const model = {
+      findOneAndUpdate: jest.fn(), updateMany: jest.fn(), updateOne: jest.fn(),
+    };
+    model.updateMany.mockReturnValue({ exec: jest.fn().mockResolvedValue({ modifiedCount: 0 }) });
+    model.updateOne.mockReturnValue({ exec: jest.fn().mockResolvedValue({ modifiedCount: 1 }) });
+    const config = { get: jest.fn((key: string, fallback?: string) => values[key] ?? fallback) };
+    const client = { publishFollowRequestApproved: jest.fn() };
+    const logger = { log: jest.fn(), warn: jest.fn(), error: jest.fn() };
+    const connection = { db: { admin: () => ({ command: jest.fn().mockResolvedValue({ setName: 'rs0' }) }) } };
+    const service = new OutboxDispatcherService(model as never, connection as never, config as never, client as never, logger as never);
+    return { service, model, client, logger };
+  }
+
+  it('does not start or claim when either phase-1 flag is false', async () => {
+    const ctx = setup({ NOTIFICATION_SERVICE_ENABLED: 'false', OUTBOX_DISPATCHER_ENABLED: 'true' });
+    await ctx.service.onApplicationBootstrap();
+    await ctx.service.poll();
+    expect(ctx.model.findOneAndUpdate).not.toHaveBeenCalled();
+    expect(ctx.client.publishFollowRequestApproved).not.toHaveBeenCalled();
+  });
+
+  it('uses an atomic claim and recovers expired processing leases', async () => {
+    const ctx = setup({ NOTIFICATION_SERVICE_ENABLED: 'true', OUTBOX_DISPATCHER_ENABLED: 'true', OUTBOX_LEASE_TIMEOUT_MS: '1000' });
+    ctx.model.findOneAndUpdate.mockReturnValue({ exec: jest.fn().mockResolvedValue(event()) });
+    await ctx.service.claimNext(new Date('2026-08-03T12:00:05.000Z'));
+    expect(ctx.model.findOneAndUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({ status: { $in: ['pending', 'failed'] }, $and: expect.any(Array) }),
+      { $set: { status: 'processing', lockedAt: expect.any(Date), lockedBy: expect.any(String) } },
+      { new: true, sort: { createdAt: 1 } },
+    );
+    await ctx.service.recoverExpiredLeases(new Date('2026-08-03T12:00:05.000Z'));
+    expect(ctx.model.updateMany).toHaveBeenCalledWith(
+      expect.objectContaining({ status: 'processing', lockedAt: { $lte: new Date('2026-08-03T12:00:04.000Z') } }),
+      expect.objectContaining({ $set: expect.objectContaining({ status: 'failed', lockedAt: null, lockedBy: null }) }),
+    );
+  });
+
+  it('applies bounded exponential backoff and moves exhausted events to dead', async () => {
+    const ctx = setup({
+      NOTIFICATION_SERVICE_ENABLED: 'true', OUTBOX_DISPATCHER_ENABLED: 'true', OUTBOX_BATCH_SIZE: '1',
+      OUTBOX_MAX_ATTEMPTS: '2', OUTBOX_RETRY_BASE_MS: '100', OUTBOX_RETRY_MAX_MS: '250',
+    });
+    expect(ctx.service.calculateBackoff(1)).toBe(100);
+    expect(ctx.service.calculateBackoff(4)).toBe(250);
+    ctx.model.findOneAndUpdate.mockReturnValue({ exec: jest.fn().mockResolvedValue(event(1)) });
+    ctx.client.publishFollowRequestApproved.mockRejectedValue(new Error('amqp://secret:password@broker failed'));
+    await ctx.service.poll();
+    expect(ctx.model.updateOne).toHaveBeenCalledWith(
+      expect.objectContaining({ status: 'processing' }),
+      expect.objectContaining({ $set: expect.objectContaining({ status: 'dead', attempts: 2, lockedAt: null, lockedBy: null }) }),
+    );
+    expect(ctx.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ error: expect.not.stringContaining('secret:password') }), 'OutboxDispatcherService');
+  });
+
+  it('retries after a lost RPC response and marks processed only after a later response', async () => {
+    const ctx = setup({ NOTIFICATION_SERVICE_ENABLED: 'true', OUTBOX_DISPATCHER_ENABLED: 'true', OUTBOX_BATCH_SIZE: '1' });
+    ctx.model.findOneAndUpdate
+      .mockReturnValueOnce({ exec: jest.fn().mockResolvedValue(event(0)) })
+      .mockReturnValueOnce({ exec: jest.fn().mockResolvedValue(event(1)) });
+    ctx.client.publishFollowRequestApproved
+      .mockRejectedValueOnce(new Error('RPC response timeout'))
+      .mockResolvedValueOnce(undefined);
+    await ctx.service.poll();
+    await ctx.service.poll();
+    expect(ctx.client.publishFollowRequestApproved).toHaveBeenCalledTimes(2);
+    expect(ctx.model.updateOne.mock.calls[0][1].$set.status).toBe('failed');
+    expect(ctx.model.updateOne.mock.calls[1][1].$set.status).toBe('processed');
+  });
+});

src/modules/outbox/outbox-dispatcher.service.ts

@@ -0,0 +1,159 @@
+import { FOLLOW_REQUEST_APPROVED_EVENT } from '../../../libs/events/src';
+import { Injectable, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { InjectConnection, InjectModel } from '@nestjs/mongoose';
+import { randomUUID } from 'crypto';
+import { Connection, Model } from 'mongoose';
+import { AppLoggerService } from '../../infrastructure/logging/app-logger.service';
+import { NotificationServiceClient } from '../notifications/notification-service.client';
+import { OutboxEvent, OutboxEventDocument } from './schemas/outbox-event.schema';
+
+@Injectable()
+export class OutboxDispatcherService implements OnApplicationBootstrap, OnModuleDestroy {
+  private timer?: NodeJS.Timeout;
+  private running = false;
+  private stopped = false;
+  private readonly instanceId = `${process.env.HOSTNAME ?? 'gateway'}-${process.pid}-${randomUUID()}`;
+
+  constructor(
+    @InjectModel(OutboxEvent.name) private readonly model: Model<OutboxEventDocument>,
+    @InjectConnection() private readonly connection: Connection,
+    private readonly config: ConfigService,
+    private readonly client: NotificationServiceClient,
+    private readonly logger: AppLoggerService,
+  ) {}
+
+  async onApplicationBootstrap(): Promise<void> {
+    if (!this.enabled()) return;
+    await this.assertTransactionCapableMongo();
+    this.schedule(0);
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    this.stopped = true;
+    if (this.timer) clearTimeout(this.timer);
+    while (this.running) await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+
+  async poll(): Promise<void> {
+    if (!this.enabled() || this.running || this.stopped) return;
+    this.running = true;
+    let claimed = 0;
+    let processed = 0;
+    let failed = 0;
+    try {
+      await this.recoverExpiredLeases();
+      for (let index = 0; index < this.number('OUTBOX_BATCH_SIZE', 25); index += 1) {
+        const event = await this.claimNext();
+        if (!event) break;
+        claimed += 1;
+        try {
+          await this.dispatch(event);
+          await this.markProcessed(event);
+          processed += 1;
+        } catch (error) {
+          await this.markFailed(event, error);
+          failed += 1;
+        }
+      }
+      if (claimed > 0) this.logger.log({ message: 'outbox poll completed', instanceId: this.instanceId, claimed, processed, failed }, OutboxDispatcherService.name);
+    } finally {
+      this.running = false;
+    }
+  }
+
+  async claimNext(now = new Date()): Promise<OutboxEventDocument | null> {
+    const staleBefore = new Date(now.getTime() - this.number('OUTBOX_LEASE_TIMEOUT_MS', 60_000));
+    return this.model.findOneAndUpdate(
+      {
+        eventType: 'follow_request_approved_notification',
+        status: { $in: ['pending', 'failed'] },
+        $and: [
+          { $or: [{ nextRetryAt: { $lte: now } }, { nextRetryAt: { $exists: false } }] },
+          { $or: [{ lockedAt: null }, { lockedAt: { $exists: false } }, { lockedAt: { $lte: staleBefore } }] },
+        ],
+      },
+      { $set: { status: 'processing', lockedAt: now, lockedBy: this.instanceId } },
+      { new: true, sort: { createdAt: 1 } },
+    ).exec();
+  }
+
+  async recoverExpiredLeases(now = new Date()): Promise<number> {
+    const staleBefore = new Date(now.getTime() - this.number('OUTBOX_LEASE_TIMEOUT_MS', 60_000));
+    const result = await this.model.updateMany(
+      { eventType: 'follow_request_approved_notification', status: 'processing', lockedAt: { $lte: staleBefore } },
+      { $set: { status: 'failed', nextRetryAt: now, lastError: 'processing lease expired', lockedAt: null, lockedBy: null } },
+    ).exec();
+    return result.modifiedCount;
+  }
+
+  calculateBackoff(attempts: number): number {
+    return Math.min(
+      this.number('OUTBOX_RETRY_MAX_MS', 60_000),
+      this.number('OUTBOX_RETRY_BASE_MS', 1_000) * (2 ** Math.max(0, attempts - 1)),
+    );
+  }
+
+  private async dispatch(event: OutboxEventDocument): Promise<void> {
+    await this.client.publishFollowRequestApproved({
+      eventId: event.eventId || `follow.request.approved:${String(event.payload.requestId ?? '')}`,
+      eventType: FOLLOW_REQUEST_APPROVED_EVENT,
+      occurredAt: event.createdAt.toISOString(),
+      payload: {
+        requestId: String(event.payload.requestId ?? ''),
+        actorId: String(event.payload.actorId ?? ''),
+        recipientId: String(event.payload.recipientId ?? ''),
+      },
+    });
+  }
+
+  private async markProcessed(event: OutboxEventDocument): Promise<void> {
+    await this.model.updateOne(
+      { _id: event._id, status: 'processing', lockedBy: this.instanceId },
+      { $set: { status: 'processed', processedAt: new Date(), lastError: '', lockedAt: null, lockedBy: null } },
+    ).exec();
+  }
+
+  private async markFailed(event: OutboxEventDocument, error: unknown): Promise<void> {
+    const attempts = event.attempts + 1;
+    const dead = attempts >= this.number('OUTBOX_MAX_ATTEMPTS', 8);
+    const message = this.safeError(error);
+    await this.model.updateOne(
+      { _id: event._id, status: 'processing', lockedBy: this.instanceId },
+      { $set: {
+        status: dead ? 'dead' : 'failed', attempts, lastError: message,
+        nextRetryAt: new Date(Date.now() + this.calculateBackoff(attempts)),
+        deadAt: dead ? new Date() : null, lockedAt: null, lockedBy: null,
+      } },
+    ).exec();
+    this.logger.warn({ eventId: event.eventId, attempts, status: dead ? 'dead' : 'failed', error: message }, OutboxDispatcherService.name);
+  }
+
+  private schedule(delay: number): void {
+    if (this.stopped) return;
+    this.timer = setTimeout(async () => {
+      try { await this.poll(); }
+      catch (error) { this.logger.error({ error: this.safeError(error) }, OutboxDispatcherService.name); }
+      finally { this.schedule(this.number('OUTBOX_POLL_INTERVAL_MS', 1_000)); }
+    }, delay);
+    this.timer.unref?.();
+  }
+
+  private enabled(): boolean {
+    return String(this.config.get('OUTBOX_DISPATCHER_ENABLED') ?? 'false').toLowerCase() === 'true'
+      && String(this.config.get('NOTIFICATION_SERVICE_ENABLED') ?? 'false').toLowerCase() === 'true';
+  }
+
+  private async assertTransactionCapableMongo(): Promise<void> {
+    const hello = await this.connection.db?.admin().command({ hello: 1 }) as { setName?: string; msg?: string };
+    if (!hello?.setName && hello?.msg !== 'isdbgrid') throw new Error('Outbox dispatcher requires a transaction-capable MongoDB replica set or mongos');
+  }
+
+  private number(key: string, fallback: number): number {
+    return Number(this.config.get<string>(key) ?? fallback);
+  }
+
+  private safeError(error: unknown): string {
+    return (error instanceof Error ? error.message : String(error)).replace(/amqps?:\/\/[^@\s]+@/gi, 'amqp://[redacted]@').slice(0, 500);
+  }
+}

src/modules/outbox/outbox.module.ts

@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { MongooseModule } from '@nestjs/mongoose';
 import { NotificationsModule } from '../notifications/notifications.module';
 import { OutboxService } from './outbox.service';
+import { OutboxDispatcherService } from './outbox-dispatcher.service';
 import { OutboxEvent, OutboxEventSchema } from './schemas/outbox-event.schema';
 
 @Module({
@@ -9,7 +10,7 @@ import { OutboxEvent, OutboxEventSchema } from './schemas/outbox-event.schema';
     MongooseModule.forFeature([{ name: OutboxEvent.name, schema: OutboxEventSchema }]),
     NotificationsModule,
   ],
-  providers: [OutboxService],
-  exports: [OutboxService],
+  providers: [OutboxService, OutboxDispatcherService],
+  exports: [OutboxService, OutboxDispatcherService],
 })
 export class OutboxModule {}

src/modules/outbox/outbox.service.spec.ts

@@ -4,7 +4,7 @@ import { NotificationsService } from '../notifications/notifications.service';
 import { OutboxService } from './outbox.service';
 
 describe('OutboxService', () => {
-  const model = { create: jest.fn(), findById: jest.fn() };
+  const model = { create: jest.fn(), findOne: jest.fn(), updateOne: jest.fn() };
   const notifications = {
     createFollowNotification: jest.fn(),
     createFollowRequestApprovedNotification: jest.fn(),
@@ -18,6 +18,10 @@ describe('OutboxService', () => {
     jest.clearAllMocks();
   });
 
+  afterEach(() => {
+    delete process.env.NOTIFICATION_SERVICE_ENABLED;
+  });
+
   it('registers its processor and enqueues persisted follow events', async () => {
     service.onModuleInit();
     expect(queue.registerProcessor).toHaveBeenCalledWith('process_outbox_event', expect.any(Function));
@@ -28,47 +32,63 @@ describe('OutboxService', () => {
     await processor({});
     expect(processSpy).toHaveBeenLastCalledWith('');
 
-    model.create.mockResolvedValue({ id: 'event-1' });
+    model.create.mockResolvedValue({ eventId: 'event-1' });
     await service.enqueueFollowNotification('actor', 'recipient');
-    expect(model.create).toHaveBeenCalledWith({
+    expect(model.create).toHaveBeenCalledWith(expect.objectContaining({
+      eventId: expect.any(String),
       eventType: 'follow_notification',
       payload: { actorId: 'actor', recipientId: 'recipient', referenceId: '' },
       status: 'pending',
-    });
+    }));
     expect(queue.enqueue).toHaveBeenCalledWith('process_outbox_event', { eventId: 'event-1' });
 
-    model.create.mockResolvedValue({ id: 'event-2' });
+    model.updateOne.mockReturnValue({ exec: jest.fn().mockResolvedValue({ upsertedCount: 1 }) });
     await service.enqueueFollowRequestApprovedNotification('actor', 'recipient', 'request');
-    expect(model.create).toHaveBeenCalledWith({
-      eventType: 'follow_request_approved_notification',
-      payload: { actorId: 'actor', recipientId: 'recipient', requestId: 'request' },
-      status: 'pending',
-    });
+    expect(model.updateOne).toHaveBeenCalledWith(
+      { eventId: 'follow.request.approved:request' },
+      expect.objectContaining({ $setOnInsert: expect.objectContaining({ eventType: 'follow_request_approved_notification' }) }),
+      { upsert: true, session: undefined },
+    );
   });
 
   it('processes follow request approval notifications once', async () => {
     const event = {
       id: 'e-approved',
+      eventId: 'e-approved',
       eventType: 'follow_request_approved_notification',
       status: 'pending',
       attempts: 0,
       payload: { actorId: 'owner', recipientId: 'requester', requestId: 'request' },
       save: jest.fn(),
     };
-    model.findById.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
+    model.findOne.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
     await service.processEvent(event.id);
     expect(notifications.createFollowRequestApprovedNotification).toHaveBeenCalledWith({
       actorId: 'owner',
       recipientId: 'requester',
       requestId: 'request',
+      eventId: 'e-approved',
     });
     expect(event.status).toBe('processed');
   });
 
+  it('never lets a legacy Gateway worker create approval notifications in remote mode', async () => {
+    process.env.NOTIFICATION_SERVICE_ENABLED = 'true';
+    const event = {
+      id: 'legacy-job', eventId: 'approval-event', eventType: 'follow_request_approved_notification',
+      status: 'pending', attempts: 0, payload: {}, save: jest.fn(),
+    };
+    model.findOne.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
+    await service.processEvent('approval-event');
+    expect(notifications.createFollowRequestApprovedNotification).not.toHaveBeenCalled();
+    expect(event.status).toBe('pending');
+    expect(event.save).not.toHaveBeenCalled();
+  });
+
   it('ignores missing and already processed events', async () => {
-    model.findById.mockReturnValueOnce({ exec: jest.fn().mockResolvedValue(null) });
+    model.findOne.mockReturnValueOnce({ exec: jest.fn().mockResolvedValue(null) });
     await service.processEvent('missing');
-    model.findById.mockReturnValueOnce({ exec: jest.fn().mockResolvedValue({ status: 'processed' }) });
+    model.findOne.mockReturnValueOnce({ exec: jest.fn().mockResolvedValue({ status: 'processed' }) });
     await service.processEvent('done');
     expect(notifications.createFollowNotification).not.toHaveBeenCalled();
   });
@@ -78,7 +98,7 @@ describe('OutboxService', () => {
       id: 'e1', eventType: 'follow_notification', status: 'pending', attempts: 0,
       payload: { actorId: 'a', recipientId: 'r', referenceId: 'post' }, save: jest.fn(),
     };
-    model.findById.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
+    model.findOne.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
     await service.processEvent('e1');
     expect(notifications.createFollowNotification).toHaveBeenCalledWith('a', 'r', 'post');
     expect(event).toEqual(expect.objectContaining({ status: 'processed', attempts: 1, lastError: '', processedAt: expect.any(Date) }));
@@ -90,9 +110,9 @@ describe('OutboxService', () => {
       id: 'e2', eventType: 'follow_notification', status: 'pending', attempts: 2,
       payload: {}, save: jest.fn(),
     };
-    model.findById.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
+    model.findOne.mockReturnValue({ exec: jest.fn().mockResolvedValue(event) });
     (notifications.createFollowNotification as jest.Mock).mockRejectedValue('network');
-    await service.processEvent('e2');
+    await expect(service.processEvent('e2')).rejects.toBe('network');
     expect(event).toEqual(expect.objectContaining({ status: 'failed', lastError: 'unknown outbox error', attempts: 3 }));
     expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ eventId: 'e2' }), 'OutboxService');
     expect(event.save).toHaveBeenCalled();

src/modules/outbox/outbox.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, OnModuleInit } from '@nestjs/common';
 import { InjectModel } from '@nestjs/mongoose';
-import { Model } from 'mongoose';
+import { randomUUID } from 'crypto';
+import { ClientSession, Model } from 'mongoose';
 import { AppLoggerService } from '../../infrastructure/logging/app-logger.service';
 import { AppQueueService } from '../../infrastructure/queue/app-queue.service';
 import { NotificationsService } from '../notifications/notifications.service';
@@ -25,16 +26,12 @@ export class OutboxService implements OnModuleInit {
 
   async enqueueFollowNotification(actorId: string, recipientId: string, referenceId?: string): Promise<void> {
     const event = await this.outboxEventModel.create({
+      eventId: randomUUID(),
       eventType: 'follow_notification',
-      payload: {
-        actorId,
-        recipientId,
-        referenceId: referenceId ?? '',
-      },
+      payload: { actorId, recipientId, referenceId: referenceId ?? '' },
       status: 'pending',
     });
-
-    await this.queueService.enqueue(OutboxService.PROCESS_EVENT_JOB, { eventId: event.id });
+    await this.enqueuePersistedEvent(event.eventId);
   }
 
   async enqueueFollowRequestApprovedNotification(
@@ -42,24 +39,59 @@ export class OutboxService implements OnModuleInit {
     recipientId: string,
     requestId: string,
   ): Promise<void> {
-    const event = await this.outboxEventModel.create({
-      eventType: 'follow_request_approved_notification',
-      payload: {
-        actorId: String(actorId),
-        recipientId: String(recipientId),
-        requestId: String(requestId),
-      },
-      status: 'pending',
-    });
+    const eventId = await this.createFollowRequestApprovedEvent(actorId, recipientId, requestId);
+    await this.enqueuePersistedEvent(eventId);
+  }
 
-    await this.queueService.enqueue(OutboxService.PROCESS_EVENT_JOB, { eventId: event.id });
+  async createFollowRequestApprovedEvent(
+    actorId: string,
+    recipientId: string,
+    requestId: string,
+    session?: ClientSession,
+  ): Promise<string> {
+    const eventId = `follow.request.approved:${String(requestId)}`;
+    await this.outboxEventModel.updateOne(
+      { eventId },
+      { $setOnInsert: {
+        eventId,
+        eventType: 'follow_request_approved_notification',
+        payload: { actorId: String(actorId), recipientId: String(recipientId), requestId: String(requestId) },
+        correlationId: eventId,
+        requestId: String(requestId),
+        status: 'pending',
+        attempts: 0,
+        nextRetryAt: new Date(),
+      } },
+      { upsert: true, session },
+    ).exec();
+    return eventId;
+  }
+
+  async enqueuePersistedEvent(eventId: string): Promise<void> {
+    await this.queueService.enqueue(OutboxService.PROCESS_EVENT_JOB, { eventId });
+  }
+
+  async listByStatuses(statuses: OutboxEvent['status'][], limit = 100): Promise<OutboxEventDocument[]> {
+    return this.outboxEventModel.find({ status: { $in: statuses } }).sort({ createdAt: 1 }).limit(limit).exec();
+  }
+
+  async retryEvent(eventId: string): Promise<boolean> {
+    const result = await this.outboxEventModel.updateOne(
+      { eventId, status: { $in: ['failed', 'dead'] } },
+      { $set: { status: 'pending', nextRetryAt: new Date(), lastError: '', lockedAt: null, lockedBy: null }, $unset: { deadAt: 1 } },
+    ).exec();
+    return result.modifiedCount === 1;
   }
 
   async processEvent(eventId: string): Promise<void> {
-    const event = await this.outboxEventModel.findById(eventId).exec();
-    if (!event || event.status === 'processed') {
-      return;
-    }
+    const event = await this.outboxEventModel.findOne({
+      $or: [{ eventId }, ...(eventId.match(/^[a-f\d]{24}$/i) ? [{ _id: eventId }] : [])],
+    }).exec();
+    if (!event || event.status === 'processed') return;
+    // A legacy BullMQ job may still exist during cutover. The dispatcher is the sole
+    // owner of approval events whenever remote notification routing is enabled.
+    if (event.eventType === 'follow_request_approved_notification'
+      && process.env.NOTIFICATION_SERVICE_ENABLED === 'true') return;
 
     try {
       if (event.eventType === 'follow_notification') {
@@ -68,29 +100,22 @@ export class OutboxService implements OnModuleInit {
           String(event.payload.recipientId ?? ''),
           String(event.payload.referenceId ?? ''),
         );
-      }
-      if (event.eventType === 'follow_request_approved_notification') {
+      } else if (event.eventType === 'follow_request_approved_notification') {
         await this.notificationsService.createFollowRequestApprovedNotification({
           actorId: String(event.payload.actorId ?? ''),
           recipientId: String(event.payload.recipientId ?? ''),
           requestId: String(event.payload.requestId ?? ''),
+          eventId: event.eventId || event.id,
         });
       }
-
       event.status = 'processed';
       event.processedAt = new Date();
       event.lastError = '';
     } catch (error) {
       event.status = 'failed';
       event.lastError = error instanceof Error ? error.message : 'unknown outbox error';
-      this.logger.warn(
-        {
-          eventId: event.id,
-          eventType: event.eventType,
-          error: event.lastError,
-        },
-        OutboxService.name,
-      );
+      this.logger.warn({ eventId: event.eventId || event.id, eventType: event.eventType, error: event.lastError }, OutboxService.name);
+      throw error;
     } finally {
       event.attempts += 1;
       await event.save();

src/modules/outbox/schemas/outbox-event.schema.ts

@@ -5,14 +5,26 @@ export type OutboxEventDocument = HydratedDocument<OutboxEvent>;
 
 @Schema({ timestamps: true, versionKey: false })
 export class OutboxEvent {
+  declare createdAt: Date;
+
+  declare updatedAt: Date;
+
+  @Prop({ required: true })
+  eventId!: string;
+
   @Prop({ required: true, index: true })
   eventType!: string;
 
   @Prop({ required: true, type: Object })
   payload!: Record<string, unknown>;
 
-  @Prop({ required: true, default: 'pending', enum: ['pending', 'processed', 'failed'], index: true })
-  status!: 'pending' | 'processed' | 'failed';
+  @Prop({
+    required: true,
+    default: 'pending',
+    enum: ['pending', 'processing', 'processed', 'failed', 'dead'],
+    index: true,
+  })
+  status!: 'pending' | 'processing' | 'processed' | 'failed' | 'dead';
 
   @Prop({ required: true, default: 0, min: 0 })
   attempts!: number;
@@ -20,9 +32,31 @@ export class OutboxEvent {
   @Prop({ default: '' })
   lastError!: string;
 
+  @Prop({ type: Date, default: Date.now, index: true })
+  nextRetryAt!: Date;
+
+  @Prop({ type: Date, default: null, index: true })
+  lockedAt?: Date | null;
+
+  @Prop({ type: String, default: null })
+  lockedBy?: string | null;
+
+  @Prop({ default: '' })
+  correlationId!: string;
+
+  @Prop({ default: '' })
+  requestId!: string;
+
   @Prop({ type: Date, default: null, index: true })
   processedAt?: Date | null;
+
+  @Prop({ type: Date, default: null })
+  deadAt?: Date | null;
 }
 
 export const OutboxEventSchema = SchemaFactory.createForClass(OutboxEvent);
-OutboxEventSchema.index({ status: 1, createdAt: 1 });
+OutboxEventSchema.index(
+  { eventId: 1 },
+  { unique: true, partialFilterExpression: { eventId: { $type: 'string' } } },
+);
+OutboxEventSchema.index({ eventType: 1, status: 1, nextRetryAt: 1, lockedAt: 1, createdAt: 1 });

src/modules/users/users.repository.ts

@@ -61,12 +61,12 @@ export class UsersRepository {
       .exec();
   }
 
-  async setFollowersCount(userId: string, followersCount: number): Promise<void> {
-    await this.userModel.findByIdAndUpdate(userId, { followersCount }, { new: false }).exec();
+  async setFollowersCount(userId: string, followersCount: number, session?: ClientSession): Promise<void> {
+    await this.userModel.findByIdAndUpdate(userId, { followersCount }, { new: false, session }).exec();
   }
 
-  async setFollowingCount(userId: string, followingCount: number): Promise<void> {
-    await this.userModel.findByIdAndUpdate(userId, { followingCount }, { new: false }).exec();
+  async setFollowingCount(userId: string, followingCount: number, session?: ClientSession): Promise<void> {
+    await this.userModel.findByIdAndUpdate(userId, { followingCount }, { new: false, session }).exec();
   }
 
   async setPresence(userId: string, isOnline: boolean, lastSeenAt: Date): Promise<void> {