/opt/mawid/apps/api/src/whatsapp
Edit: /opt/mawid/apps/api/src/whatsapp/wa-outbound.worker.ts (2723B)
import {
Inject,
Injectable,
Logger,
OnModuleDestroy,
OnModuleInit,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Worker, type Job } from 'bullmq';
import IORedis from 'ioredis';
import { PrismaService } from '../prisma/prisma.service';
import { WA_HTTP_CLIENT, type WaHttpClient } from './wa-http.client';
import { WA_OUTBOUND_QUEUE_NAME, type WaOutboundJob } from './wa-queue';
import type { WaOutboundRequest } from './wa-types';
@Injectable()
export class WaOutboundWorker implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(WaOutboundWorker.name);
private worker?: Worker
;
constructor(
private readonly prisma: PrismaService,
private readonly config: ConfigService,
@Inject(WA_HTTP_CLIENT) private readonly httpClient: WaHttpClient,
) {}
onModuleInit() {
const connection = new IORedis(this.config.getOrThrow('REDIS_URL'), {
maxRetriesPerRequest: null,
});
this.worker = new Worker(
WA_OUTBOUND_QUEUE_NAME,
(job) => this.handle(job),
{ connection, concurrency: 5 },
);
this.worker.on('failed', (job, err) => {
this.logger.warn(`send attempt ${job?.attemptsMade} failed: ${err.message}`);
});
}
async onModuleDestroy() {
await this.worker?.close();
}
async handle(job: Job): Promise {
const message = await this.prisma.message.findUniqueOrThrow({
where: { id: job.data.messageId },
include: { conversation: { include: { clinic: true } } },
});
const payload = (message.payload ?? {}) as { request?: WaOutboundRequest; status?: string };
if (!payload.request) throw new Error(`Message ${message.id} has no outbound request payload`);
if (payload.status === 'sent') return; // idempotent on retries after partial failure
const phoneNumberId =
message.conversation.clinic.waPhoneNumberId ??
this.config.get('WA_PHONE_NUMBER_ID') ??
'';
try {
const res = await this.httpClient.send(phoneNumberId, payload.request);
await this.prisma.message.update({
where: { id: message.id },
data: {
waMessageId: res.messages[0]?.id,
payload: { ...payload, status: 'sent' } as object,
},
});
} catch (err) {
const isFinalAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1);
if (isFinalAttempt) {
await this.prisma.message.update({
where: { id: message.id },
data: { payload: { ...payload, status: 'failed', error: String(err) } as object },
});
}
throw err; // let BullMQ schedule the backoff retry
}
}
}