move push notifications to queues

packages/backend/utils/pushNotifications.ts

···
-import { ExpoPushErrorTicket } from "expo-server-sdk"
-import { Expo } from "expo-server-sdk"
+import { Expo, type ExpoPushErrorTicket } from "expo-server-sdk"
 import { logger } from "./logger.js"
 import { Notification, PushNotificationToken } from "../db.js"
+import { Queue } from "bullmq"
+import { environment } from "../environment.js"
+
+const sendPushNotificationQueue = new Queue('sendPushNotification', {
+  connection: environment.bullmqConnection,
+  defaultJobOptions: {
+    removeOnComplete: true,
+    attempts: 3,
+    backoff: {
+      type: 'exponential',
+      delay: 1000
+    }
+  }
+})
 
 export type NotificationBody = {
   notifiedUserId: string
···
 export async function createNotification(notification: NotificationBody, context?: NotificationContext) {
   await Promise.all([
     Notification.create(notification),
-    sendNotification(notification, context)
+    sendPushNotificationQueue.add('sendPushNotification', { notification, context })
   ])
 }
             
 // Error codes reference: https://docs.expo.io/push-notifications/sending-notifications/#individual-errors
-async function handleDeliveryError(response: ExpoPushErrorTicket) {
+export async function handleDeliveryError(response: ExpoPushErrorTicket) {
   logger.error(response)
   const error = response.details?.error
   
···
     }
   }
 }
-
-const verbMap = {
-  LIKE: 'liked',
-  REWOOT: 'rewooted',
-  MENTION: 'replied to',
-  QUOTE: 'quoted',
-  EMOJIREACT: 'reacted to',
-}
-
-function getNotificationTitle(notification: NotificationBody, context?: NotificationContext) {
-  if (notification.notificationType === 'FOLLOW') {
-    return 'New user followed you'
-  }
-
-  if (notification.notificationType === 'EMOJIREACT' && context?.emoji) {
-    return `${context?.userUrl || 'someone'} reacted with ${context.emoji} to your post`
-  }
-
-  return `${context?.userUrl || 'someone'} ${verbMap[notification.notificationType]} your post`
-}
-
-function getNotificationBody(notification: NotificationBody, context?: NotificationContext) {
-  if (notification.notificationType === 'FOLLOW') {
-    return context?.userUrl ? `@${context?.userUrl.replace(/^@/, '')}` : ''
-  }
-
-  return `${context?.postContent}`
-}
-
-async function sendNotification(notification: NotificationBody, context?: NotificationContext) {
-  const userId = notification.notifiedUserId
-  const tokenRows = await PushNotificationToken.findAll({
-    where: {
-      userId
-    }
-  })
-
-  if (tokenRows.length === 0) {
-    return
-  }
-
-  const payloads = tokenRows.map((row) => ({
-    to: row.token,
-    sound: 'default',
-    title: getNotificationTitle(notification, context),
-    body: getNotificationBody(notification, context),
-    data: notification
-  }))
-
-  // this will chunk the payloads into chunks of 1000 (max) and compress notifications with similar content
-  const chunks = expoClient.chunkPushNotifications(payloads)
-  const okTickets = []
-
-  // TODO: handle in a queue with retry logic and exponential backoff
-  for (const chunk of chunks) {
-    try {
-      const responses = await expoClient.sendPushNotificationsAsync(chunk)
-      for (const response of responses) {
-        if (response.status === 'ok') {
-          okTickets.push(response.id)
-        } else {
-          await handleDeliveryError(response)
-        }
-      }
-    } catch (error) {
-      logger.error(error)
-      // TODO: retry sending the notification after some time
-    }
-  }
-
-  scheduleNotificationCheck(okTickets)
-}
-
-function scheduleNotificationCheck(ticketIds: string[]) {
-  // TODO: enqueue a task in the queue to check that the okTickets are actually ok and were delivered to the device
-}
-
-export async function checkNotificationDelivery(ticketIds: string[]) {
-  let receiptIdChunks = expoClient.chunkPushNotificationReceiptIds(ticketIds);
-  for (const chunk of receiptIdChunks) {
-    try {
-      const receipts = await expoClient.getPushNotificationReceiptsAsync(chunk)
-
-      // The receipts specify whether Apple or Google successfully received the
-      // notification and information about an error, if one occurred.
-      for (const receiptId in receipts) {
-        const receipt = receipts[receiptId]
-        if (receipt.status === 'error') {
-          await handleDeliveryError(receipt)
-        }
-      }
-    } catch (error) {
-      // TODO: retry checking the delivery of the notification after some time
-      logger.error(error)
-    }
-  }
-}

packages/backend/utils/queueProcessors/checkPushNotificationDelivery.ts

···
+import { Job } from "bullmq";
+import { logger } from "../logger.js";
+import { handleDeliveryError } from "../pushNotifications.js";
+import { Expo } from "expo-server-sdk";
+
+const expoClient = new Expo()
+
+export async function checkPushNotificationDelivery(job: Job<{ ticketIds: string[] }>) {
+  const { ticketIds } = job.data
+  let receiptIdChunks = expoClient.chunkPushNotificationReceiptIds(ticketIds);
+  for (const chunk of receiptIdChunks) {
+    try {
+      const receipts = await expoClient.getPushNotificationReceiptsAsync(chunk)
+
+      // The receipts specify whether Apple or Google successfully received the
+      // notification and information about an error, if one occurred.
+      for (const receiptId in receipts) {
+        const receipt = receipts[receiptId]
+        if (receipt.status === 'error') {
+          await handleDeliveryError(receipt)
+        }
+      }
+    } catch (error) {
+      // TODO: retry checking the delivery of the notification after some time
+      logger.error(error)
+    }
+  }
+}

