${bodyText}
${linkHtml}`, + undefined, + undefined, + notif.portId ?? undefined, ); await db diff --git a/src/lib/queue/workers/reports.ts b/src/lib/queue/workers/reports.ts index e9c8d2c6..b5a1e1a8 100644 --- a/src/lib/queue/workers/reports.ts +++ b/src/lib/queue/workers/reports.ts @@ -22,69 +22,92 @@ export const reportsWorker = new Worker( // weekly/monthly reports that's an instant flood of dupe // emails to recipients. Now we compute the next fire from // the cron expression and UPDATE the row atomically. + // + // L6: this poller does a select-due → per-row update. With a + // single `crm-worker` (concurrency 1) that's safe, but the moment + // `MULTI_NODE_DEPLOYMENT` adds a second replica two pollers would + // both read the same due rows and double-fire (duplicate runs + + // email blasts). We now atomically CLAIM due rows with + // `FOR UPDATE SKIP LOCKED` inside a transaction: a concurrent + // replica skips rows this tx already holds, so each due row is + // claimed by exactly one poller. `nextRunAt` is row-specific + // (cron-derived) so we keep the per-row update — the row lock, + // not a bulk UPDATE, is what makes the claim atomic. Enqueues are + // deferred to AFTER commit so a rolled-back claim never leaves an + // orphaned generate-report job. const { db } = await import('@/lib/db'); const { scheduledReports } = await import('@/lib/db/schema/operations'); const { generatedReports } = await import('@/lib/db/schema/operations'); const { eq, and, lte } = await import('drizzle-orm'); const { CronExpressionParser } = await import('cron-parser'); - const dueReports = await db - .select() - .from(scheduledReports) - .where( - and(eq(scheduledReports.isActive, true), lte(scheduledReports.nextRunAt, new Date())), - ); + const enqueueIds: string[] = []; - for (const report of dueReports) { - const { getQueue } = await import('@/lib/queue'); + await db.transaction(async (tx) => { + const dueReports = await tx + .select() + .from(scheduledReports) + .where( + and(eq(scheduledReports.isActive, true), lte(scheduledReports.nextRunAt, new Date())), + ) + .for('update', { skipLocked: true }); - // Compute next_run_at BEFORE the enqueue so a failure in the - // parse path (malformed cron) doesn't get repeat-fired. - let nextRunAt: Date | null = null; - try { - nextRunAt = CronExpressionParser.parse(report.schedule, { - currentDate: new Date(), - tz: process.env.SCHEDULER_TZ ?? 'Europe/Warsaw', - }) - .next() - .toDate(); - } catch (err) { - logger.error( - { err, reportId: report.id, schedule: report.schedule }, - 'Failed to parse cron schedule for scheduled report; pausing it', - ); - // Disable the row so we don't re-attempt the malformed cron - // every minute. - await db + for (const report of dueReports) { + // Compute next_run_at BEFORE the enqueue so a failure in the + // parse path (malformed cron) doesn't get repeat-fired. + let nextRunAt: Date | null = null; + try { + nextRunAt = CronExpressionParser.parse(report.schedule, { + currentDate: new Date(), + tz: process.env.SCHEDULER_TZ ?? 'Europe/Warsaw', + }) + .next() + .toDate(); + } catch (err) { + logger.error( + { err, reportId: report.id, schedule: report.schedule }, + 'Failed to parse cron schedule for scheduled report; pausing it', + ); + // Disable the row so we don't re-attempt the malformed cron + // every minute. + await tx + .update(scheduledReports) + .set({ isActive: false, updatedAt: new Date() }) + .where(eq(scheduledReports.id, report.id)); + continue; + } + + await tx .update(scheduledReports) - .set({ isActive: false, updatedAt: new Date() }) + .set({ nextRunAt, updatedAt: new Date() }) .where(eq(scheduledReports.id, report.id)); - continue; + + const [genReport] = await tx + .insert(generatedReports) + .values({ + portId: report.portId, + scheduledReportId: report.id, + reportType: report.reportType, + name: `${report.name} - ${new Date().toISOString().split('T')[0]}`, + status: 'queued', + parameters: (report.config as RecordYour scheduled ${run.kind} report is attached.
`; - for (const recipient of recipients) { - await sendEmail(recipient.email, subject, html, undefined, undefined, run.portId, [ - { fileId: fileRow.id, filename: fileRow.originalName ?? `${run.kind}-report.pdf` }, - ]); - } - + // M9: stamp `emailedAt` BEFORE sending so a transient SMTP failure + // mid-loop cannot cause a BullMQ retry to re-send to the recipients + // who already received the report (the early-return above now fires). await updateReportRunStatus(run.id, run.portId, { status: 'complete', emailedAt: new Date(), }); + + // M9: log-not-throw per recipient — one bad address must not re-blast + // the rest on retry (the run is already marked emailed). At-most-once + // delivery per recipient; a failed send is logged and dropped. + for (const recipient of recipients) { + try { + await sendEmail(recipient.email, subject, html, undefined, undefined, run.portId, [ + { fileId: fileRow.id, filename: fileRow.originalName ?? `${run.kind}-report.pdf` }, + ]); + } catch (err) { + logger.error( + { err, reportRunId: run.id, recipient: recipient.email }, + 'Failed to email scheduled report to recipient; skipping (run already marked emailed)', + ); + } + } } /** @@ -456,7 +515,10 @@ async function renderStandaloneReportRun(run: ReportRun): Promise