Messenger/packages/backend/src/modules/notifications/notifications.queue.ts

import { Queue } from 'bullmq';
import { Redis } from 'ioredis';

// A new message that may warrant a push. Only ids travel through the queue; the
// worker re-loads fresh state so nothing goes stale between enqueue and delivery.
export const MESSAGE_JOB = 'message';

export interface MessageJobData {
  conversationId: string;
  messageId: string;
  senderId: string;
}

export interface NotificationQueue {
  enqueueMessage(data: MessageJobData): Promise<void>;
  close(): Promise<void>;
}

// BullMQ requires a dedicated connection with retry-per-request disabled.
export const createBullConnection = (redisUrl: string): Redis =>
  new Redis(redisUrl, { maxRetriesPerRequest: null });

export const createNotificationQueue = (
  queueName: string,
  connection: Redis,
): NotificationQueue => {
  const queue = new Queue<MessageJobData>(queueName, {
    connection,
    defaultJobOptions: {
      attempts: 5,
      backoff: { type: 'exponential', delay: 2000 },
      removeOnComplete: { count: 1000 },
      removeOnFail: { count: 5000 },
    },
  });

  return {
    enqueueMessage: async (data) => {
      await queue.add(MESSAGE_JOB, data);
    },
    close: async () => {
      await queue.close();
    },
  };
};