const waJobModel = require('../model/waJob.model') const { logInfo, logError } = require('../utils/logger') /** * Enqueue job kirim WA ke queue (asynchronous) * Context: WA_CREATE_LAPORAN, WA_REMINDER, dll */ exports.enqueueWa = async (templateId, contacts, body, context = '', buttons = []) => { try { const job = await waJobModel.create({ templateId, contacts, body, buttons, context, status: 'pending', nextRunAt: new Date(), }) logInfo('WA_JOB', 'Job enqueued', { jobId: job._id, context, contactsCount: contacts.length }) return job } catch (err) { logError('WA_JOB', 'Failed to enqueue WA job', { error: err.message, context }) throw err } } /** * Claim & process next pending job * Menggunakan findOneAndUpdate untuk atomic lock (cegah double-process di multi-instance) */ exports.processNextJob = async () => { try { const job = await waJobModel.findOneAndUpdate( { status: 'pending', nextRunAt: { $lte: new Date() }, }, { $set: { status: 'processing', claimedAt: new Date(), }, }, { new: true, sort: { nextRunAt: 1 } } ) return job } catch (err) { logError('WA_JOB', 'Failed to claim next job', { error: err.message }) return null } } /** * Mark job sukses */ exports.markJobSuccess = async (jobId, response) => { try { await waJobModel.findByIdAndUpdate(jobId, { status: 'sent', response, lastError: null, }) logInfo('WA_JOB', 'Job marked as sent', { jobId }) } catch (err) { logError('WA_JOB', 'Failed to mark job success', { error: err.message, jobId }) } } /** * Mark job gagal, increment attempt, jadwal retry */ exports.markJobFailed = async (jobId, error, attempt) => { try { const maxAttempt = 3 const backoffMs = Math.pow(2, attempt) * 1000 // 2s, 4s, 8s const nextRunAt = new Date(Date.now() + backoffMs) const willRetry = attempt < maxAttempt await waJobModel.findByIdAndUpdate(jobId, { status: willRetry ? 'pending' : 'failed', lastError: error.message || String(error), attempt, nextRunAt: willRetry ? nextRunAt : null, claimedAt: null, }) logInfo('WA_JOB', `Job ${willRetry ? 'will retry' : 'permanently failed'}`, { jobId, attempt, nextRunAtMs: willRetry ? backoffMs : null, error: error.message, }) } catch (err) { logError('WA_JOB', 'Failed to mark job failed', { error: err.message, jobId }) } } /** * Reset in-progress jobs saat startup (kalau server crashed saat processing) */ exports.resetStuckJobs = async () => { try { const result = await waJobModel.updateMany( { status: 'processing', claimedAt: { $lt: new Date(Date.now() - 60 * 1000) } }, { $set: { status: 'pending', claimedAt: null } } ) if (result.modifiedCount > 0) { logInfo('WA_JOB', 'Reset stuck jobs', { count: result.modifiedCount }) } } catch (err) { logError('WA_JOB', 'Failed to reset stuck jobs', { error: err.message }) } } /** * Get job status by ID */ exports.getJobStatus = async (jobId) => { try { return await waJobModel.findById(jobId) } catch (err) { logError('WA_JOB', 'Failed to get job status', { error: err.message, jobId }) return null } }