Skip to content

Commit 02c0232

Browse files
Arun-kushwaha007Arun-kushwaha007
committed
feat: Implement BullMQ and Redis for job queueing, handling notifications, receipt processing, and payment webhooks.
1 parent 0e9df8f commit 02c0232

7 files changed

Lines changed: 192 additions & 80 deletions

File tree

‎apps/backend/package.json‎

Lines changed: 69 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -1,71 +1,71 @@
11
{
2-
"name": "backend",
3-
"version": "1.0.0",
4-
"private": true,
5-
"scripts": {
6-
"build": "tsc -p tsconfig.build.json",
7-
"dev": "nest start --watch --entryFile apps/backend/src/main",
8-
"start": "node dist/apps/backend/src/main.js",
9-
"lint": "tsc --noEmit",
10-
"test": "jest --runInBand --passWithNoTests",
11-
"prisma:generate": "prisma generate",
12-
"prisma:migrate": "prisma migrate dev",
13-
"seed": "ts-node prisma/seed.ts"
14-
},
15-
"dependencies": {
16-
"@aws-sdk/client-s3": "^3.888.0",
17-
"@aws-sdk/s3-request-presigner": "^3.888.0",
18-
"@fairshare/shared-types": "workspace:*",
19-
"@nestjs/common": "^11.0.0",
20-
"@nestjs/config": "^4.0.0",
21-
"@nestjs/core": "^11.0.0",
22-
"@nestjs/jwt": "^11.0.0",
23-
"@nestjs/passport": "^11.0.5",
24-
"@nestjs/platform-express": "^11.0.0",
25-
"@nestjs/platform-socket.io": "^11.1.14",
26-
"@nestjs/websockets": "^11.1.14",
27-
"@opentelemetry/api": "^1.9.0",
28-
"@opentelemetry/auto-instrumentations-node": "^0.65.0",
29-
"@opentelemetry/sdk-node": "^0.206.0",
30-
"@prisma/client": "^6.4.1",
31-
"bcrypt": "^5.1.1",
32-
"class-transformer": "^0.5.1",
33-
"class-validator": "^0.14.1",
34-
"compression": "^1.8.0",
35-
"cookie-parser": "^1.4.7",
36-
"bullmq": "^5.63.1",
37-
"csurf": "^1.11.0",
38-
"expo-server-sdk": "^3.15.0",
39-
"helmet": "^8.0.0",
40-
"ioredis": "^5.4.2",
41-
"passport": "^0.7.0",
42-
"passport-google-oauth20": "^2.0.0",
43-
"passport-jwt": "^4.0.1",
44-
"prisma": "^6.4.1",
45-
"reflect-metadata": "^0.2.2",
46-
"rxjs": "^7.8.1",
47-
"socket.io": "^4.8.1",
48-
"prom-client": "^15.1.3",
49-
"@sentry/node": "^10.6.0",
50-
"@nestjs/throttler": "^6.4.0",
51-
"stripe": "^18.5.0"
52-
},
53-
"devDependencies": {
54-
"@nestjs/cli": "^11.0.0",
55-
"@nestjs/schematics": "^11.0.0",
56-
"@nestjs/testing": "^11.0.0",
57-
"@types/bcrypt": "^5.0.2",
58-
"@types/compression": "^1.7.5",
59-
"@types/cookie-parser": "^1.4.8",
60-
"@types/csurf": "^1.11.5",
61-
"@types/express": "^5.0.0",
62-
"@types/jest": "^29.5.12",
63-
"@types/node": "^22.10.2",
64-
"@types/passport-google-oauth20": "^2.0.16",
65-
"@types/passport-jwt": "^4.0.1",
66-
"jest": "^29.7.0",
67-
"ts-jest": "^29.2.5",
68-
"ts-node": "^10.9.2",
69-
"typescript": "^5.7.3"
70-
}
2+
"name": "backend",
3+
"version": "1.0.0",
4+
"private": true,
5+
"scripts": {
6+
"build": "tsc -p tsconfig.build.json",
7+
"dev": "nest start --watch --entryFile apps/backend/src/main",
8+
"start": "node dist/apps/backend/src/main.js",
9+
"lint": "tsc --noEmit",
10+
"test": "jest --runInBand --passWithNoTests",
11+
"prisma:generate": "prisma generate",
12+
"prisma:migrate": "prisma migrate dev",
13+
"seed": "ts-node prisma/seed.ts"
14+
},
15+
"dependencies": {
16+
"@aws-sdk/client-s3": "^3.888.0",
17+
"@aws-sdk/s3-request-presigner": "^3.888.0",
18+
"@fairshare/shared-types": "workspace:*",
19+
"@nestjs/common": "^11.0.0",
20+
"@nestjs/config": "^4.0.0",
21+
"@nestjs/core": "^11.0.0",
22+
"@nestjs/jwt": "^11.0.0",
23+
"@nestjs/passport": "^11.0.5",
24+
"@nestjs/platform-express": "^11.0.0",
25+
"@nestjs/platform-socket.io": "^11.1.14",
26+
"@nestjs/websockets": "^11.1.14",
27+
"@opentelemetry/api": "^1.9.0",
28+
"@opentelemetry/auto-instrumentations-node": "^0.65.0",
29+
"@opentelemetry/sdk-node": "^0.206.0",
30+
"@prisma/client": "^6.4.1",
31+
"bcrypt": "^5.1.1",
32+
"class-transformer": "^0.5.1",
33+
"class-validator": "^0.14.1",
34+
"compression": "^1.8.0",
35+
"cookie-parser": "^1.4.7",
36+
"bullmq": "^5.63.1",
37+
"csurf": "^1.11.0",
38+
"expo-server-sdk": "^3.15.0",
39+
"helmet": "^8.0.0",
40+
"ioredis": "^5.4.2",
41+
"passport": "^0.7.0",
42+
"passport-google-oauth20": "^2.0.0",
43+
"passport-jwt": "^4.0.1",
44+
"prisma": "^6.4.1",
45+
"reflect-metadata": "^0.2.2",
46+
"rxjs": "^7.8.1",
47+
"socket.io": "^4.8.1",
48+
"prom-client": "^15.1.3",
49+
"@sentry/node": "^10.6.0",
50+
"@nestjs/throttler": "^6.4.0",
51+
"stripe": "^18.5.0"
52+
},
53+
"devDependencies": {
54+
"@nestjs/cli": "^11.0.0",
55+
"@nestjs/schematics": "^11.0.0",
56+
"@nestjs/testing": "^11.0.0",
57+
"@types/bcrypt": "^5.0.2",
58+
"@types/compression": "^1.7.5",
59+
"@types/cookie-parser": "^1.4.8",
60+
"@types/csurf": "^1.11.5",
61+
"@types/express": "^5.0.0",
62+
"@types/jest": "^29.5.12",
63+
"@types/node": "^22.10.2",
64+
"@types/passport-google-oauth20": "^2.0.16",
65+
"@types/passport-jwt": "^4.0.1",
66+
"jest": "^29.7.0",
67+
"ts-jest": "^29.2.5",
68+
"ts-node": "^10.9.2",
69+
"typescript": "^5.7.3"
70+
}
7171
}