packages/backend/utils/queueProcessors/sendPushNotification.ts

···
+import { Expo } from "expo-server-sdk"
+import { PushNotificationToken } from "../../db.js"
+import { logger } from "../logger.js"
+import { handleDeliveryError, type NotificationBody, type NotificationContext } from "../pushNotifications.js"
+import { Job, Queue } from "bullmq"
+import { environment } from "../../environment.js"
+
+const deliveryCheckQueue = new Queue('checkPushNotificationDelivery', {
+  connection: environment.bullmqConnection,
+  defaultJobOptions: {
+    removeOnComplete: true,
+    attempts: 3,
+    backoff: {
+      type: 'exponential',
+      delay: 1000
+    }
+  }
+})
+
+const verbMap = {
+  LIKE: 'liked',
+  REWOOT: 'rewooted',
+  MENTION: 'replied to',
+  QUOTE: 'quoted',
+  EMOJIREACT: 'reacted to',
+}
+
+const expoClient = new Expo()
+
+function getNotificationTitle(notification: NotificationBody, context?: NotificationContext) {
+  if (notification.notificationType === 'FOLLOW') {
+    return 'New user followed you'
+  }
+
+  if (notification.notificationType === 'EMOJIREACT' && context?.emoji) {
+    return `${context?.userUrl || 'someone'} reacted with ${context.emoji} to your post`
+  }
+
+  return `${context?.userUrl || 'someone'} ${verbMap[notification.notificationType]} your post`
+}
+
+function getNotificationBody(notification: NotificationBody, context?: NotificationContext) {
+  if (notification.notificationType === 'FOLLOW') {
+    return context?.userUrl ? `@${context?.userUrl.replace(/^@/, '')}` : ''
+  }
+
+  return `${context?.postContent}`
+}
+
+type PushNotificationPayload = {
+  notification: NotificationBody
+  context?: NotificationContext
+}
+
+export async function sendPushNotification(job: Job<PushNotificationPayload>) {
+  const { notification, context } = job.data
+  const userId = notification.notifiedUserId
+  const tokenRows = await PushNotificationToken.findAll({
+    where: {
+      userId
+    }
+  })
+
+  if (tokenRows.length === 0) {
+    return
+  }
+
+  const payloads = tokenRows.map((row) => ({
+    to: row.token,
+    sound: 'default',
+    title: getNotificationTitle(notification, context),
+    body: getNotificationBody(notification, context),
+    data: notification
+  }))
+
+  // this will chunk the payloads into chunks of 1000 (max) and compress notifications with similar content
+  const chunks = expoClient.chunkPushNotifications(payloads)
+  const okTickets = []
+
+  // TODO: handle in a queue with retry logic and exponential backoff
+  for (const chunk of chunks) {
+    try {
+      const responses = await expoClient.sendPushNotificationsAsync(chunk)
+      for (const response of responses) {
+        if (response.status === 'ok') {
+          okTickets.push(response.id)
+        } else {
+          await handleDeliveryError(response)
+        }
+      }
+    } catch (error) {
+      logger.error(error)
+      // TODO: retry sending the notification after some time
+    }
+  }
+
+  await scheduleNotificationCheck(okTickets)
+}
+
+function scheduleNotificationCheck(ticketIds: string[]) {
+  const delay = 1000 * 60 * 30 // 30 minutes
+  return deliveryCheckQueue.add('checkPushNotificationDelivery', { ticketIds}, { delay })
+}

packages/backend/utils/workers.ts

···
 import { processRemotePostView } from './queueProcessors/processRemotePostView.js'
 import { processRemoteMedia } from './queueProcessors/remoteMediaProcessor.js'
 import { processFirehose } from '../atproto/workers/processFirehoseWorker.js'
+import { sendPushNotification } from './queueProcessors/sendPushNotification.js'
+import { checkPushNotificationDelivery } from './queueProcessors/checkPushNotificationDelivery.js'
 
 logger.info('starting workers')
 const workerInbox = new Worker('inbox', (job: Job) => inboxWorker(job), {
···
     })
   : null
 
+const workerSendPushNotification = new Worker(
+  'sendPushNotification',
+  async (job: Job) => await sendPushNotification(job),
+  {
+    connection: environment.bullmqConnection,
+    metrics: {
+      maxDataPoints: MetricsTime.ONE_WEEK * 2
+    },
+    concurrency: environment.workers.medium,
+  }
+)
+
+const workerCheckPushNotificationDelivery = new Worker(
+  'checkPushNotificationDelivery',
+  async (job: Job) => await checkPushNotificationDelivery(job),
+  {
+    connection: environment.bullmqConnection,
+    metrics: {
+      maxDataPoints: MetricsTime.ONE_WEEK * 2
+    },
+    concurrency: environment.workers.medium,
+  }
+)
+
 const workers = [
   workerInbox,
   workerDeletePost,
···
   workerProcessRemotePostView,
   workerSendPostChunk,
   workerProcessRemotePostView,
-  workerProcessRemoteMediaData
+  workerProcessRemoteMediaData,
+  workerSendPushNotification,
+  workerCheckPushNotificationDelivery
 ]
 if (environment.enableBsky) {
   workers.push(workerProcessFirehose as Worker)
···
   workerDeletePost,
   workerProcessRemotePostView,
   workerProcessRemoteMediaData,
-  workerProcessFirehose
+  workerProcessFirehose,
+  workerSendPushNotification,
+  workerCheckPushNotificationDelivery
 }