diff --git a/.dockerignore b/.dockerignore index c91929e..c7fa3da 100644 --- a/.dockerignore +++ b/.dockerignore @@ -15,3 +15,6 @@ npm-debug.log* .idea postman docs +oudelaa_dashboard +stream_0 +stream_1 diff --git a/.env.example b/.env.example index 834e107..983a567 100644 --- a/.env.example +++ b/.env.example @@ -81,6 +81,43 @@ QUEUE_DEFAULT_BACKOFF_MS=1000 QUEUE_REMOVE_ON_COMPLETE=true QUEUE_WORKER_CONCURRENCY=5 +# Phase 1 microservice routing (false keeps the current in-process path) +NOTIFICATION_SERVICE_ENABLED=false +# The polling dispatcher is independently disabled for the current production monolith. +OUTBOX_DISPATCHER_ENABLED=false +OUTBOX_POLL_INTERVAL_MS=1000 +OUTBOX_BATCH_SIZE=25 +OUTBOX_MAX_ATTEMPTS=8 +OUTBOX_LEASE_TIMEOUT_MS=60000 +OUTBOX_RETRY_BASE_MS=1000 +OUTBOX_RETRY_MAX_MS=60000 +# Local development only. Production must inject non-guest credentials and TLS/amqps. +RABBITMQ_URL=amqp://guest:guest@127.0.0.1:5672 +RABBITMQ_HOST=127.0.0.1 +RABBITMQ_PORT=5672 +RABBITMQ_USERNAME= +RABBITMQ_PASSWORD= +RABBITMQ_VHOST=/ +RABBITMQ_TLS_ENABLED=false +RABBITMQ_TLS_REQUIRED=true +RABBITMQ_CA_CERT_BASE64= +RABBITMQ_CA_CERT_PATH= +RABBITMQ_CLIENT_CERT_BASE64= +RABBITMQ_CLIENT_CERT_PATH= +RABBITMQ_CLIENT_KEY_BASE64= +RABBITMQ_CLIENT_KEY_PATH= +RABBITMQ_TLS_REJECT_UNAUTHORIZED=true +RABBITMQ_SERVERNAME= +RABBITMQ_HEARTBEAT_SECONDS=10 +RABBITMQ_RECONNECT_SECONDS=5 +RABBITMQ_CONNECTION_TIMEOUT_MS=10000 +NOTIFICATION_RPC_TIMEOUT_MS=3000 +NOTIFICATION_RPC_RETRIES=1 +NOTIFICATION_CIRCUIT_FAILURE_THRESHOLD=3 +NOTIFICATION_CIRCUIT_RESET_MS=15000 +NOTIFICATION_SERVICE_PORT=4001 +NOTIFICATION_RMQ_PREFETCH=10 + STORAGE_PROVIDER=local MEDIA_ACCESS_MODE=direct STORAGE_BASE_PATH=uploads diff --git a/Dockerfile.microservices b/Dockerfile.microservices new file mode 100644 index 0000000..3736b35 --- /dev/null +++ b/Dockerfile.microservices @@ -0,0 +1,26 @@ +FROM node:22-alpine AS dependencies +WORKDIR /app +COPY package*.json ./ +RUN npm ci + +FROM dependencies AS build +COPY . . +RUN npm run build + +FROM node:22-alpine AS api-gateway +WORKDIR /app +ENV NODE_ENV=production +COPY --from=dependencies /app/node_modules ./node_modules +COPY --from=build /app/dist ./dist +COPY package.json ./ +EXPOSE 4000 +CMD ["node", "dist/main.js"] + +FROM node:22-alpine AS notification-service +WORKDIR /app +ENV NODE_ENV=production +COPY --from=dependencies /app/node_modules ./node_modules +COPY --from=build /app/dist ./dist +COPY package.json ./ +EXPOSE 4001 +CMD ["node", "dist/apps/notification-service/main.js"] diff --git a/apps/api-gateway/src/main.ts b/apps/api-gateway/src/main.ts new file mode 100644 index 0000000..477bd2c --- /dev/null +++ b/apps/api-gateway/src/main.ts @@ -0,0 +1,3 @@ +import { bootstrap } from '../../../src/main'; + +void bootstrap(); diff --git a/apps/api-gateway/tsconfig.app.json b/apps/api-gateway/tsconfig.app.json new file mode 100644 index 0000000..c7c75f9 --- /dev/null +++ b/apps/api-gateway/tsconfig.app.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "declaration": false, + "outDir": "../../dist/apps/api-gateway" + }, + "include": ["src/**/*.ts", "../../src/**/*.ts", "../../libs/**/*.ts"], + "exclude": ["node_modules", "dist", "test", "**/*.spec.ts"] +} diff --git a/apps/notification-service/src/actor.schema.ts b/apps/notification-service/src/actor.schema.ts new file mode 100644 index 0000000..2298bf5 --- /dev/null +++ b/apps/notification-service/src/actor.schema.ts @@ -0,0 +1,14 @@ +import { Schema } from 'mongoose'; + +// A deliberately small read model. Notifications only populate these public actor fields. +export const ActorSchema = new Schema( + { + name: String, + username: String, + stageName: String, + avatar: Schema.Types.Mixed, + isVerified: Boolean, + isDisabled: Boolean, + }, + { collection: 'users', strict: false, versionKey: false }, +); diff --git a/apps/notification-service/src/app.module.ts b/apps/notification-service/src/app.module.ts new file mode 100644 index 0000000..ec561a2 --- /dev/null +++ b/apps/notification-service/src/app.module.ts @@ -0,0 +1,47 @@ +import { NOTIFICATION_RMQ_QUEUE } from '@app/contracts'; +import { Module } from '@nestjs/common'; +import { ConfigModule, ConfigService } from '@nestjs/config'; +import { MongooseModule } from '@nestjs/mongoose'; +import { NotificationRealtimeSink } from './notification-realtime.sink'; +import { NotificationConsumer } from './notification.consumer'; +import { HealthController } from './health.controller'; +import { NOTIFICATION_SERVICE_STATE, NotificationServiceState } from './service-state'; +import { ActorSchema } from './actor.schema'; +import { NotificationsGateway } from '../../../src/modules/notifications/notifications.gateway'; +import { NotificationsRepository } from '../../../src/modules/notifications/notifications.repository'; +import { NotificationsService } from '../../../src/modules/notifications/notifications.service'; +import { Notification, NotificationSchema } from '../../../src/modules/notifications/schemas/notification.schema'; + +export const notificationServiceState: NotificationServiceState = { + rabbitmqReady: false, + draining: false, +}; + +@Module({ + imports: [ + ConfigModule.forRoot({ isGlobal: true }), + MongooseModule.forRootAsync({ + inject: [ConfigService], + useFactory: (config: ConfigService) => ({ + uri: config.get('NOTIFICATION_MONGODB_URI') + ?? config.get('MONGODB_URI') + ?? 'mongodb://127.0.0.1:27017/oudelaa', + autoIndex: config.get('MONGODB_AUTO_INDEX', 'true') === 'true', + }), + }), + MongooseModule.forFeature([ + { name: Notification.name, schema: NotificationSchema, collection: 'notifications' }, + { name: 'User', schema: ActorSchema, collection: 'users' }, + ]), + ], + controllers: [HealthController, NotificationConsumer], + providers: [ + NotificationsRepository, + NotificationsService, + NotificationRealtimeSink, + { provide: NotificationsGateway, useExisting: NotificationRealtimeSink }, + { provide: NOTIFICATION_SERVICE_STATE, useValue: notificationServiceState }, + { provide: 'NOTIFICATION_QUEUE_NAME', useValue: NOTIFICATION_RMQ_QUEUE }, + ], +}) +export class NotificationServiceModule {} diff --git a/apps/notification-service/src/health.controller.ts b/apps/notification-service/src/health.controller.ts new file mode 100644 index 0000000..8f4efdc --- /dev/null +++ b/apps/notification-service/src/health.controller.ts @@ -0,0 +1,31 @@ +import { Controller, Get, Inject, ServiceUnavailableException } from '@nestjs/common'; +import { InjectConnection } from '@nestjs/mongoose'; +import { Connection } from 'mongoose'; +import { NOTIFICATION_SERVICE_STATE, NotificationServiceState } from './service-state'; + +@Controller('health') +export class HealthController { + constructor( + @InjectConnection() private readonly connection: Connection, + @Inject(NOTIFICATION_SERVICE_STATE) private readonly state: NotificationServiceState, + ) {} + + @Get() + @Get('live') + live() { + return { status: this.state.draining ? 'draining' : 'ok', service: 'notification-service' }; + } + + @Get('ready') + async ready() { + const mongodbReady = this.connection.readyState === 1; + const ready = !this.state.draining && mongodbReady && this.state.rabbitmqReady; + const response = { + status: ready ? 'ready' : 'not-ready', + service: 'notification-service', + checks: { mongodb: mongodbReady, rabbitmq: this.state.rabbitmqReady }, + }; + if (!ready) throw new ServiceUnavailableException(response); + return response; + } +} diff --git a/apps/notification-service/src/main.ts b/apps/notification-service/src/main.ts new file mode 100644 index 0000000..431b2ec --- /dev/null +++ b/apps/notification-service/src/main.ts @@ -0,0 +1,44 @@ +import { NOTIFICATION_RMQ_QUEUE } from '@app/contracts'; +import { processEnvSource, resolveRabbitMqConfig } from '@app/common'; +import { Logger, ShutdownSignal, ValidationPipe } from '@nestjs/common'; +import { NestFactory } from '@nestjs/core'; +import { MicroserviceOptions, Transport } from '@nestjs/microservices'; +import { NotificationServiceModule, notificationServiceState } from './app.module'; +import { assertNotificationTopology, notificationQueueArguments } from './rabbitmq-topology'; + +export async function bootstrapNotificationService(): Promise { + const app = await NestFactory.create(NotificationServiceModule, { bufferLogs: true }); + const logger = new Logger('NotificationServiceBootstrap'); + const rabbit = resolveRabbitMqConfig(processEnvSource(), 'notification-service', true); + const port = Number(process.env.NOTIFICATION_SERVICE_PORT ?? 4001); + + app.enableShutdownHooks([ShutdownSignal.SIGINT, ShutdownSignal.SIGTERM]); + app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true })); + const microservice = app.connectMicroservice({ + transport: Transport.RMQ, + options: { + urls: [rabbit.url], + socketOptions: rabbit.socketOptions, + queue: NOTIFICATION_RMQ_QUEUE, + noAck: false, + prefetchCount: Number(process.env.NOTIFICATION_RMQ_PREFETCH ?? 10), + queueOptions: { durable: true, autoDelete: false, exclusive: false, arguments: notificationQueueArguments }, + }, + }); + const status = (microservice as unknown as { status: { subscribe: (callback: (value: string) => void) => void } }).status; + status.subscribe((value) => { + notificationServiceState.rabbitmqReady = value === 'connected' || value === 'unblocked'; + }); + + const markDraining = () => { notificationServiceState.draining = true; }; + process.once('SIGTERM', markDraining); + process.once('SIGINT', markDraining); + + await assertNotificationTopology(rabbit); + await app.startAllMicroservices(); + notificationServiceState.rabbitmqReady = true; + await app.listen(port, process.env.NOTIFICATION_SERVICE_HOST ?? '0.0.0.0'); + logger.log(`HTTP health endpoint listening on ${port}; consuming ${NOTIFICATION_RMQ_QUEUE}; broker=${rabbit.sanitizedEndpoint}`); +} + +if (require.main === module) void bootstrapNotificationService(); diff --git a/apps/notification-service/src/notification-realtime.sink.ts b/apps/notification-service/src/notification-realtime.sink.ts new file mode 100644 index 0000000..b4155ea --- /dev/null +++ b/apps/notification-service/src/notification-realtime.sink.ts @@ -0,0 +1,8 @@ +import { Injectable } from '@nestjs/common'; + +// Socket.IO remains owned by the gateway. The RMQ response carries realtime data back to it. +@Injectable() +export class NotificationRealtimeSink { + emitCreated(): void {} + emitUnreadCount(): void {} +} diff --git a/apps/notification-service/src/notification.consumer.spec.ts b/apps/notification-service/src/notification.consumer.spec.ts new file mode 100644 index 0000000..33269ee --- /dev/null +++ b/apps/notification-service/src/notification.consumer.spec.ts @@ -0,0 +1,100 @@ +import { FOLLOW_REQUEST_APPROVED_EVENT, FollowRequestApprovedEvent } from '@app/events'; +import { NotificationConsumer } from './notification.consumer'; + +const event: FollowRequestApprovedEvent = { + eventId: 'event-1', + eventType: FOLLOW_REQUEST_APPROVED_EVENT, + occurredAt: '2026-08-03T10:00:00.000Z', + payload: { requestId: 'request-1', actorId: 'actor-1', recipientId: 'recipient-1' }, +}; + +function rmqContext() { + const channel = { ack: jest.fn(), nack: jest.fn() }; + const message = { + properties: { headers: { correlationId: 'correlation-1', requestId: 'request-1' } }, + }; + return { + channel, + message, + context: { + getChannelRef: () => channel, + getMessage: () => message, + } as never, + }; +} + +describe('NotificationConsumer', () => { + const notification = { toJSON: () => ({ _id: 'notification-1' }) }; + let notifications: Record; + let consumer: NotificationConsumer; + + beforeEach(() => { + notifications = { + createFollowRequestApprovedNotification: jest.fn().mockResolvedValue(notification), + getUnreadCounts: jest.fn().mockResolvedValue({ total: 1, follows: 0, followRequests: 1 }), + }; + consumer = new NotificationConsumer(notifications as never); + jest.spyOn(console, 'log').mockImplementation(() => undefined); + jest.spyOn(console, 'error').mockImplementation(() => undefined); + }); + + afterEach(() => jest.restoreAllMocks()); + + it('consumes follow.request.approved and acknowledges only after persistence', async () => { + const rmq = rmqContext(); + const result = await consumer.followRequestApproved(event, rmq.context); + expect(notifications.createFollowRequestApprovedNotification).toHaveBeenCalledWith({ + ...event.payload, + eventId: event.eventId, + }); + expect(result).toEqual(expect.objectContaining({ + result: notification, + realtime: expect.objectContaining({ recipientId: 'recipient-1', unreadCount: 1 }), + })); + expect(rmq.channel.ack).toHaveBeenCalledWith(rmq.message); + expect(rmq.channel.nack).not.toHaveBeenCalled(); + }); + + it('retries transient failures with a finite limit', async () => { + notifications.createFollowRequestApprovedNotification + .mockRejectedValueOnce(new Error('temporary-1')) + .mockRejectedValueOnce(new Error('temporary-2')) + .mockResolvedValueOnce(notification); + const rmq = rmqContext(); + await consumer.followRequestApproved(event, rmq.context); + expect(notifications.createFollowRequestApprovedNotification).toHaveBeenCalledTimes(3); + expect(rmq.channel.ack).toHaveBeenCalledTimes(1); + }); + + it('rejects to the DLQ after retry exhaustion', async () => { + notifications.createFollowRequestApprovedNotification.mockRejectedValue(new Error('mongo down')); + const rmq = rmqContext(); + await expect(consumer.followRequestApproved(event, rmq.context)).rejects.toThrow('mongo down'); + expect(notifications.createFollowRequestApprovedNotification).toHaveBeenCalledTimes(3); + expect(rmq.channel.nack).toHaveBeenCalledWith(rmq.message, false, false); + expect(rmq.channel.ack).not.toHaveBeenCalled(); + }); + + it('rejects an invalid contract without persistence', async () => { + const rmq = rmqContext(); + await expect(consumer.followRequestApproved({ ...event, eventId: '' }, rmq.context)).rejects.toThrow( + 'Invalid follow.request.approved contract', + ); + expect(notifications.createFollowRequestApprovedNotification).not.toHaveBeenCalled(); + expect(rmq.channel.nack).toHaveBeenCalledWith(rmq.message, false, false); + }); + + it('requires non-empty string IDs and a canonical ISO occurredAt value', async () => { + for (const invalid of [ + { ...event, occurredAt: 'August 3, 2026' }, + { ...event, payload: { ...event.payload, actorId: '' } }, + ]) { + const rmq = rmqContext(); + await expect(consumer.followRequestApproved(invalid, rmq.context)).rejects.toThrow( + 'Invalid follow.request.approved contract', + ); + expect(rmq.channel.nack).toHaveBeenCalledWith(rmq.message, false, false); + } + expect(notifications.createFollowRequestApprovedNotification).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/notification-service/src/notification.consumer.ts b/apps/notification-service/src/notification.consumer.ts new file mode 100644 index 0000000..406f113 --- /dev/null +++ b/apps/notification-service/src/notification.consumer.ts @@ -0,0 +1,155 @@ +import { withLimitedRetry } from '@app/common'; +import { + MessageContext, + NOTIFICATION_RPC, + NotificationListRequest, + NotificationMarkAllReadRequest, + NotificationMarkReadRequest, + NotificationRpcResponse, + NotificationUnreadCountRequest, +} from '@app/contracts'; +import { + FOLLOW_REQUEST_APPROVED_EVENT, + FollowRequestApprovedEvent, + isFollowRequestApprovedEvent, +} from '@app/events'; +import { Controller } from '@nestjs/common'; +import { Ctx, MessagePattern, Payload, RmqContext, RpcException } from '@nestjs/microservices'; +import { NotificationCategory, NotificationQueryDto } from '../../../src/modules/notifications/dto/notification-query.dto'; +import { NotificationsService } from '../../../src/modules/notifications/notifications.service'; + +@Controller() +export class NotificationConsumer { + constructor(private readonly notifications: NotificationsService) {} + + @MessagePattern(NOTIFICATION_RPC.list) + async list(@Payload() input: NotificationListRequest, @Ctx() context: RmqContext) { + return this.rpc(context, () => this.notifications.getMine( + input.recipientId, + input.query as unknown as NotificationQueryDto, + )); + } + + @MessagePattern(NOTIFICATION_RPC.unreadCount) + async unreadCount(@Payload() input: NotificationUnreadCountRequest, @Ctx() context: RmqContext) { + return this.rpc(context, () => this.notifications.getUnreadCountByCategory( + input.recipientId, + input.category as NotificationCategory | undefined, + )); + } + + @MessagePattern(NOTIFICATION_RPC.markRead) + async markRead(@Payload() input: NotificationMarkReadRequest, @Ctx() context: RmqContext) { + return this.rpc(context, async () => { + const result = await this.notifications.markRead(input.recipientId, input.notificationId); + const unreadCounts = await this.notifications.getUnreadCounts(input.recipientId); + return { result, realtime: { + recipientId: input.recipientId, + unreadCount: unreadCounts.total, + unreadCounts, + } }; + }, true); + } + + @MessagePattern(NOTIFICATION_RPC.markAllRead) + async markAllRead(@Payload() input: NotificationMarkAllReadRequest, @Ctx() context: RmqContext) { + return this.rpc(context, async () => { + const result = await this.notifications.markAllRead(input.recipientId); + const unreadCounts = await this.notifications.getUnreadCounts(input.recipientId); + return { result, realtime: { + recipientId: input.recipientId, + unreadCount: unreadCounts.total, + unreadCounts, + } }; + }, true); + } + + @MessagePattern(FOLLOW_REQUEST_APPROVED_EVENT) + async followRequestApproved( + @Payload() event: FollowRequestApprovedEvent, + @Ctx() context: RmqContext, + ) { + if (!isFollowRequestApprovedEvent(event)) { + return this.reject(context, new Error('Invalid follow.request.approved contract')); + } + + const metadata = this.metadata(context); + try { + const notification = await withLimitedRetry( + () => this.notifications.createFollowRequestApprovedNotification({ + ...event.payload, + eventId: event.eventId, + }), + 3, + 25, + ); + const unreadCounts = await this.notifications.getUnreadCounts(event.payload.recipientId); + this.ack(context); + this.log('info', 'follow.request.approved consumed', metadata, { eventId: event.eventId }); + return { + result: notification, + realtime: { + recipientId: event.payload.recipientId, + notification, + unreadCount: unreadCounts.total, + unreadCounts, + }, + }; + } catch (error) { + return this.reject(context, error, metadata, event.eventId); + } + } + + private async rpc( + context: RmqContext, + operation: () => Promise>, + alreadyWrapped = false, + ): Promise> { + try { + const value = await operation(); + this.ack(context); + return (alreadyWrapped ? value : { result: value }) as NotificationRpcResponse; + } catch (error) { + this.ack(context); + throw new RpcException(error instanceof Error ? error.message : 'Notification operation failed'); + } + } + + private ack(context: RmqContext): void { + context.getChannelRef().ack(context.getMessage()); + } + + private reject( + context: RmqContext, + error: unknown, + metadata = this.metadata(context), + eventId?: string, + ): never { + context.getChannelRef().nack(context.getMessage(), false, false); + const message = error instanceof Error ? error.message : 'Notification event failed'; + this.log('error', message, metadata, { eventId }); + throw new RpcException(message); + } + + private metadata(context: RmqContext): MessageContext { + const message = context.getMessage() as { properties?: { headers?: Record; correlationId?: string } }; + const headers = message.properties?.headers ?? {}; + return { + correlationId: String(headers.correlationId ?? message.properties?.correlationId ?? ''), + requestId: String(headers.requestId ?? ''), + }; + } + + private log( + level: 'info' | 'error', + message: string, + context: MessageContext, + extra: Record, + ): void { + const entry = JSON.stringify({ + timestamp: new Date().toISOString(), level, service: 'notification-service', message, ...context, ...extra, + }); + if (level === 'error') console.error(entry); + else console.log(entry); + } +} diff --git a/apps/notification-service/src/rabbitmq-topology.spec.ts b/apps/notification-service/src/rabbitmq-topology.spec.ts new file mode 100644 index 0000000..3052c65 --- /dev/null +++ b/apps/notification-service/src/rabbitmq-topology.spec.ts @@ -0,0 +1,29 @@ +import { NOTIFICATION_RMQ_DLQ, NOTIFICATION_RMQ_QUEUE } from '@app/contracts'; +import { assertNotificationTopology, notificationQueueArguments } from './rabbitmq-topology'; + +describe('RabbitMQ notification adapter', () => { + it('asserts durable main and dead-letter queues', async () => { + const channel = { + assertQueue: jest.fn().mockResolvedValue({}), + close: jest.fn().mockResolvedValue(undefined), + }; + const connection = { + createChannel: jest.fn().mockResolvedValue(channel), + close: jest.fn().mockResolvedValue(undefined), + }; + + await assertNotificationTopology('amqp://rabbit', jest.fn().mockResolvedValue(connection) as never); + + expect(channel.assertQueue).toHaveBeenNthCalledWith(1, NOTIFICATION_RMQ_DLQ, { + durable: true, autoDelete: false, exclusive: false, + }); + expect(channel.assertQueue).toHaveBeenNthCalledWith(2, NOTIFICATION_RMQ_QUEUE, { + durable: true, + autoDelete: false, + exclusive: false, + arguments: notificationQueueArguments, + }); + expect(channel.close).toHaveBeenCalled(); + expect(connection.close).toHaveBeenCalled(); + }); +}); diff --git a/apps/notification-service/src/rabbitmq-topology.ts b/apps/notification-service/src/rabbitmq-topology.ts new file mode 100644 index 0000000..687cac4 --- /dev/null +++ b/apps/notification-service/src/rabbitmq-topology.ts @@ -0,0 +1,35 @@ +import { NOTIFICATION_RMQ_DLQ, NOTIFICATION_RMQ_QUEUE } from '@app/contracts'; +import { ChannelModel, connect } from 'amqplib'; +import { ResolvedRabbitMqConfig } from '@app/common'; + +type Connect = (url: string, options?: Record) => Promise; + +export const notificationQueueArguments = { + 'x-dead-letter-exchange': '', + 'x-dead-letter-routing-key': NOTIFICATION_RMQ_DLQ, +}; + +export async function assertNotificationTopology( + config: string | ResolvedRabbitMqConfig, + connectToRabbit: Connect = connect as unknown as Connect, +): Promise { + const url = typeof config === 'string' ? config : config.url; + const options = typeof config === 'string' ? undefined : config.socketOptions.connectionOptions; + const connection = await connectToRabbit(url, options); + try { + const channel = await connection.createChannel(); + try { + await channel.assertQueue(NOTIFICATION_RMQ_DLQ, { durable: true, autoDelete: false, exclusive: false }); + await channel.assertQueue(NOTIFICATION_RMQ_QUEUE, { + durable: true, + autoDelete: false, + exclusive: false, + arguments: notificationQueueArguments, + }); + } finally { + await channel.close(); + } + } finally { + await connection.close(); + } +} diff --git a/apps/notification-service/src/service-state.ts b/apps/notification-service/src/service-state.ts new file mode 100644 index 0000000..32351f6 --- /dev/null +++ b/apps/notification-service/src/service-state.ts @@ -0,0 +1,6 @@ +export const NOTIFICATION_SERVICE_STATE = Symbol('NOTIFICATION_SERVICE_STATE'); + +export interface NotificationServiceState { + rabbitmqReady: boolean; + draining: boolean; +} diff --git a/apps/notification-service/tsconfig.app.json b/apps/notification-service/tsconfig.app.json new file mode 100644 index 0000000..288004a --- /dev/null +++ b/apps/notification-service/tsconfig.app.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "declaration": false, + "outDir": "../../dist/apps/notification-service" + }, + "include": ["src/**/*.ts", "../../src/modules/notifications/**/*.ts", "../../src/common/**/*.ts", "../../libs/**/*.ts"], + "exclude": ["node_modules", "dist", "test", "**/*.spec.ts"] +} diff --git a/docker-compose.microservices.yml b/docker-compose.microservices.yml new file mode 100644 index 0000000..58fe437 --- /dev/null +++ b/docker-compose.microservices.yml @@ -0,0 +1,111 @@ +# Local phase-1 stack only. Never reuse guest/guest or this topology as production deployment configuration. +services: + api-gateway: + build: + context: . + dockerfile: Dockerfile.microservices + target: api-gateway + ports: + - "4000:4000" + environment: + NODE_ENV: development + PORT: 4000 + HOST: 0.0.0.0 + MONGODB_URI: mongodb://mongodb:27017/oudelaa?replicaSet=rs0 + REDIS_ENABLED: "true" + REDIS_URL: redis://redis:6379 + REDIS_SOCKET_ADAPTER_ENABLED: "true" + QUEUE_ENABLED: "true" + RABBITMQ_URL: amqp://guest:guest@rabbitmq:5672 + NOTIFICATION_SERVICE_ENABLED: "true" + OUTBOX_DISPATCHER_ENABLED: "true" + JWT_ACCESS_SECRET: ${JWT_ACCESS_SECRET:?Set JWT_ACCESS_SECRET in the selected local env file} + JWT_REFRESH_SECRET: ${JWT_REFRESH_SECRET:?Set JWT_REFRESH_SECRET in the selected local env file} + REFRESH_TOKEN_HASH_SECRET: ${REFRESH_TOKEN_HASH_SECRET:-} + SUPERADMIN_EMAIL: ${SUPERADMIN_EMAIL:?Set SUPERADMIN_EMAIL in the selected local env file} + SUPERADMIN_ACCESS_SECRET: ${SUPERADMIN_ACCESS_SECRET:?Set SUPERADMIN_ACCESS_SECRET in the selected local env file} + SUPERADMIN_REFRESH_SECRET: ${SUPERADMIN_REFRESH_SECRET:?Set SUPERADMIN_REFRESH_SECRET in the selected local env file} + STORAGE_PROVIDER: local + STORAGE_BASE_PATH: uploads + depends_on: + mongodb: + condition: service_healthy + redis: + condition: service_healthy + rabbitmq: + condition: service_healthy + notification-service: + condition: service_healthy + healthcheck: + test: ["CMD", "wget", "-q", "--spider", "http://127.0.0.1:4000/api/v1/health/ready"] + interval: 10s + timeout: 5s + retries: 12 + volumes: + - gateway_uploads:/app/uploads + restart: unless-stopped + + notification-service: + build: + context: . + dockerfile: Dockerfile.microservices + target: notification-service + ports: + - "127.0.0.1:4001:4001" + environment: + NODE_ENV: development + NOTIFICATION_SERVICE_PORT: 4001 + NOTIFICATION_MONGODB_URI: mongodb://mongodb:27017/oudelaa?replicaSet=rs0 + RABBITMQ_URL: amqp://guest:guest@rabbitmq:5672 + NOTIFICATION_RMQ_PREFETCH: 10 + MONGODB_AUTO_INDEX: "true" + depends_on: + mongodb: + condition: service_healthy + rabbitmq: + condition: service_healthy + healthcheck: + test: ["CMD", "wget", "-q", "--spider", "http://127.0.0.1:4001/health/ready"] + interval: 10s + timeout: 5s + retries: 12 + restart: unless-stopped + + rabbitmq: + image: rabbitmq:3.13-management-alpine + ports: + - "127.0.0.1:15672:15672" + healthcheck: + test: ["CMD-SHELL", "rabbitmq-diagnostics -q check_running && rabbitmq-diagnostics -q check_port_connectivity"] + interval: 5s + timeout: 5s + retries: 12 + volumes: + - rabbitmq_data:/var/lib/rabbitmq + + redis: + image: redis:7-alpine + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 12 + volumes: + - redis_data:/data + + mongodb: + image: mongo:8 + command: ["mongod", "--replSet", "rs0", "--bind_ip_all"] + healthcheck: + test: ["CMD", "mongosh", "--quiet", "--eval", "try { if (rs.status().ok === 1) quit(0); quit(1) } catch (e) { rs.initiate({_id:'rs0',members:[{_id:0,host:'mongodb:27017'}]}); quit(0) }"] + interval: 5s + timeout: 5s + retries: 12 + volumes: + - mongodb_data:/data/db + +volumes: + gateway_uploads: + rabbitmq_data: + redis_data: + mongodb_data: diff --git a/docs/MICROSERVICES_MIGRATION.md b/docs/MICROSERVICES_MIGRATION.md new file mode 100644 index 0000000..f70b8ef --- /dev/null +++ b/docs/MICROSERVICES_MIGRATION.md @@ -0,0 +1,114 @@ +# Oudelaa microservices migration + +## Current architecture + +The backend is a NestJS modular monolith backed by MongoDB. Redis is used for cache, +BullMQ jobs, and the optional Socket.IO adapter. Domain modules call the in-process +`NotificationsService`. Public HTTP routes use the global `/api/v1` prefix and the +notifications Socket.IO namespace is owned by the same application. + +## Phase 1 boundary + +- `apps/api-gateway` is the new monorepo entry point. It deliberately bootstraps the + existing `src/AppModule`, so the application remains runnable while modules are moved + incrementally. +- `apps/notification-service` owns notification persistence, queries, unread counts, + read mutations, and the `follow.request.approved` consumer. +- The four existing Flutter notification endpoints remain on the gateway. The gateway + delegates them over Nest RMQ only when `NOTIFICATION_SERVICE_ENABLED=true`; otherwise + it uses the original in-process service. +- Socket.IO authentication, namespace, rooms, and public event names remain in the + gateway. Internal RMQ responses include realtime data which the gateway emits. +- Other notification producers stay in-process in phase 1. They can migrate one event + at a time in later phases. + +## Reliability and topology + +The main durable queue is `oudelaa.notifications`. Rejected messages are dead-lettered +to the durable `oudelaa.notifications.dlq` queue through the default exchange. Consumers +use manual acknowledgement and a prefetch limit. Event handling retries three times +inside one delivery, then rejects without requeue. Gateway calls have a bounded timeout, +one configurable retry, and a small circuit breaker. Rabbit message headers carry +`correlationId` and `requestId`; logs are JSON records containing both. + +Follow-request approval is the first true transactional-outbox flow. The request state, +follow relationship, user counters, and deterministic outbox event are committed in one +MongoDB transaction. No BullMQ or RabbitMQ call occurs inside that transaction. When both +phase-1 flags are enabled, a polling dispatcher atomically leases pending/failed events, +waits for the Notification Service RPC response, and only then marks the event processed. +Timeouts use bounded exponential backoff and eventually move the event to `dead`; expired +`processing` leases are recovered after restart. `OutboxService.listByStatuses()` and +`retryEvent()` are the intentionally non-public administrative surface in this phase. + +`eventId` is stored on notifications under a unique partial index. The consumer checks +it before insert and also handles Mongo duplicate-key races. The existing unique index on +recipient/type/reference remains as an additional guard for follow approval records. + +## Planned migration + +1. Phase 1: monorepo shell, contracts/events/common libraries, gateway compatibility + facade, and Notification Service. +2. Phase 2: move Auth and Users behind versioned internal contracts; centralize JWT + verification primitives in `libs/auth`; add tracing/metrics in `libs/observability`. +3. Phase 3: extract Posts and Media; replace direct notification calls with outbox events. +4. Phase 4: extract Feed and Search read models, consuming post/user events. +5. Phase 5: extract Moderation, then stabilize cross-service workflows and data ownership. +6. Phase 6: remove transitional imports and local fallbacks after traffic comparison, + replay tests, and a controlled cutover. + +At every phase the gateway retains public paths and response envelopes. Internal event +contracts evolve additively and are versioned before any breaking change. + +## Configuration and operating modes + +`NOTIFICATION_SERVICE_ENABLED=false` and `OUTBOX_DISPATCHER_ENABLED=false` remain the +safe defaults. In that mode the gateway does not instantiate or connect an RMQ client; +the in-process notification path and public Flutter contracts remain unchanged. The +standalone-Mongo fallback exists only for this disabled legacy transition mode. + +The reliable phase-1 path requires both flags to be `true`. Dispatcher interval, batch, +lease, attempt limit, and bounded backoff are configured by `OUTBOX_*`. RabbitMQ accepts +an injected `RABBITMQ_URL` or separate host, port, username, password, and vhost values. +TLS supports `amqps://`, CA/client material injected as base64 or mounted paths, peer +verification, and SNI. Passwords and URL userinfo are never logged. + +`docker-compose.microservices.yml` is local-only. It deliberately uses guest/guest inside +the private Compose network, exposes management only on loopback, and runs a single-node +Mongo replica set so transactions are exercised locally. It is not a production template. +Queue names remain `oudelaa.notifications` and `oudelaa.notifications.dlq` to avoid an +undeclared migration; both are durable, non-exclusive, and non-auto-delete. RPC messages +are persistent, and consumers use manual acknowledgement with a prefetch limit. + +## Production prerequisites + +- Managed RabbitMQ (or an operated HA cluster), a dedicated vhost, non-guest + least-privilege credentials, and TLS/amqps. Inject credentials and certificates from + the deployment secret store; never copy local Compose values. +- A transaction-capable MongoDB replica set or sharded cluster. Gateway startup rejects + dispatcher activation against standalone MongoDB. +- Production Redis, service-specific secrets, persistent storage where applicable, and + network policy that permits RMQ/Mongo/Redis access only from the services. +- Liveness probes at `/api/v1/health` and `/health/live`; readiness probes at + `/api/v1/health/ready` and `/health/ready`. A temporary RMQ outage fails readiness but + does not fail liveness. RabbitMQ management must not be publicly exposed. +- Alerts for pending/failed/dead outbox counts, oldest event age, dispatcher failures, + notification DLQ depth, RMQ connectivity, and duplicate-key activity. + +### Controlled activation and rollback + +1. Provision production RabbitMQ with TLS, a dedicated vhost, and least-privilege user. +2. Confirm MongoDB is a replica set/mongos and Redis is production-ready. +3. Deploy Notification Service with injected secrets and verify liveness/readiness. +4. Deploy Gateway with both phase-1 flags still `false`; verify the monolith path. +5. Inject the same RMQ endpoint/trust settings into Gateway without logging them. +6. Enable both flags on one canary Gateway replica. +7. Approve a controlled private follow and observe outbox + `pending -> processing -> processed`, one notification, and one Socket emission. +8. Monitor retries, dead events, DLQ, readiness, latency, and duplicate-key metrics. +9. Expand the flags gradually; atomic leases make multiple Gateway replicas safe. +10. Roll back by setting both flags to `false`. Do not delete pending events or queues; + diagnose or retry them before the next canary, then drain Notification Service. + +Notification Service still uses the same Mongo database and the same `notifications` +collection as Gateway. This is a transitional shared-database model, not independent data +ownership; physical separation belongs to a later migration phase. diff --git a/jest.config.js b/jest.config.js index c3ccbc7..e451c04 100644 --- a/jest.config.js +++ b/jest.config.js @@ -7,6 +7,11 @@ module.exports = { transform: { '^.+\\.(t|j)s$': 'ts-jest', }, + moduleNameMapper: { + '^@app/contracts$': '/libs/contracts/src', + '^@app/events$': '/libs/events/src', + '^@app/common$': '/libs/common/src', + }, collectCoverageFrom: ['src/**/*.(t|j)s'], coverageDirectory: './coverage', coverageReporters: ['text', 'json-summary', 'lcov'], diff --git a/libs/common/src/index.ts b/libs/common/src/index.ts new file mode 100644 index 0000000..1fdc990 --- /dev/null +++ b/libs/common/src/index.ts @@ -0,0 +1,2 @@ +export * from './retry'; +export * from './rabbitmq'; diff --git a/libs/common/src/rabbitmq.spec.ts b/libs/common/src/rabbitmq.spec.ts new file mode 100644 index 0000000..457d4cf --- /dev/null +++ b/libs/common/src/rabbitmq.spec.ts @@ -0,0 +1,39 @@ +import { resolveRabbitMqConfig } from './rabbitmq'; + +const source = (values: Record) => ({ get: (key: string) => values[key] }); + +describe('RabbitMQ production configuration', () => { + it('rejects guest credentials and plaintext in production microservices', () => { + expect(() => resolveRabbitMqConfig(source({ NODE_ENV: 'production', RABBITMQ_URL: 'amqps://guest:guest@broker/vhost' }), 'gateway', true)).toThrow('non-guest'); + expect(() => resolveRabbitMqConfig(source({ NODE_ENV: 'production', RABBITMQ_URL: 'amqp://service:strong-password@broker/vhost' }), 'gateway', true)).toThrow('TLS/amqps'); + }); + + it('supports secure URLs and separate TLS credentials without exposing userinfo', () => { + const direct = resolveRabbitMqConfig(source({ + NODE_ENV: 'production', RABBITMQ_URL: 'amqps://service:strong-password@broker.example/vhost', + RABBITMQ_SERVERNAME: 'broker.example', RABBITMQ_TLS_REJECT_UNAUTHORIZED: 'true', + }), 'notification-service', true); + expect(direct.sanitizedEndpoint).toBe('amqps://broker.example:5671/vhost'); + expect(direct.sanitizedEndpoint).not.toContain('strong-password'); + expect(direct.socketOptions.connectionOptions).toEqual(expect.objectContaining({ rejectUnauthorized: true, servername: 'broker.example' })); + + const split = resolveRabbitMqConfig(source({ + RABBITMQ_HOST: 'rabbit', RABBITMQ_PORT: '5671', RABBITMQ_USERNAME: 'service', + RABBITMQ_PASSWORD: 'password', RABBITMQ_VHOST: 'oudelaa', RABBITMQ_TLS_ENABLED: 'true', + }), 'gateway', false); + expect(split.url).toBe('amqps://service:password@rabbit:5671/oudelaa'); + + const urlWithSeparateCredentials = resolveRabbitMqConfig(source({ + RABBITMQ_URL: 'amqps://rabbit:5671/oudelaa', RABBITMQ_USERNAME: 'service', + RABBITMQ_PASSWORD: 'password', + }), 'gateway', false); + expect(urlWithSeparateCredentials.url).toBe('amqps://service:password@rabbit:5671/oudelaa'); + }); + + it('requires client certificate and key as a pair', () => { + expect(() => resolveRabbitMqConfig(source({ + RABBITMQ_URL: 'amqps://service:password@rabbit/vhost', + RABBITMQ_CLIENT_CERT_BASE64: Buffer.from('cert').toString('base64'), + }), 'gateway', false)).toThrow('configured together'); + }); +}); diff --git a/libs/common/src/rabbitmq.ts b/libs/common/src/rabbitmq.ts new file mode 100644 index 0000000..1d61284 --- /dev/null +++ b/libs/common/src/rabbitmq.ts @@ -0,0 +1,95 @@ +import { readFileSync } from 'fs'; + +export interface RabbitConfigSource { + get(key: string): string | undefined; +} + +export interface ResolvedRabbitMqConfig { + url: string; + sanitizedEndpoint: string; + socketOptions: { + heartbeatIntervalInSeconds: number; + reconnectTimeInSeconds: number; + connectionOptions: Record; + }; + tlsEnabled: boolean; +} + +const bool = (value: string | undefined, fallback = false) => value === undefined ? fallback : value === 'true'; + +function material(source: RabbitConfigSource, base64Key: string, pathKey: string): Buffer | undefined { + const encoded = source.get(base64Key); + if (encoded) return Buffer.from(encoded, 'base64'); + const path = source.get(pathKey); + return path ? readFileSync(path) : undefined; +} + +export function resolveRabbitMqConfig( + source: RabbitConfigSource, + serviceName: string, + requireProductionSafety: boolean, +): ResolvedRabbitMqConfig { + const explicitUrl = source.get('RABBITMQ_URL'); + const tlsEnabled = bool(source.get('RABBITMQ_TLS_ENABLED'), explicitUrl?.startsWith('amqps://')); + const protocol = tlsEnabled ? 'amqps' : 'amqp'; + const host = source.get('RABBITMQ_HOST') || '127.0.0.1'; + const port = source.get('RABBITMQ_PORT') || (tlsEnabled ? '5671' : '5672'); + const username = source.get('RABBITMQ_USERNAME') || 'guest'; + const password = source.get('RABBITMQ_PASSWORD') || 'guest'; + const vhost = source.get('RABBITMQ_VHOST') || '/'; + let url = explicitUrl + ? (tlsEnabled ? explicitUrl.replace(/^amqp:\/\//, 'amqps://') : explicitUrl) + : `${protocol}://${encodeURIComponent(username)}:${encodeURIComponent(password)}@${host}:${port}/${vhost === '/' ? '' : encodeURIComponent(vhost.replace(/^\//, ''))}`; + let parsed: URL; + try { parsed = new URL(url); } + catch { throw new Error('RabbitMQ configuration contains an invalid URL'); } + if (explicitUrl && !parsed.username && source.get('RABBITMQ_USERNAME')) { + parsed.username = username; + parsed.password = password; + url = parsed.toString(); + } + const configuredUser = decodeURIComponent(parsed.username || username); + const configuredPassword = decodeURIComponent(parsed.password || password); + const production = source.get('NODE_ENV') === 'production'; + const tlsRequired = bool(source.get('RABBITMQ_TLS_REQUIRED'), true); + if (production && requireProductionSafety) { + if (!configuredUser || !configuredPassword || configuredUser === 'guest' || configuredPassword === 'guest') { + throw new Error('RabbitMQ production credentials must be non-guest and non-empty'); + } + if (tlsRequired && parsed.protocol !== 'amqps:' && !tlsEnabled) { + throw new Error('RabbitMQ TLS/amqps is required for production microservices'); + } + } + + const connectionOptions: Record = { + timeout: Number(source.get('RABBITMQ_CONNECTION_TIMEOUT_MS') || 10_000), + clientProperties: { connection_name: `${serviceName}:${source.get('HOSTNAME') || process.pid}` }, + }; + if (tlsEnabled || parsed.protocol === 'amqps:') { + connectionOptions.rejectUnauthorized = bool(source.get('RABBITMQ_TLS_REJECT_UNAUTHORIZED'), true); + const servername = source.get('RABBITMQ_SERVERNAME'); + if (servername) connectionOptions.servername = servername; + const ca = material(source, 'RABBITMQ_CA_CERT_BASE64', 'RABBITMQ_CA_CERT_PATH'); + const cert = material(source, 'RABBITMQ_CLIENT_CERT_BASE64', 'RABBITMQ_CLIENT_CERT_PATH'); + const key = material(source, 'RABBITMQ_CLIENT_KEY_BASE64', 'RABBITMQ_CLIENT_KEY_PATH'); + if (ca) connectionOptions.ca = [ca]; + if (cert) connectionOptions.cert = cert; + if (key) connectionOptions.key = key; + if ((cert && !key) || (!cert && key)) throw new Error('RabbitMQ client certificate and key must be configured together'); + } + + return { + url, + sanitizedEndpoint: `${parsed.protocol}//${parsed.hostname}:${parsed.port || (parsed.protocol === 'amqps:' ? '5671' : '5672')}${parsed.pathname}`, + tlsEnabled: tlsEnabled || parsed.protocol === 'amqps:', + socketOptions: { + heartbeatIntervalInSeconds: Number(source.get('RABBITMQ_HEARTBEAT_SECONDS') || 10), + reconnectTimeInSeconds: Number(source.get('RABBITMQ_RECONNECT_SECONDS') || 5), + connectionOptions, + }, + }; +} + +export function processEnvSource(): RabbitConfigSource { + return { get: (key) => process.env[key] }; +} diff --git a/libs/common/src/retry.ts b/libs/common/src/retry.ts new file mode 100644 index 0000000..41bf314 --- /dev/null +++ b/libs/common/src/retry.ts @@ -0,0 +1,18 @@ +export async function withLimitedRetry( + operation: () => Promise, + maxAttempts: number, + backoffMs = 25, +): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + try { + return await operation(); + } catch (error) { + lastError = error; + if (attempt < maxAttempts) { + await new Promise((resolve) => setTimeout(resolve, backoffMs * attempt)); + } + } + } + throw lastError; +} diff --git a/libs/common/tsconfig.lib.json b/libs/common/tsconfig.lib.json new file mode 100644 index 0000000..15e75cf --- /dev/null +++ b/libs/common/tsconfig.lib.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { "declaration": true, "outDir": "../../dist/libs/common" }, + "include": ["src/**/*.ts"] +} diff --git a/libs/contracts/src/index.ts b/libs/contracts/src/index.ts new file mode 100644 index 0000000..57abe50 --- /dev/null +++ b/libs/contracts/src/index.ts @@ -0,0 +1 @@ +export * from './notifications.contract'; diff --git a/libs/contracts/src/notifications.contract.ts b/libs/contracts/src/notifications.contract.ts new file mode 100644 index 0000000..0eafd4f --- /dev/null +++ b/libs/contracts/src/notifications.contract.ts @@ -0,0 +1,45 @@ +export const NOTIFICATION_RMQ_QUEUE = 'oudelaa.notifications'; +export const NOTIFICATION_RMQ_DLQ = 'oudelaa.notifications.dlq'; + +export const NOTIFICATION_RPC = { + list: 'notifications.list', + unreadCount: 'notifications.unread-count', + markRead: 'notifications.mark-read', + markAllRead: 'notifications.mark-all-read', +} as const; + +export interface MessageContext { + correlationId: string; + requestId: string; +} + +export interface NotificationListRequest { + recipientId: string; + query: Record; +} + +export interface NotificationUnreadCountRequest { + recipientId: string; + category?: string; +} + +export interface NotificationMarkReadRequest { + recipientId: string; + notificationId: string; +} + +export interface NotificationMarkAllReadRequest { + recipientId: string; +} + +export interface NotificationRealtimeUpdate { + recipientId: string; + notification?: unknown; + unreadCount: number; + unreadCounts?: Record; +} + +export interface NotificationRpcResponse { + result: T; + realtime?: NotificationRealtimeUpdate; +} diff --git a/libs/contracts/tsconfig.lib.json b/libs/contracts/tsconfig.lib.json new file mode 100644 index 0000000..9171edb --- /dev/null +++ b/libs/contracts/tsconfig.lib.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { "declaration": true, "outDir": "../../dist/libs/contracts" }, + "include": ["src/**/*.ts"] +} diff --git a/libs/events/src/follow-request-approved.event.ts b/libs/events/src/follow-request-approved.event.ts new file mode 100644 index 0000000..c5f4a0c --- /dev/null +++ b/libs/events/src/follow-request-approved.event.ts @@ -0,0 +1,34 @@ +export const FOLLOW_REQUEST_APPROVED_EVENT = 'follow.request.approved' as const; + +export interface FollowRequestApprovedEvent { + eventId: string; + eventType: typeof FOLLOW_REQUEST_APPROVED_EVENT; + occurredAt: string; + payload: { + requestId: string; + actorId: string; + recipientId: string; + }; +} + +export function isFollowRequestApprovedEvent(value: unknown): value is FollowRequestApprovedEvent { + if (!value || typeof value !== 'object') return false; + const event = value as Partial; + const payload = event.payload as Partial | undefined; + return event.eventType === FOLLOW_REQUEST_APPROVED_EVENT + && isNonEmptyString(event.eventId) + && isIsoDateString(event.occurredAt) + && isNonEmptyString(payload?.requestId) + && isNonEmptyString(payload.actorId) + && isNonEmptyString(payload.recipientId); +} + +const isNonEmptyString = (value: unknown): value is string => + typeof value === 'string' && value.trim().length > 0; + +const isIsoDateString = (value: unknown): value is string => { + if (typeof value !== 'string') return false; + const timestamp = Date.parse(value); + if (Number.isNaN(timestamp)) return false; + return new Date(timestamp).toISOString() === value; +}; diff --git a/libs/events/src/index.ts b/libs/events/src/index.ts new file mode 100644 index 0000000..19097da --- /dev/null +++ b/libs/events/src/index.ts @@ -0,0 +1 @@ +export * from './follow-request-approved.event'; diff --git a/libs/events/tsconfig.lib.json b/libs/events/tsconfig.lib.json new file mode 100644 index 0000000..219131b --- /dev/null +++ b/libs/events/tsconfig.lib.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { "declaration": true, "outDir": "../../dist/libs/events" }, + "include": ["src/**/*.ts"] +} diff --git a/nest-cli.json b/nest-cli.json index f9aa683..95bc322 100644 --- a/nest-cli.json +++ b/nest-cli.json @@ -1,8 +1,59 @@ { "$schema": "https://json.schemastore.org/nest-cli", "collection": "@nestjs/schematics", - "sourceRoot": "src", + "monorepo": true, + "root": "apps/api-gateway", + "sourceRoot": "apps/api-gateway/src", "compilerOptions": { "deleteOutDir": true + }, + "projects": { + "api-gateway": { + "type": "application", + "root": "apps/api-gateway", + "entryFile": "main", + "sourceRoot": "apps/api-gateway/src", + "compilerOptions": { + "webpack": false, + "tsConfigPath": "apps/api-gateway/tsconfig.app.json" + } + }, + "notification-service": { + "type": "application", + "root": "apps/notification-service", + "entryFile": "main", + "sourceRoot": "apps/notification-service/src", + "compilerOptions": { + "webpack": true, + "tsConfigPath": "apps/notification-service/tsconfig.app.json" + } + }, + "contracts": { + "type": "library", + "root": "libs/contracts", + "entryFile": "index", + "sourceRoot": "libs/contracts/src", + "compilerOptions": { + "tsConfigPath": "libs/contracts/tsconfig.lib.json" + } + }, + "events": { + "type": "library", + "root": "libs/events", + "entryFile": "index", + "sourceRoot": "libs/events/src", + "compilerOptions": { + "tsConfigPath": "libs/events/tsconfig.lib.json" + } + }, + "common": { + "type": "library", + "root": "libs/common", + "entryFile": "index", + "sourceRoot": "libs/common/src", + "compilerOptions": { + "tsConfigPath": "libs/common/tsconfig.lib.json" + } + } } } diff --git a/package-lock.json b/package-lock.json index 1dfa5a1..b86ba80 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,9 +8,6 @@ "name": "oudelaa-backend", "version": "1.0.0", "license": "UNLICENSED", - "engines": { - "node": ">=20 <25" - }, "dependencies": { "@aws-sdk/client-s3": "^3.1041.0", "@aws-sdk/lib-storage": "^3.1041.0", @@ -19,6 +16,7 @@ "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.28", "@nestjs/jwt": "^11.0.2", + "@nestjs/microservices": "^11.1.28", "@nestjs/mongoose": "^11.0.4", "@nestjs/passport": "^11.0.5", "@nestjs/platform-express": "^11.1.28", @@ -27,6 +25,8 @@ "@nestjs/websockets": "^11.1.28", "@socket.io/redis-adapter": "^8.3.0", "@types/passport-google-oauth20": "^2.0.17", + "amqp-connection-manager": "^4.1.14", + "amqplib": "^0.10.8", "bcrypt": "^6.0.0", "bullmq": "^5.76.5", "class-transformer": "^0.5.1", @@ -50,6 +50,7 @@ "@nestjs/cli": "^11.0.24", "@nestjs/schematics": "^11.1.0", "@nestjs/testing": "^11.1.28", + "@types/amqplib": "^0.10.7", "@types/bcrypt": "^5.0.2", "@types/compression": "^1.8.1", "@types/express": "^4.17.21", @@ -73,6 +74,9 @@ "tsconfig-paths": "^4.2.0", "typescript": "^5.6.2", "typescript-eslint": "^8.64.0" + }, + "engines": { + "node": ">=20 <25" } }, "node_modules/@angular-devkit/core": { @@ -3080,6 +3084,64 @@ } } }, + "node_modules/@nestjs/microservices": { + "version": "11.1.28", + "resolved": "https://registry.npmjs.org/@nestjs/microservices/-/microservices-11.1.28.tgz", + "integrity": "sha512-8uRs6/UrhXvd8YCrYKcNUwWA7b8jcbYH03WBKWJ2A3bc6+WcHA6yXq7Px30yemAGBtIIMXkHeL88dO2usVI5zg==", + "license": "MIT", + "dependencies": { + "iterare": "1.2.1", + "tslib": "2.8.1" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/nest" + }, + "peerDependencies": { + "@grpc/grpc-js": "*", + "@nestjs/common": "^11.0.0", + "@nestjs/core": "^11.0.0", + "@nestjs/websockets": "^11.0.0", + "amqp-connection-manager": "*", + "amqplib": "*", + "cache-manager": "*", + "ioredis": "*", + "kafkajs": "*", + "mqtt": "*", + "nats": "*", + "reflect-metadata": "^0.1.12 || ^0.2.0", + "rxjs": "^7.1.0" + }, + "peerDependenciesMeta": { + "@grpc/grpc-js": { + "optional": true + }, + "@nestjs/websockets": { + "optional": true + }, + "amqp-connection-manager": { + "optional": true + }, + "amqplib": { + "optional": true + }, + "cache-manager": { + "optional": true + }, + "ioredis": { + "optional": true + }, + "kafkajs": { + "optional": true + }, + "mqtt": { + "optional": true + }, + "nats": { + "optional": true + } + } + }, "node_modules/@nestjs/mongoose": { "version": "11.0.4", "resolved": "https://registry.npmjs.org/@nestjs/mongoose/-/mongoose-11.0.4.tgz", @@ -4222,6 +4284,16 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/amqplib": { + "version": "0.10.7", + "resolved": "https://registry.npmjs.org/@types/amqplib/-/amqplib-0.10.7.tgz", + "integrity": "sha512-IVj3avf9AQd2nXCx0PGk/OYq7VmHiyNxWFSb5HhU9ATh+i+gHWvVcljFTcTWQ/dyHJCTrzCixde+r/asL2ErDA==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/babel__core": { "version": "7.20.5", "resolved": "https://registry.npmjs.org/@types/babel__core/-/babel__core-7.20.5.tgz", @@ -5234,6 +5306,35 @@ "ajv": "^8.8.2" } }, + "node_modules/amqp-connection-manager": { + "version": "4.1.14", + "resolved": "https://registry.npmjs.org/amqp-connection-manager/-/amqp-connection-manager-4.1.14.tgz", + "integrity": "sha512-1km47dIvEr0HhMUazqovSvNwIlSvDX2APdUpULaINtHpiki1O+cLRaTeXb/jav4OLtH+k6GBXx5gsKOT9kcGKQ==", + "license": "MIT", + "dependencies": { + "promise-breaker": "^6.0.0" + }, + "engines": { + "node": ">=10.0.0", + "npm": ">5.0.0" + }, + "peerDependencies": { + "amqplib": "*" + } + }, + "node_modules/amqplib": { + "version": "0.10.8", + "resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.10.8.tgz", + "integrity": "sha512-Tfn1O9sFgAP8DqeMEpt2IacsVTENBpblB3SqLdn0jK2AeX8iyCvbptBc8lyATT9bQ31MsjVwUSQ1g8f4jHOUfw==", + "license": "MIT", + "dependencies": { + "buffer-more-ints": "~1.0.0", + "url-parse": "~1.5.10" + }, + "engines": { + "node": ">=10" + } + }, "node_modules/ansi-colors": { "version": "4.1.3", "resolved": "https://registry.npmjs.org/ansi-colors/-/ansi-colors-4.1.3.tgz", @@ -5752,6 +5853,12 @@ "integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==", "license": "MIT" }, + "node_modules/buffer-more-ints": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz", + "integrity": "sha512-EMetuGFz5SLsT0QTnXzINh4Ksr+oo4i+UGTXEshiGCQWnsgSs7ZhJ8fzlwQ+OzEMs0MpDAMr1hxnblp5a4vcHg==", + "license": "MIT" + }, "node_modules/bullmq": { "version": "5.76.5", "resolved": "https://registry.npmjs.org/bullmq/-/bullmq-5.76.5.tgz", @@ -10439,6 +10546,12 @@ "url": "https://github.com/chalk/ansi-styles?sponsor=1" } }, + "node_modules/promise-breaker": { + "version": "6.0.0", + "resolved": "https://registry.npmjs.org/promise-breaker/-/promise-breaker-6.0.0.tgz", + "integrity": "sha512-BthzO9yTPswGf7etOBiHCVuugs2N01/Q/94dIPls48z2zCmrnDptUUZzfIb+41xq0MnYZ/BzmOd6ikDR4ibNZA==", + "license": "MIT" + }, "node_modules/prompts": { "version": "2.4.2", "resolved": "https://registry.npmjs.org/prompts/-/prompts-2.4.2.tgz", @@ -10508,6 +10621,12 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/querystringify": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/querystringify/-/querystringify-2.2.0.tgz", + "integrity": "sha512-FIqgj2EUvTa7R50u0rGsyTftzjYmv/a3hO345bZNrqabNqjtgiDMgmo4mkUjd+nzU5oF3dClKqFIPUKybUyqoQ==", + "license": "MIT" + }, "node_modules/range-parser": { "version": "1.3.0", "resolved": "https://registry.npmjs.org/range-parser/-/range-parser-1.3.0.tgz", @@ -10618,6 +10737,12 @@ "node": ">=0.10.0" } }, + "node_modules/requires-port": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz", + "integrity": "sha512-KigOCHcocU3XODJxsu8i/j8T9tzT4adHiecwORRQ0ZZFcp7ahwXuRU1m+yuO90C5ZUyGeGfocHDI14M3L3yDAQ==", + "license": "MIT" + }, "node_modules/resolve": { "version": "1.22.11", "resolved": "https://registry.npmjs.org/resolve/-/resolve-1.22.11.tgz", @@ -12102,6 +12227,16 @@ "punycode": "^2.1.0" } }, + "node_modules/url-parse": { + "version": "1.5.10", + "resolved": "https://registry.npmjs.org/url-parse/-/url-parse-1.5.10.tgz", + "integrity": "sha512-WypcfiRhfeUP9vvF0j6rw0J3hrWrw6iZv3+22h6iRMJ/8z1Tj6XfLP4DsUix5MhMPnXpiHDoKyoZ/bdCkwBCiQ==", + "license": "MIT", + "dependencies": { + "querystringify": "^2.1.1", + "requires-port": "^1.0.0" + } + }, "node_modules/util-deprecate": { "version": "1.0.2", "resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz", diff --git a/package.json b/package.json index 60cf522..23a8bc8 100644 --- a/package.json +++ b/package.json @@ -8,12 +8,17 @@ "node": ">=20 <25" }, "scripts": { - "build": "nest build", - "start": "nest start", - "start:dev": "nest start --watch", - "start:prod": "node dist/main", - "lint": "eslint \"src/**/*.ts\"", - "lint:fix": "eslint \"src/**/*.ts\" --fix", + "build": "nest build api-gateway && nest build notification-service && tsc -p tsconfig.legacy.json && node scripts/copy-legacy-entry.js", + "build:gateway": "nest build api-gateway", + "build:notifications": "nest build notification-service", + "start": "nest start api-gateway", + "start:dev": "nest start api-gateway --watch", + "start:notification-service": "nest start notification-service", + "start:notification-service:dev": "nest start notification-service --watch", + "start:prod": "node dist/main.js", + "start:notification-service:prod": "node dist/apps/notification-service/main", + "lint": "eslint \"{src,apps,libs}/**/*.ts\"", + "lint:fix": "eslint \"{src,apps,libs}/**/*.ts\" --fix", "test": "jest", "test:coverage": "jest --runInBand --coverage && node scripts/check-coverage.js", "test:watch": "jest --watch", @@ -33,6 +38,7 @@ "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.28", "@nestjs/jwt": "^11.0.2", + "@nestjs/microservices": "^11.1.28", "@nestjs/mongoose": "^11.0.4", "@nestjs/passport": "^11.0.5", "@nestjs/platform-express": "^11.1.28", @@ -41,6 +47,8 @@ "@nestjs/websockets": "^11.1.28", "@socket.io/redis-adapter": "^8.3.0", "@types/passport-google-oauth20": "^2.0.17", + "amqp-connection-manager": "^4.1.14", + "amqplib": "^0.10.8", "bcrypt": "^6.0.0", "bullmq": "^5.76.5", "class-transformer": "^0.5.1", @@ -64,6 +72,7 @@ "@nestjs/cli": "^11.0.24", "@nestjs/schematics": "^11.1.0", "@nestjs/testing": "^11.1.28", + "@types/amqplib": "^0.10.7", "@types/bcrypt": "^5.0.2", "@types/compression": "^1.8.1", "@types/express": "^4.17.21", diff --git a/scripts/copy-legacy-entry.js b/scripts/copy-legacy-entry.js new file mode 100644 index 0000000..f0ad19a --- /dev/null +++ b/scripts/copy-legacy-entry.js @@ -0,0 +1,13 @@ +const { copyFileSync, existsSync } = require('node:fs'); +const { join } = require('node:path'); + +const projectRoot = join(__dirname, '..'); +const legacyBundle = join(projectRoot, 'dist', 'legacy', 'src', 'main.js'); +const legacyShim = join(projectRoot, 'scripts', 'legacy-main.js'); +const legacyEntry = join(projectRoot, 'dist', 'main.js'); + +if (!existsSync(legacyBundle)) { + throw new Error(`Legacy monolith bundle not found: ${legacyBundle}`); +} + +copyFileSync(legacyShim, legacyEntry); diff --git a/scripts/legacy-main.js b/scripts/legacy-main.js new file mode 100644 index 0000000..0b3601e --- /dev/null +++ b/scripts/legacy-main.js @@ -0,0 +1,3 @@ +const { bootstrap } = require('./legacy/src/main'); + +void bootstrap(); diff --git a/src/adapter-contracts.spec.ts b/src/adapter-contracts.spec.ts index 916c8d1..12f9f66 100644 --- a/src/adapter-contracts.spec.ts +++ b/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) { diff --git a/src/app.module.ts b/src/app.module.ts index 216de95..9853944 100644 --- a/src/app.module.ts +++ b/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'; diff --git a/src/app.service.spec.ts b/src/app.service.spec.ts index 1c97239..7d54046 100644 --- a/src/app.service.spec.ts +++ b/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, }; }; diff --git a/src/app.service.ts b/src/app.service.ts index cf7f935..f99e0e1 100644 --- a/src/app.service.ts +++ b/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; diff --git a/src/config/validation.schema.ts b/src/config/validation.schema.ts index ec7361f..08e7d95 100644 --- a/src/config/validation.schema.ts +++ b/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'); diff --git a/src/modules/follows/follows.repository.spec.ts b/src/modules/follows/follows.repository.spec.ts new file mode 100644 index 0000000..d975a2b --- /dev/null +++ b/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) => 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) => 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); + }); +}); diff --git a/src/modules/follows/follows.repository.ts b/src/modules/follows/follows.repository.ts index e610585..b85a06b 100644 --- a/src/modules/follows/follows.repository.ts +++ b/src/modules/follows/follows.repository.ts @@ -12,12 +12,17 @@ export class FollowsRepository { private readonly followRequestModel: Model, ) {} - async findOne(followerId: string, followingId: string): Promise { + async findOne( + followerId: string, + followingId: string, + session?: ClientSession, + ): Promise { 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( + operation: (session?: ClientSession) => Promise, + required: boolean, + ): Promise { + 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 { await this.followRequestModel .deleteOne({ @@ -162,16 +188,33 @@ export class FollowsRepository { requestId: string, targetUserId: string, status: 'approved' | 'rejected', + session?: ClientSession, ): Promise { 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 { + 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, diff --git a/src/modules/follows/follows.service.spec.ts b/src/modules/follows/follows.service.spec.ts index 4cd8c32..42e94ba 100644 --- a/src/modules/follows/follows.service.spec.ts +++ b/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) => 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 = { diff --git a/src/modules/follows/follows.service.ts b/src/modules/follows/follows.service.ts index 28236a0..4d5b3ab 100644 --- a/src/modules/follows/follows.service.ts +++ b/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 { - 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, diff --git a/src/modules/notifications/dto/create-notification.dto.ts b/src/modules/notifications/dto/create-notification.dto.ts index 1366348..d703bca 100644 --- a/src/modules/notifications/dto/create-notification.dto.ts +++ b/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; diff --git a/src/modules/notifications/notification-contract.spec.ts b/src/modules/notifications/notification-contract.spec.ts new file mode 100644 index 0000000..bccd00a --- /dev/null +++ b/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(); + }); +}); diff --git a/src/modules/notifications/notification-idempotency.spec.ts b/src/modules/notifications/notification-idempotency.spec.ts new file mode 100644 index 0000000..511c267 --- /dev/null +++ b/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); + }); +}); diff --git a/src/modules/notifications/notification-service.client.ts b/src/modules/notifications/notification-service.client.ts new file mode 100644 index 0000000..dcdc25a --- /dev/null +++ b/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 { + 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 { + 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 { + 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(pattern: string, payload: unknown, requestId?: string, correlationId?: string): Promise { + 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>(pattern, record).pipe( + timeout(Number(this.config.get('NOTIFICATION_RPC_TIMEOUT_MS') ?? 3000)), + retry({ count: Number(this.config.get('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('NOTIFICATION_CIRCUIT_FAILURE_THRESHOLD') ?? 3); + if (this.consecutiveFailures >= threshold) { + this.circuitOpenUntil = Date.now() + Number(this.config.get('NOTIFICATION_CIRCUIT_RESET_MS') ?? 15_000); + this.consecutiveFailures = 0; + } + throw new ServiceUnavailableException('Notification service is unavailable', { cause: error }); + } + } +} diff --git a/src/modules/notifications/notifications.controller.ts b/src/modules/notifications/notifications.controller.ts index 91767b0..c55bec1 100644 --- a/src/modules/notifications/notifications.controller.ts +++ b/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; } } diff --git a/src/modules/notifications/notifications.module.ts b/src/modules/notifications/notifications.module.ts index 5a405ec..543b0b6 100644 --- a/src/modules/notifications/notifications.module.ts +++ b/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 {} diff --git a/src/modules/notifications/notifications.repository.ts b/src/modules/notifications/notifications.repository.ts index 5bed8d1..b76c253 100644 --- a/src/modules/notifications/notifications.repository.ts +++ b/src/modules/notifications/notifications.repository.ts @@ -78,6 +78,13 @@ export class NotificationsRepository { .exec(); } + async findByEventId(eventId: string): Promise { + return this.notificationModel + .findOne({ eventId }) + .populate({ path: 'actorId', select: 'name username stageName avatar isVerified isDisabled' }) + .exec(); + } + async countUnreadByFilter( recipientId: string, filter: FilterQuery = {}, diff --git a/src/modules/notifications/notifications.service.ts b/src/modules/notifications/notifications.service.ts index c0f35cb..c0114c9 100644 --- a/src/modules/notifications/notifications.service.ts +++ b/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; + } } diff --git a/src/modules/notifications/schemas/notification.schema.ts b/src/modules/notifications/schemas/notification.schema.ts index fcd32ba..eb527ac 100644 --- a/src/modules/notifications/schemas/notification.schema.ts +++ b/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 }, { diff --git a/src/modules/outbox/outbox-dispatcher.service.spec.ts b/src/modules/outbox/outbox-dispatcher.service.spec.ts new file mode 100644 index 0000000..9a76fa0 --- /dev/null +++ b/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 = {}) { + 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'); + }); +}); diff --git a/src/modules/outbox/outbox-dispatcher.service.ts b/src/modules/outbox/outbox-dispatcher.service.ts new file mode 100644 index 0000000..fadcd07 --- /dev/null +++ b/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, + @InjectConnection() private readonly connection: Connection, + private readonly config: ConfigService, + private readonly client: NotificationServiceClient, + private readonly logger: AppLoggerService, + ) {} + + async onApplicationBootstrap(): Promise { + if (!this.enabled()) return; + await this.assertTransactionCapableMongo(); + this.schedule(0); + } + + async onModuleDestroy(): Promise { + this.stopped = true; + if (this.timer) clearTimeout(this.timer); + while (this.running) await new Promise((resolve) => setTimeout(resolve, 10)); + } + + async poll(): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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(key) ?? fallback); + } + + private safeError(error: unknown): string { + return (error instanceof Error ? error.message : String(error)).replace(/amqps?:\/\/[^@\s]+@/gi, 'amqp://[redacted]@').slice(0, 500); + } +} diff --git a/src/modules/outbox/outbox.module.ts b/src/modules/outbox/outbox.module.ts index 514890e..69e804c 100644 --- a/src/modules/outbox/outbox.module.ts +++ b/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 {} diff --git a/src/modules/outbox/outbox.service.spec.ts b/src/modules/outbox/outbox.service.spec.ts index 8f40e82..d5e1743 100644 --- a/src/modules/outbox/outbox.service.spec.ts +++ b/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(); diff --git a/src/modules/outbox/outbox.service.ts b/src/modules/outbox/outbox.service.ts index 917198d..76ab624 100644 --- a/src/modules/outbox/outbox.service.ts +++ b/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 { 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 { - 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 { + 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 { + await this.queueService.enqueue(OutboxService.PROCESS_EVENT_JOB, { eventId }); + } + + async listByStatuses(statuses: OutboxEvent['status'][], limit = 100): Promise { + return this.outboxEventModel.find({ status: { $in: statuses } }).sort({ createdAt: 1 }).limit(limit).exec(); + } + + async retryEvent(eventId: string): Promise { + 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 { - 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(); diff --git a/src/modules/outbox/schemas/outbox-event.schema.ts b/src/modules/outbox/schemas/outbox-event.schema.ts index 03814c6..e205eff 100644 --- a/src/modules/outbox/schemas/outbox-event.schema.ts +++ b/src/modules/outbox/schemas/outbox-event.schema.ts @@ -5,14 +5,26 @@ export type OutboxEventDocument = HydratedDocument; @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; - @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 }); diff --git a/src/modules/users/users.repository.ts b/src/modules/users/users.repository.ts index afa5c7f..24fccd1 100644 --- a/src/modules/users/users.repository.ts +++ b/src/modules/users/users.repository.ts @@ -61,12 +61,12 @@ export class UsersRepository { .exec(); } - async setFollowersCount(userId: string, followersCount: number): Promise { - await this.userModel.findByIdAndUpdate(userId, { followersCount }, { new: false }).exec(); + async setFollowersCount(userId: string, followersCount: number, session?: ClientSession): Promise { + await this.userModel.findByIdAndUpdate(userId, { followersCount }, { new: false, session }).exec(); } - async setFollowingCount(userId: string, followingCount: number): Promise { - await this.userModel.findByIdAndUpdate(userId, { followingCount }, { new: false }).exec(); + async setFollowingCount(userId: string, followingCount: number, session?: ClientSession): Promise { + await this.userModel.findByIdAndUpdate(userId, { followingCount }, { new: false, session }).exec(); } async setPresence(userId: string, isOnline: boolean, lastSeenAt: Date): Promise { diff --git a/tsconfig.json b/tsconfig.json index 2dfda9f..e7a3905 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -10,6 +10,11 @@ "sourceMap": true, "outDir": "./dist", "baseUrl": "./", + "paths": { + "@app/contracts": ["libs/contracts/src"], + "@app/events": ["libs/events/src"], + "@app/common": ["libs/common/src"] + }, "incremental": true, "skipLibCheck": true, "strict": true, diff --git a/tsconfig.legacy.json b/tsconfig.legacy.json new file mode 100644 index 0000000..8cedbfd --- /dev/null +++ b/tsconfig.legacy.json @@ -0,0 +1,11 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "rootDir": ".", + "outDir": "./dist/legacy", + "declaration": false, + "incremental": false + }, + "include": ["src/**/*.ts", "libs/**/*.ts"], + "exclude": ["node_modules", "dist", "test", "oudelaa_dashboard", "**/*.spec.ts"] +}