Improve local dev startup resilience across backend, web, and mobile

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

@@ -1,5 +1,6 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
 import { Queue } from 'bullmq';
+import Redis from 'ioredis';
 import { AppConfigService } from '../config/app-config.service';
 import { NOTIFICATION_QUEUE, PAYMENT_WEBHOOKS_QUEUE, RECEIPT_PROCESSING_QUEUE } from './jobs.constants';
 
@@ -8,15 +9,115 @@ const RETRY_DELAY_MS = 250;
 const MAX_RETRY_DELAY_MS = 1000;
 
 @Injectable()
-export class JobsQueueService {
+export class JobsQueueService implements OnModuleDestroy {
   private readonly logger = new Logger(JobsQueueService.name);
-  private readonly notificationQueue: Queue;
-  private readonly receiptQueue: Queue;
-  private readonly paymentWebhookQueue: Queue;
+  private readonly config: AppConfigService;
+  private notificationQueue: Queue | null = null;
+  private receiptQueue: Queue | null = null;
+  private paymentWebhookQueue: Queue | null = null;
+  private redisAvailable: boolean | null = null;
 
   constructor(config: AppConfigService) {
+    this.config = config;
+  }
+
+  async enqueueNotification(payload: {
+    userIds: string[];
+    payload: {
+      type: 'expense_created' | 'expense_deleted' | 'settlement_created' | 'group_invite';
+      title: string;
+      body: string;
+      data?: Record<string, unknown>;
+    };
+  }): Promise<void> {
+    const queue = await this.getNotificationQueue();
+    if (!queue) {
+      return;
+    }
+
+    try {
+      await queue.add('send', payload, {
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 500 },
+        removeOnComplete: true,
+        removeOnFail: 100,
+      });
+    } catch (error) {
+      this.logger.warn(`Notification queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
+    }
+  }
+
+  async enqueueReceiptProcessing(payload: { receiptId: string; expenseId: string }): Promise<void> {
+    const queue = await this.getReceiptQueue();
+    if (!queue) {
+      return;
+    }
+
+    try {
+      await queue.add('process', payload, {
+        attempts: 3,
+        removeOnComplete: true,
+        removeOnFail: 100,
+      });
+    } catch (error) {
+      this.logger.warn(`Receipt queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
+    }
+  }
+
+  async enqueuePaymentWebhook(payload: { signature?: string; body: Record<string, unknown> }): Promise<void> {
+    const queue = await this.getPaymentWebhookQueue();
+    if (!queue) {
+      return;
+    }
+
+    try {
+      await queue.add('handle', payload, {
+        attempts: 5,
+        backoff: { type: 'exponential', delay: 1000 },
+        removeOnComplete: true,
+        removeOnFail: 100,
+      });
+    } catch (error) {
+      this.logger.warn(`Payment webhook queue unavailable: ${error instanceof Error ? error.message : 'unknown error'}`);
+    }
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await Promise.allSettled(
+      [this.notificationQueue, this.receiptQueue, this.paymentWebhookQueue]
+        .filter((queue): queue is Queue => Boolean(queue))
+        .map((queue) => queue.close()),
+    );
+  }
+
+  private async getNotificationQueue(): Promise<Queue | null> {
+    const ready = await this.ensureQueues();
+    return ready ? this.notificationQueue : null;
+  }
+
+  private async getReceiptQueue(): Promise<Queue | null> {
+    const ready = await this.ensureQueues();
+    return ready ? this.receiptQueue : null;
+  }
+
+  private async getPaymentWebhookQueue(): Promise<Queue | null> {
+    const ready = await this.ensureQueues();
+    return ready ? this.paymentWebhookQueue : null;
+  }
+
+  private async ensureQueues(): Promise<boolean> {
+    if (this.redisAvailable !== null) {
+      return this.redisAvailable;
+    }
+
+    this.redisAvailable = await this.canReachRedis();
+    if (!this.redisAvailable) {
+      this.logger.warn('Redis unavailable, BullMQ queue producers disabled for local startup');
+      return false;
+    }
+
     const connection = {
-      url: config.redisUrl,
+      url: this.config.redisUrl,
       lazyConnect: true,
       maxRetriesPerRequest: null,
       enableOfflineQueue: false,
@@ -34,39 +135,27 @@ export class JobsQueueService {
     this.notificationQueue = new Queue(NOTIFICATION_QUEUE, { connection });
     this.receiptQueue = new Queue(RECEIPT_PROCESSING_QUEUE, { connection });
     this.paymentWebhookQueue = new Queue(PAYMENT_WEBHOOKS_QUEUE, { connection });
+    return true;
   }
 
-  async enqueueNotification(payload: {
-    userIds: string[];
-    payload: {
-      type: 'expense_created' | 'expense_deleted' | 'settlement_created' | 'group_invite';
-      title: string;
-      body: string;
-      data?: Record<string, unknown>;
-    };
-  }): Promise<void> {
-    await this.notificationQueue.add('send', payload, {
-      attempts: 3,
-      backoff: { type: 'exponential', delay: 500 },
-      removeOnComplete: true,
-      removeOnFail: 100,
-    });
-  }
-
-  async enqueueReceiptProcessing(payload: { receiptId: string; expenseId: string }): Promise<void> {
-    await this.receiptQueue.add('process', payload, {
-      attempts: 3,
-      removeOnComplete: true,
-      removeOnFail: 100,
+  private async canReachRedis(): Promise<boolean> {
+    const probe = new Redis(this.config.redisUrl, {
+      lazyConnect: true,
+      maxRetriesPerRequest: 1,
+      enableOfflineQueue: false,
+      connectTimeout: 750,
+      retryStrategy: () => null,
     });
-  }
 
-  async enqueuePaymentWebhook(payload: { signature?: string; body: Record<string, unknown> }): Promise<void> {
-    await this.paymentWebhookQueue.add('handle', payload, {
-      attempts: 5,
-      backoff: { type: 'exponential', delay: 1000 },
-      removeOnComplete: true,
-      removeOnFail: 100,
-    });
+    try {
+      await probe.connect();
+      await probe.ping();
+      await probe.quit();
+      return true;
+    } catch (error) {
+      this.logger.warn(`Redis queue probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
+      probe.disconnect();
+      return false;
+    }
   }
 }

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

@@ -1,5 +1,6 @@
 import { forwardRef, Inject, Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { Worker } from 'bullmq';
+import Redis from 'ioredis';
 import { AppConfigService } from '../config/app-config.service';
 import { NotificationsService } from '../notifications/notifications.service';
 import { PaymentsService } from '../payments/payments.service';
@@ -22,6 +23,12 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
   ) {}
 
   async onModuleInit(): Promise<void> {
+    const redisAvailable = await this.canReachRedis();
+    if (!redisAvailable) {
+      this.logger.warn('Redis unavailable, BullMQ workers disabled for local startup');
+      return;
+    }
+
     const connection = {
       url: this.config.redisUrl,
       lazyConnect: true,
@@ -76,4 +83,25 @@ export class JobsWorkerService implements OnModuleInit, OnModuleDestroy {
   async onModuleDestroy(): Promise<void> {
     await Promise.all(this.workers.map((worker) => worker.close()));
   }
+
+  private async canReachRedis(): Promise<boolean> {
+    const probe = new Redis(this.config.redisUrl, {
+      lazyConnect: true,
+      maxRetriesPerRequest: 1,
+      enableOfflineQueue: false,
+      connectTimeout: 750,
+      retryStrategy: () => null,
+    });
+
+    try {
+      await probe.connect();
+      await probe.ping();
+      await probe.quit();
+      return true;
+    } catch (error) {
+      this.logger.warn(`Redis probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
+      probe.disconnect();
+      return false;
+    }
+  }
 }

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

@@ -24,56 +24,44 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
   private static readonly baseRetryDelayMs = 500;
   private readonly logger = new Logger(NotificationsService.name);
   private readonly channel = 'fairshare:notifications';
-  private readonly publisher: Redis;
-  private readonly subscriber: Redis;
+  private readonly config: AppConfigService;
+  private publisher: Redis | null = null;
+  private subscriber: Redis | null = null;
   private readonly expo: Expo;
+  private pubSubEnabled = true;
 
   constructor(
     config: AppConfigService,
     private readonly prisma: PrismaService,
     @Inject(forwardRef(() => JobsQueueService))
     private readonly jobsQueueService: JobsQueueService,
   ) {
-    const redisOptions = (connectionName: string) => ({
-      lazyConnect: true,
-      maxRetriesPerRequest: 1,
-      enableOfflineQueue: false,
-      connectionName,
-      retryStrategy: (times: number) => {
-        if (times > MAX_REDIS_RETRIES) {
-          this.logger.warn(`${connectionName} Redis reconnect attempts exhausted`);
-          return null;
-        }
-
-        return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
-      },
-    });
-
-    this.publisher = new Redis(config.redisUrl, redisOptions('fairshare:notifications-publisher'));
-    this.subscriber = new Redis(config.redisUrl, redisOptions('fairshare:notifications-subscriber'));
-
-    for (const [client, name] of [
-      [this.publisher, 'notifications-publisher'],
-      [this.subscriber, 'notifications-subscriber'],
-    ] as const) {
-      client.on('error', (error) => {
-        this.logger.warn(`${name} Redis error: ${error.message}`);
-      });
-      client.on('end', () => {
-        this.logger.warn(`${name} Redis connection closed`);
-      });
-    }
-
+    this.config = config;
     this.expo = new Expo();
   }
 
   async onModuleInit(): Promise<void> {
+    const redisAvailable = await this.canReachRedis();
+    if (!redisAvailable) {
+      this.pubSubEnabled = false;
+      this.logger.warn('Redis unavailable, notifications pub/sub disabled for local startup');
+      return;
+    }
+
+    this.publisher = this.createRedisClient('fairshare:notifications-publisher');
+    this.subscriber = this.createRedisClient('fairshare:notifications-subscriber');
+
     try {
       await this.subscriber.subscribe(this.channel);
     } catch (error) {
+      this.pubSubEnabled = false;
       this.logger.warn(
         `Redis pub/sub unavailable during startup: ${error instanceof Error ? error.message : 'unknown error'}`,
       );
+      this.publisher?.disconnect();
+      this.subscriber?.disconnect();
+      this.publisher = null;
+      this.subscriber = null;
       return;
     }
 
@@ -93,7 +81,9 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
   }
 
   async onModuleDestroy(): Promise<void> {
-    await Promise.allSettled([this.subscriber.quit(), this.publisher.quit()]);
+    await Promise.allSettled(
+      [this.subscriber, this.publisher].filter((client): client is Redis => Boolean(client)).map((client) => client.quit()),
+    );
   }
 
   async sendPushNotification(userIds: string[], payload: NotificationEventPayload): Promise<void> {
@@ -107,7 +97,13 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
       queuedAt: new Date().toISOString(),
     };
 
-    await this.publisher.publish(this.channel, JSON.stringify(event));
+    if (this.pubSubEnabled && this.publisher) {
+      try {
+        await this.publisher.publish(this.channel, JSON.stringify(event));
+      } catch (error) {
+        this.logger.warn(`Redis publish skipped: ${error instanceof Error ? error.message : 'unknown error'}`);
+      }
+    }
 
     this.logger.log(
       JSON.stringify({
@@ -169,7 +165,7 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
       this.logger.error('Expo push send failed', error instanceof Error ? error.stack : undefined);
       if (retries > 0) {
         const attempt = NotificationsService.maxChunkRetries - retries + 1;
-        const delayMs = NotificationsService.baseRetryDelayMs * (2 ** (attempt - 1));
+        const delayMs = NotificationsService.baseRetryDelayMs * 2 ** (attempt - 1);
         await new Promise((resolve) => setTimeout(resolve, delayMs));
         await this.sendChunkWithRetry(chunk, retries - 1);
       }
@@ -212,4 +208,51 @@ export class NotificationsService implements OnModuleInit, OnModuleDestroy {
       }),
     );
   }
+
+  private createRedisClient(connectionName: string): Redis {
+    const client = new Redis(this.config.redisUrl, {
+      lazyConnect: true,
+      maxRetriesPerRequest: 1,
+      enableOfflineQueue: false,
+      connectionName,
+      retryStrategy: (times: number) => {
+        if (times > MAX_REDIS_RETRIES) {
+          this.logger.warn(`${connectionName} Redis reconnect attempts exhausted`);
+          return null;
+        }
+
+        return Math.min(times * RETRY_DELAY_MS, MAX_RETRY_DELAY_MS);
+      },
+    });
+
+    client.on('error', (error) => {
+      this.logger.warn(`${connectionName} Redis error: ${error.message}`);
+    });
+    client.on('end', () => {
+      this.logger.warn(`${connectionName} Redis connection closed`);
+    });
+
+    return client;
+  }
+
+  private async canReachRedis(): Promise<boolean> {
+    const probe = new Redis(this.config.redisUrl, {
+      lazyConnect: true,
+      maxRetriesPerRequest: 1,
+      enableOfflineQueue: false,
+      connectTimeout: 750,
+      retryStrategy: () => null,
+    });
+
+    try {
+      await probe.connect();
+      await probe.ping();
+      await probe.quit();
+      return true;
+    } catch (error) {
+      this.logger.warn(`Redis notifications probe failed: ${error instanceof Error ? error.message : 'unknown error'}`);
+      probe.disconnect();
+      return false;
+    }
+  }
 }