‎apps/backend/src/jobs/jobs-queue.service.ts‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,36 @@
1-
import { Injectable } from '@nestjs/common';
1+
import { Injectable, Logger } from '@nestjs/common';
22
import { Queue } from 'bullmq';
33
import { AppConfigService } from '../config/app-config.service';
44
import { NOTIFICATION_QUEUE, PAYMENT_WEBHOOKS_QUEUE, RECEIPT_PROCESSING_QUEUE } from './jobs.constants';
55

6+
const MAX_REDIS_RETRIES = 3;
7+
const RETRY_DELAY_MS = 250;
8+
const MAX_RETRY_DELAY_MS = 1000;
9+
610
@Injectable()
711
export class JobsQueueService {
12+
private readonly logger = new Logger(JobsQueueService.name);
813
private readonly notificationQueue: Queue;
914
private readonly receiptQueue: Queue;
1015
private readonly paymentWebhookQueue: Queue;
1116

1217
constructor(config: AppConfigService) {
13-
const connection = { url: config.redisUrl };
18+
const connection = {
19+
url: config.redisUrl,
20+
lazyConnect: true,
21+
maxRetriesPerRequest: null,
22+
enableOfflineQueue: false,
23+
connectionName: 'fairshare:jobs-queue',
24+
retryStrategy: (times: number) => {
25+
if (times > MAX_REDIS_RETRIES) {
26+
this.logger.warn('jobs-queue Redis reconnect attempts exhausted');
27+
return null;
28+
}
29+
30+
return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
31+
},
32+
};
33+
1434
this.notificationQueue = new Queue(NOTIFICATION_QUEUE, { connection });
1535
this.receiptQueue = new Queue(RECEIPT_PROCESSING_QUEUE, { connection });
1636
this.paymentWebhookQueue = new Queue(PAYMENT_WEBHOOKS_QUEUE, { connection });

‎apps/backend/src/jobs/jobs-worker.service.ts‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,10 @@ import { NotificationsService } from '../notifications/notifications.service';
55
import { PaymentsService } from '../payments/payments.service';
66
import { NOTIFICATION_QUEUE, PAYMENT_WEBHOOKS_QUEUE, RECEIPT_PROCESSING_QUEUE } from './jobs.constants';
77

8+
const MAX_REDIS_RETRIES = 3;
9+
const RETRY_DELAY_MS = 250;
10+
const MAX_RETRY_DELAY_MS = 1000;
11+
812
@Injectable()
913
export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
1014
private readonly logger = new Logger(JobsWorkerService.name);
@@ -18,7 +22,21 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
1822
) {}
1923

2024
async onModuleInit(): Promise<void> {
21-
const connection = { url: this.config.redisUrl };
25+
const connection = {
26+
url: this.config.redisUrl,
27+
lazyConnect: true,
28+
maxRetriesPerRequest: null,
29+
enableOfflineQueue: false,
30+
connectionName: 'fairshare:jobs-worker',
31+
retryStrategy: (times: number) => {
32+
if (times > MAX_REDIS_RETRIES) {
33+
this.logger.warn('jobs-worker Redis reconnect attempts exhausted');
34+
return null;
35+
}
36+
37+
return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
38+
},
39+
};
2240

2341
const notificationWorker = new Worker(
2442
NOTIFICATION_QUEUE,
@@ -48,6 +66,11 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
4866
);
4967

5068
this.workers = [notificationWorker, receiptWorker, paymentWorker];
69+
for (const worker of this.workers) {
70+
worker.on('error', (error) => {
71+
this.logger.warn(`BullMQ worker error: ${error.message}`);
72+
});
73+
}
5174
}
5275

5376
async onModuleDestroy(): Promise<void> {

‎apps/backend/src/notifications/notifications.service.ts‎

Lines changed: 44 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,10 @@ import { AppConfigService } from '../config/app-config.service';
55
import { PrismaService } from '../common/prisma.service';
66
import { JobsQueueService } from '../jobs/jobs-queue.service';
77

8+
const MAX_REDIS_RETRIES = 3;
9+
const RETRY_DELAY_MS = 250;
10+
const MAX_RETRY_DELAY_MS = 1000;
11+
812
export type NotificationType = 'expense_created' | 'expense_deleted' | 'settlement_created' | 'group_invite';
913

1014
type NotificationEventPayload = {
@@ -30,13 +34,49 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
3034
@Inject(forwardRef(() => JobsQueueService))
3135
private readonly jobsQueueService: JobsQueueService,
3236
) {
33-
this.publisher = new Redis(config.redisUrl);
34-
this.subscriber = new Redis(config.redisUrl);
37+
const redisOptions = (connectionName: string) => ({
38+
lazyConnect: true,
39+
maxRetriesPerRequest: 1,
40+
enableOfflineQueue: false,
41+
connectionName,
42+
retryStrategy: (times: number) => {
43+
if (times > MAX_REDIS_RETRIES) {
44+
this.logger.warn(`${connectionName} Redis reconnect attempts exhausted`);
45+
return null;
46+
}
47+
48+
return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
49+
},
50+
});
51+
52+
this.publisher = new Redis(config.redisUrl, redisOptions('fairshare:notifications-publisher'));
53+
this.subscriber = new Redis(config.redisUrl, redisOptions('fairshare:notifications-subscriber'));
54+
55+
for (const [client, name] of [
56+
[this.publisher, 'notifications-publisher'],
57+
[this.subscriber, 'notifications-subscriber'],
58+
] as const) {
59+
client.on('error', (error) => {
60+
this.logger.warn(`${name} Redis error: ${error.message}`);
61+
});
62+
client.on('end', () => {
63+
this.logger.warn(`${name} Redis connection closed`);
64+
});
65+
}
66+
3567
this.expo = new Expo();
3668
}
3769

3870
async onModuleInit(): Promise<void> {
39-
await this.subscriber.subscribe(this.channel);
71+
try {
72+
await this.subscriber.subscribe(this.channel);
73+
} catch (error) {
74+
this.logger.warn(
75+
`Redis pub/sub unavailable during startup: ${error instanceof Error ? error.message : 'unknown error'}`,
76+
);
77+
return;
78+
}
79+
4080
this.subscriber.on('message', (channel, message) => {
4181
if (channel !== this.channel) {
4282
return;
@@ -53,8 +93,7 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
5393
}
5494

5595
async onModuleDestroy(): Promise<void> {
56-
await this.subscriber.quit();
57-
await this.publisher.quit();
96+
await Promise.allSettled([this.subscriber.quit(), this.publisher.quit()]);
5897
}
5998

6099
async sendPushNotification(userIds: string[], payload: NotificationEventPayload): Promise<void> {

‎apps/backend/src/redis/redis.module.ts‎

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,46 @@
1-
import { Global, Module } from '@nestjs/common';
1+
import { Global, Logger, Module } from '@nestjs/common';
22
import Redis from 'ioredis';
33
import { AppConfigModule } from '../config/app-config.module';
44
import { AppConfigService } from '../config/app-config.service';
55
import { RedisService } from './redis.service';
66

7+
const MAX_REDIS_RETRIES = 3;
8+
const RETRY_DELAY_MS = 250;
9+
const MAX_RETRY_DELAY_MS = 1000;
10+
711
@Global()
812
@Module({
913
imports: [AppConfigModule],
1014
providers: [
1115
{
1216
provide: 'REDIS_CLIENT',
1317
inject: [AppConfigService],
14-
useFactory: (config: AppConfigService) => new Redis(config.redisUrl),
18+
useFactory: (config: AppConfigService) => {
19+
const logger = new Logger('RedisClient');
20+
const client = new Redis(config.redisUrl, {
21+
lazyConnect: true,
22+
maxRetriesPerRequest: 1,
23+
enableOfflineQueue: false,
24+
connectionName: 'fairshare:cache',
25+
retryStrategy: (times) => {
26+
if (times > MAX_REDIS_RETRIES) {
27+
logger.warn('cache Redis reconnect attempts exhausted');
28+
return null;
29+
}
30+
31+
return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
32+
},
33+
});
34+
35+
client.on('error', (error) => {
36+
logger.warn(`cache Redis error: ${error.message}`);
37+
});
38+
client.on('end', () => {
39+
logger.warn('cache Redis connection closed');
40+
});
41+
42+
return client;
43+
},
1544
},
1645
RedisService,
1746
],

‎docker-compose.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ services:
1616
image: redis:7-alpine
1717
container_name: fairshare-redis
1818
ports:
19-
- "6379:6379"
19+
- "6380:6379"
2020
volumes:
2121
- redis_data:/data
2222

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
"build": "turbo run build",
77
"dev": "turbo run dev --parallel",
88
"dev:backend": "pnpm --filter backend dev",
9+
"run:backend": "pnpm --filter backend dev",
910
"dev:mobile": "pnpm --filter mobile dev",
1011
"dev:web": "pnpm --filter web dev",
1112
"mobile:build": "pnpm --filter mobile eas:build",

0 commit comments

Comments
 (0)