require('dotenv').config({ path: require('path').join(__dirname, '../../.env') }); const mongoose = require('mongoose'); const { connectDB } = require('../config/database'); const WhatsAppQueue = require('../models/WhatsAppQueue'); const MessagingCampaign = require('../models/MessagingCampaign'); const Event = require('../models/Event'); const WhatsAppTemplate = require('../models/WhatsAppTemplate'); const Participant = require('../models/Participant'); const { getOrgWhatsAppCredentials, sendTemplateMessage, parseWhatsAppCloudMessagesSendResponse, } = require('../utils/metaWhatsApp'); const { buildWhatsAppTemplatePayload } = require('../utils/waTemplateSendParams'); const { getCampaignContentObject } = require('../utils/campaignContent'); const { flattenParticipantForEvent } = require('../utils/participantMembership'); const POLL_MS = Math.max(500, parseInt(process.env.WHATSAPP_WORKER_POLL_MS || '2000', 10)); const MAX_ATTEMPTS = Math.max(1, parseInt(process.env.WHATSAPP_WORKER_MAX_ATTEMPTS || '3', 10)); const MIN_INTERVAL_MS = Math.max(0, parseInt(process.env.WHATSAPP_WORKER_MIN_INTERVAL_MS || '400', 10)); let lastSendAt = 0; function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); } async function throttle() { if (MIN_INTERVAL_MS <= 0) return; const now = Date.now(); const wait = lastSendAt + MIN_INTERVAL_MS - now; if (wait > 0) await sleep(wait); lastSendAt = Date.now(); } async function finalizeCampaignIfDone(campaignId) { const pending = await WhatsAppQueue.countDocuments({ campaign: campaignId, status: { $in: ['pending', 'processing'] } }); if (pending > 0) return; const camp = await MessagingCampaign.findById(campaignId); if (!camp || camp.status === 'stopped') return; const failed = await WhatsAppQueue.countDocuments({ campaign: campaignId, status: 'failed' }); const sent = await WhatsAppQueue.countDocuments({ campaign: campaignId, status: 'sent' }); if (sent === 0 && failed > 0) { camp.status = 'failed'; } else { camp.status = 'sent'; } camp.sentAt = new Date(); await camp.save(); } async function processOne() { const now = new Date(); const job = await WhatsAppQueue.findOneAndUpdate( { status: 'pending', processAfter: { $lte: now }, }, { $set: { status: 'processing', lockedAt: now }, $inc: { attempts: 1 }, }, { sort: { createdAt: 1 }, new: true } ); if (!job) return false; const campaign = await MessagingCampaign.findById(job.campaign); if (!campaign) { job.status = 'cancelled'; await job.save(); return true; } if (campaign.status === 'stopped') { job.status = 'cancelled'; await job.save(); return true; } if (campaign.status === 'scheduled') { campaign.status = 'sending'; await campaign.save(); } const orgId = job.organization?._id || job.organization; const creds = orgId && (await getOrgWhatsAppCredentials(String(orgId))); if (!creds || !creds.accessToken) { job.status = 'failed'; job.lastError = 'WhatsApp not configured (token)'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } if (!creds.phoneNumberId) { job.status = 'failed'; job.lastError = 'WhatsApp phone number ID missing (set in Settings)'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } const eventDoc = await Event.findById(campaign.event).lean(); const tplId = getCampaignContentObject(campaign).whatsappTemplateId; const tpl = tplId && mongoose.Types.ObjectId.isValid(String(tplId)) ? await WhatsAppTemplate.findById(tplId).lean() : null; if (!eventDoc || !tpl) { job.status = 'failed'; job.lastError = 'Template or event missing'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } const p = job.participant ? await Participant.findById(job.participant).lean() : null; if (!p) { job.status = 'failed'; job.lastError = 'Participant not found'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } const flat = flattenParticipantForEvent(p, campaign.event); if (!flat) { job.status = 'failed'; job.lastError = 'Participant not in event'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } let built; try { const content = getCampaignContentObject(campaign); built = buildWhatsAppTemplatePayload(tpl, content, flat, eventDoc); } catch (e) { job.status = 'failed'; job.lastError = e.message || 'build template failed'; await job.save(); await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } try { await throttle(); const sendBody = await sendTemplateMessage(creds.phoneNumberId, creds.accessToken, job.to, built.template); const { whatsappMessageId, recipientWaId } = parseWhatsAppCloudMessagesSendResponse(sendBody); if (whatsappMessageId) job.whatsappMessageId = whatsappMessageId; if (recipientWaId) job.recipientWaId = recipientWaId; } catch (e) { const msg = e.message || 'Meta send failed'; if (job.attempts >= MAX_ATTEMPTS) { job.status = 'failed'; job.lastError = msg; } else { job.status = 'pending'; job.lastError = msg; job.processAfter = new Date(Date.now() + 60_000 * job.attempts); } await job.save(); if (job.status === 'failed') await finalizeFailed(campaign, job); await finalizeCampaignIfDone(campaign._id); return true; } job.status = 'sent'; job.sentAt = new Date(); job.lastError = undefined; await job.save(); campaign.stats = campaign.stats || {}; campaign.stats.sent = (campaign.stats.sent || 0) + 1; await campaign.save(); await finalizeCampaignIfDone(campaign._id); return true; } async function finalizeFailed(campaign, job) { if (job.status !== 'failed') return; campaign.stats = campaign.stats || {}; campaign.stats.failed = (campaign.stats.failed || 0) + 1; await campaign.save(); } async function run() { await connectDB(); console.log( '[whatsappQueueWorker] started, poll', POLL_MS, 'ms, min interval', MIN_INTERVAL_MS, 'ms', ); for (;;) { try { const did = await processOne(); if (!did) await sleep(POLL_MS); } catch (e) { console.error('[whatsappQueueWorker]', e); await sleep(POLL_MS); } } } if (require.main === module) { run(); } module.exports = { processOne, run };