import { Queue, Worker } from 'bullmq'; import { Expo } from 'expo-server-sdk'; import { redisUrl } from './env-exporter'; import { db } from '../db/db_index'; import { notifCreds } from '../db/schema'; import { eq } from 'drizzle-orm'; import { generateSignedUrlFromS3Url } from './s3-client'; import { NOTIFS_QUEUE, ORDER_PLACED_MESSAGE, PAYMENT_FAILED_MESSAGE, ORDER_PACKAGED_MESSAGE, ORDER_OUT_FOR_DELIVERY_MESSAGE, ORDER_DELIVERED_MESSAGE, ORDER_CANCELLED_MESSAGE, REFUND_INITIATED_MESSAGE } from './const-strings'; export const notificationQueue = new Queue(NOTIFS_QUEUE, { connection: { url: redisUrl }, defaultJobOptions: { removeOnComplete: true, removeOnFail: 10, attempts: 3, }, }); export const notificationWorker = new Worker(NOTIFS_QUEUE, async (job) => { if (!job) return; const { name, data } = job; console.log(`Processing notification job ${job.id} - ${name}`); if (name === 'send-admin-notification') { await sendAdminNotification(data); } else if (name === 'send-notification') { // Handle legacy notification type console.log('Legacy notification job - not implemented yet'); } }, { connection: { url: redisUrl }, concurrency: 5, }); async function sendAdminNotification(data: { userId: number; title: string; body: string; imageUrl: string | null; notificationId: number; }) { const { userId, title, body, imageUrl } = data; // Get user's push token const [cred] = await db .select({ token: notifCreds.token }) .from(notifCreds) .where(eq(notifCreds.userId, userId)) .limit(1); if (!cred || !cred.token) { console.log(`No push token found for user ${userId}`); return; } const token = cred.token; // Validate Expo push token if (!Expo.isExpoPushToken(token)) { console.error(`Invalid Expo push token for user ${userId}: ${token}`); throw new Error('Invalid push token'); } // Generate signed URL for image if provided const signedImageUrl = imageUrl ? await generateSignedUrlFromS3Url(imageUrl) : null; // Send notification const expo = new Expo(); const message = { to: token, sound: 'default', title, body, data: { imageUrl }, ...(signedImageUrl ? { attachments: [ { url: signedImageUrl, contentType: 'image/jpeg', } ] } : {}), }; try { const [ticket] = await expo.sendPushNotificationsAsync([message]); console.log(`Notification sent to user ${userId}:`, ticket); } catch (error) { console.error(`Failed to send notification to user ${userId}:`, error); throw error; } } notificationWorker.on('completed', (job) => { if (job) console.log(`Notification job ${job.id} completed`); }); notificationWorker.on('failed', (job, err) => { if (job) console.error(`Notification job ${job.id} failed:`, err); }); export async function scheduleNotification(userId: number, payload: any, options?: { delay?: number; priority?: number }) { const jobData = { userId, ...payload }; await notificationQueue.add('send-notification', jobData, options); } // Utility methods for specific notification events export async function sendOrderPlacedNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Order Placed', body: ORDER_PLACED_MESSAGE, type: 'order', orderId }); } export async function sendPaymentFailedNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Payment Failed', body: PAYMENT_FAILED_MESSAGE, type: 'payment', orderId }); } export async function sendOrderPackagedNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Order Packaged', body: ORDER_PACKAGED_MESSAGE, type: 'order', orderId }); } export async function sendOrderOutForDeliveryNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Out for Delivery', body: ORDER_OUT_FOR_DELIVERY_MESSAGE, type: 'order', orderId }); } export async function sendOrderDeliveredNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Order Delivered', body: ORDER_DELIVERED_MESSAGE, type: 'order', orderId }); } export async function sendOrderCancelledNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Order Cancelled', body: ORDER_CANCELLED_MESSAGE, type: 'order', orderId }); } export async function sendRefundInitiatedNotification(userId: number, orderId?: string) { await scheduleNotification(userId, { title: 'Refund Initiated', body: REFUND_INITIATED_MESSAGE, type: 'refund', orderId }); } process.on('SIGTERM', async () => { await notificationQueue.close(); await notificationWorker.close(); });