Skip to content

Commit 53a0d37

Browse files
Arun-kushwaha007Arun-kushwaha007
authored
Improve local dev startup resilience across backend, web, and mobile (#13)
* fix(backend): tolerate missing Redis during local startup * fix(web): handle backend outages and add legal pages * chore(mobile): align Expo packages and strict typings
1 parent e8973fd commit 53a0d37

15 files changed

Lines changed: 821 additions & 785 deletions

File tree

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

Lines changed: 125 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
1-
import { Injectable, Logger } from '@nestjs/common';
1+
import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
22
import { Queue } from 'bullmq';
3+
import Redis from 'ioredis';
34
import { AppConfigService } from '../config/app-config.service';
45
import { NOTIFICATION_QUEUE, PAYMENT_WEBHOOKS_QUEUE, RECEIPT_PROCESSING_QUEUE } from './jobs.constants';
56

@@ -8,15 +9,115 @@ const RETRY_DELAY_MS = 250;
89
const MAX_RETRY_DELAY_MS = 1000;
910

1011
@Injectable()
11-
export class JobsQueueService {
12+
export class JobsQueueService implements OnModuleDestroy {
1213
private readonly logger = new Logger(JobsQueueService.name);
13-
private readonly notificationQueue: Queue;
14-
private readonly receiptQueue: Queue;
15-
private readonly paymentWebhookQueue: Queue;
14+
private readonly config: AppConfigService;
15+
private notificationQueue: Queue | null = null;
16+
private receiptQueue: Queue | null = null;
17+
private paymentWebhookQueue: Queue | null = null;
18+
private redisAvailable: boolean | null = null;
1619

1720
constructor(config: AppConfigService) {
21+
this.config = config;
22+
}
23+
24+
async enqueueNotification(payload: {
25+
userIds: string[];
26+
payload: {
27+
type: 'expense_created' | 'expense_deleted' | 'settlement_created' | 'group_invite';
28+
title: string;
29+
body: string;
30+
data?: Record<string, unknown>;
31+
};
32+
}): Promise<void> {
33+
const queue = await this.getNotificationQueue();
34+
if (!queue) {
35+
return;
36+
}
37+
38+
try {
39+
await queue.add('send', payload, {
40+
attempts: 3,
41+
backoff: { type: 'exponential', delay: 500 },
42+
removeOnComplete: true,
43+
removeOnFail: 100,
44+
});
45+
} catch (error) {
46+
this.logger.warn(`Notification queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
47+
}
48+
}
49+
50+
async enqueueReceiptProcessing(payload: { receiptId: string; expenseId: string }): Promise<void> {
51+
const queue = await this.getReceiptQueue();
52+
if (!queue) {
53+
return;
54+
}
55+
56+
try {
57+
await queue.add('process', payload, {
58+
attempts: 3,
59+
removeOnComplete: true,
60+
removeOnFail: 100,
61+
});
62+
} catch (error) {
63+
this.logger.warn(`Receipt queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
64+
}
65+
}
66+
67+
async enqueuePaymentWebhook(payload: { signature?: string; body: Record<string, unknown> }): Promise<void> {
68+
const queue = await this.getPaymentWebhookQueue();
69+
if (!queue) {
70+
return;
71+
}
72+
73+
try {
74+
await queue.add('handle', payload, {
75+
attempts: 5,
76+
backoff: { type: 'exponential', delay: 1000 },
77+
removeOnComplete: true,
78+
removeOnFail: 100,
79+
});
80+
} catch (error) {
81+
this.logger.warn(`Payment webhook queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
82+
}
83+
}
84+
85+
async onModuleDestroy(): Promise<void> {
86+
await Promise.allSettled(
87+
[this.notificationQueue, this.receiptQueue, this.paymentWebhookQueue]
88+
.filter((queue): queue is Queue => Boolean(queue))
89+
.map((queue) => queue.close()),
90+
);
91+
}
92+
93+
private async getNotificationQueue(): Promise<Queue | null> {
94+
const ready = await this.ensureQueues();
95+
return ready ? this.notificationQueue : null;
96+
}
97+
98+
private async getReceiptQueue(): Promise<Queue | null> {
99+
const ready = await this.ensureQueues();
100+
return ready ? this.receiptQueue : null;
101+
}
102+
103+
private async getPaymentWebhookQueue(): Promise<Queue | null> {
104+
const ready = await this.ensureQueues();
105+
return ready ? this.paymentWebhookQueue : null;
106+
}
107+
108+
private async ensureQueues(): Promise<boolean> {
109+
if (this.redisAvailable !== null) {
110+
return this.redisAvailable;
111+
}
112+
113+
this.redisAvailable = await this.canReachRedis();
114+
if (!this.redisAvailable) {
115+
this.logger.warn('Redis unavailable, BullMQ queue producers disabled for local startup');
116+
return false;
117+
}
118+
18119
const connection = {
19-
url: config.redisUrl,
120+
url: this.config.redisUrl,
20121
lazyConnect: true,
21122
maxRetriesPerRequest: null,
22123
enableOfflineQueue: false,
@@ -34,39 +135,27 @@ export class JobsQueueService {
34135
this.notificationQueue = new Queue(NOTIFICATION_QUEUE, { connection });
35136
this.receiptQueue = new Queue(RECEIPT_PROCESSING_QUEUE, { connection });
36137
this.paymentWebhookQueue = new Queue(PAYMENT_WEBHOOKS_QUEUE, { connection });
138+
return true;
37139
}
38140

39-
async enqueueNotification(payload: {
40-
userIds: string[];
41-
payload: {
42-
type: 'expense_created' | 'expense_deleted' | 'settlement_created' | 'group_invite';
43-
title: string;
44-
body: string;
45-
data?: Record<string, unknown>;
46-
};
47-
}): Promise<void> {
48-
await this.notificationQueue.add('send', payload, {
49-
attempts: 3,
50-
backoff: { type: 'exponential', delay: 500 },
51-
removeOnComplete: true,
52-
removeOnFail: 100,
53-
});
54-
}
55-
56-
async enqueueReceiptProcessing(payload: { receiptId: string; expenseId: string }): Promise<void> {
57-
await this.receiptQueue.add('process', payload, {
58-
attempts: 3,
59-
removeOnComplete: true,
60-
removeOnFail: 100,
141+
private async canReachRedis(): Promise<boolean> {
142+
const probe = new Redis(this.config.redisUrl, {
143+
lazyConnect: true,
144+
maxRetriesPerRequest: 1,
145+
enableOfflineQueue: false,
146+
connectTimeout: 750,
147+
retryStrategy: () => null,
61148
});
62-
}
63149

64-
async enqueuePaymentWebhook(payload: { signature?: string; body: Record<string, unknown> }): Promise<void> {
65-
await this.paymentWebhookQueue.add('handle', payload, {
66-
attempts: 5,
67-
backoff: { type: 'exponential', delay: 1000 },
68-
removeOnComplete: true,
69-
removeOnFail: 100,
70-
});
150+
try {
151+
await probe.connect();
152+
await probe.ping();
153+
await probe.quit();
154+
return true;
155+
} catch (error) {
156+
this.logger.warn(`Redis queue probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
157+
probe.disconnect();
158+
return false;
159+
}
71160
}
72161
}

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

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { forwardRef, Inject, Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
22
import { Worker } from 'bullmq';
3+
import Redis from 'ioredis';
34
import { AppConfigService } from '../config/app-config.service';
45
import { NotificationsService } from '../notifications/notifications.service';
56
import { PaymentsService } from '../payments/payments.service';
@@ -22,6 +23,12 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
2223
) {}
2324

2425
async onModuleInit(): Promise<void> {
26+
const redisAvailable = await this.canReachRedis();
27+
if (!redisAvailable) {
28+
this.logger.warn('Redis unavailable, BullMQ workers disabled for local startup');
29+
return;
30+
}
31+
2532
const connection = {
2633
url: this.config.redisUrl,
2734
lazyConnect: true,
@@ -76,4 +83,25 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
7683
async onModuleDestroy(): Promise<void> {
7784
await Promise.all(this.workers.map((worker) => worker.close()));
7885
}
86+
87+
private async canReachRedis(): Promise<boolean> {
88+
const probe = new Redis(this.config.redisUrl, {
89+
lazyConnect: true,
90+
maxRetriesPerRequest: 1,
91+
enableOfflineQueue: false,
92+
connectTimeout: 750,
93+
retryStrategy: () => null,
94+
});
95+
96+
try {
97+
await probe.connect();
98+
await probe.ping();
99+
await probe.quit();
100+
return true;
101+
} catch (error) {
102+
this.logger.warn(`Redis probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
103+
probe.disconnect();
104+
return false;
105+
}
106+
}
79107
}

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

Lines changed: 78 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -24,56 +24,44 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
2424
private static readonly baseRetryDelayMs = 500;
2525
private readonly logger = new Logger(NotificationsService.name);
2626
private readonly channel = 'fairshare:notifications';
27-
private readonly publisher: Redis;
28-
private readonly subscriber: Redis;
27+
private readonly config: AppConfigService;
28+
private publisher: Redis | null = null;
29+
private subscriber: Redis | null = null;
2930
private readonly expo: Expo;
31+
private pubSubEnabled = true;
3032

3133
constructor(
3234
config: AppConfigService,
3335
private readonly prisma: PrismaService,
3436
@Inject(forwardRef(() => JobsQueueService))
3537
private readonly jobsQueueService: JobsQueueService,
3638
) {
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-
39+
this.config = config;
6740
this.expo = new Expo();
6841
}
6942

7043
async onModuleInit(): Promise<void> {
44+
const redisAvailable = await this.canReachRedis();
45+
if (!redisAvailable) {
46+
this.pubSubEnabled = false;
47+
this.logger.warn('Redis unavailable, notifications pub/sub disabled for local startup');
48+
return;
49+
}
50+
51+
this.publisher = this.createRedisClient('fairshare:notifications-publisher');
52+
this.subscriber = this.createRedisClient('fairshare:notifications-subscriber');
53+
7154
try {
7255
await this.subscriber.subscribe(this.channel);
7356
} catch (error) {
57+
this.pubSubEnabled = false;
7458
this.logger.warn(
7559
`Redis pub/sub unavailable during startup: ${error instanceof Error ? error.message : 'unknown error'}`,
7660
);
61+
this.publisher?.disconnect();
62+
this.subscriber?.disconnect();
63+
this.publisher = null;
64+
this.subscriber = null;
7765
return;
7866
}
7967

@@ -93,7 +81,9 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
9381
}
9482

9583
async onModuleDestroy(): Promise<void> {
96-
await Promise.allSettled([this.subscriber.quit(), this.publisher.quit()]);
84+
await Promise.allSettled(
85+
[this.subscriber, this.publisher].filter((client): client is Redis => Boolean(client)).map((client) => client.quit()),
86+
);
9787
}
9888

9989
async sendPushNotification(userIds: string[], payload: NotificationEventPayload): Promise<void> {
@@ -107,7 +97,13 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
10797
queuedAt: new Date().toISOString(),
10898
};
10999

110-
await this.publisher.publish(this.channel, JSON.stringify(event));
100+
if (this.pubSubEnabled && this.publisher) {
101+
try {
102+
await this.publisher.publish(this.channel, JSON.stringify(event));
103+
} catch (error) {
104+
this.logger.warn(`Redis publish skipped: ${error instanceof Error ? error.message : 'unknown error'}`);
105+
}
106+
}
111107

112108
this.logger.log(
113109
JSON.stringify({
@@ -169,7 +165,7 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
169165
this.logger.error('Expo push send failed', error instanceof Error ? error.stack : undefined);
170166
if (retries > 0) {
171167
const attempt = NotificationsService.maxChunkRetries - retries + 1;
172-
const delayMs = NotificationsService.baseRetryDelayMs * (2 ** (attempt - 1));
168+
const delayMs = NotificationsService.baseRetryDelayMs * 2 ** (attempt - 1);
173169
await new Promise((resolve) => setTimeout(resolve, delayMs));
174170
await this.sendChunkWithRetry(chunk, retries - 1);
175171
}
@@ -212,4 +208,51 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
212208
}),
213209
);
214210
}
211+
212+
private createRedisClient(connectionName: string): Redis {
213+
const client = new Redis(this.config.redisUrl, {
214+
lazyConnect: true,
215+
maxRetriesPerRequest: 1,
216+
enableOfflineQueue: false,
217+
connectionName,
218+
retryStrategy: (times: number) => {
219+
if (times > MAX_REDIS_RETRIES) {
220+
this.logger.warn(`${connectionName} Redis reconnect attempts exhausted`);
221+
return null;
222+
}
223+
224+
return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
225+
},
226+
});
227+
228+
client.on('error', (error) => {
229+
this.logger.warn(`${connectionName} Redis error: ${error.message}`);
230+
});
231+
client.on('end', () => {
232+
this.logger.warn(`${connectionName} Redis connection closed`);
233+
});
234+
235+
return client;
236+
}
237+
238+
private async canReachRedis(): Promise<boolean> {
239+
const probe = new Redis(this.config.redisUrl, {
240+
lazyConnect: true,
241+
maxRetriesPerRequest: 1,
242+
enableOfflineQueue: false,
243+
connectTimeout: 750,
244+
retryStrategy: () => null,
245+
});
246+
247+
try {
248+
await probe.connect();
249+
await probe.ping();
250+
await probe.quit();
251+
return true;
252+
} catch (error) {
253+
this.logger.warn(`Redis notifications probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
254+
probe.disconnect();
255+
return false;
256+
}
257+
}
215258
}

0 commit comments

Comments
 (0)