vjrj
4f8d866226
El dry-run contra prod destapo 'Error: Custom Id cannot contain :': el jobId era canal:notificationId. Cambiado a canal_notificationId. Test que lo fija.
47 lines
1.5 KiB
TypeScript
47 lines
1.5 KiB
TypeScript
import { Queue, type ConnectionOptions, type QueueOptions } from 'bullmq';
|
|
import type { Config } from './config.js';
|
|
import type { Channel, SendJob } from './types.js';
|
|
|
|
export const QUEUE_NAMES: Record<Channel, string> = {
|
|
fcm: 'tcef-fcm',
|
|
email: 'tcef-email',
|
|
};
|
|
|
|
/**
|
|
* BullMQ connection options. We hand BullMQ plain options (not a shared ioredis
|
|
* instance) so it manages its own blocking/non-blocking connections and sets
|
|
* maxRetriesPerRequest:null on workers itself.
|
|
*/
|
|
export function redisConnection(cfg: Config): ConnectionOptions {
|
|
return {
|
|
host: cfg.redis.host,
|
|
port: cfg.redis.port,
|
|
password: cfg.redis.password,
|
|
db: cfg.redis.db,
|
|
};
|
|
}
|
|
|
|
export function createQueues(cfg: Config): Record<Channel, Queue<SendJob>> {
|
|
const opts: QueueOptions = {
|
|
connection: redisConnection(cfg),
|
|
defaultJobOptions: {
|
|
attempts: cfg.retries.attempts,
|
|
backoff: { type: 'exponential', delay: cfg.retries.backoffMs },
|
|
removeOnComplete: { count: 1000 },
|
|
removeOnFail: false, // keep failures = dead-letter for inspection
|
|
},
|
|
};
|
|
return {
|
|
fcm: new Queue<SendJob>(QUEUE_NAMES.fcm, opts),
|
|
email: new Queue<SendJob>(QUEUE_NAMES.email, opts),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Job id = `${channel}_${notificationId}` so BullMQ itself dedupes enqueues:
|
|
* re-polling the same pending doc will not create a second job.
|
|
* (BullMQ rejects custom job ids containing ':', hence the underscore.)
|
|
*/
|
|
export function jobId(job: SendJob): string {
|
|
return `${job.channel}_${job.notificationId}`;
|
|
}
|