// Phase 5C §14 — a simple durable worker, same pattern this codebase already uses for // sessions.js's session sweeper and security.js's login-attempt sweeper: a plain setInterval // poll against Postgres, no Kafka/RabbitMQ/Redis. notifications.js's claimNext() does the // actual concurrency-safety work (FOR UPDATE SKIP LOCKED) — this file just calls it on a timer // and hands the claimed row to the configured email provider. const notifications = require('./notifications'); const templates = require('./email/templates'); const { getEmailProvider } = require('./email/provider'); const metrics = require('./metrics'); const POLL_MS = Number(process.env.NOTIFICATION_WORKER_POLL_MS) || 3000; function buildAppUrl(path) { const origin = process.env.PUBLIC_APP_ORIGIN; if (!origin) return null; try { const u = new URL(origin); return `${u.origin}${path}`; } catch { return null; // malformed origin — treated as unconfigured, not a crash } } // Renders the right template for a queued row. Returns null (a permanent, non-retryable // failure) if required config (PUBLIC_APP_ORIGIN for a token-bearing type) is missing — §7's // "production should fail safely if required public-origin configuration is missing." function renderFor(row) { const displayName = row.metadata?.display_name || 'there'; if (row.notification_type === 'activation') { const url = buildAppUrl(`/?activate=${encodeURIComponent(row.metadata?.token || '')}`); if (!url || !row.metadata?.token) return null; return templates.activation({ displayName, programName: row.metadata?.program_name, activationUrl: url, expiresInDays: 7 }); } if (row.notification_type === 'password_reset') { const url = buildAppUrl(`/?reset=${encodeURIComponent(row.metadata?.token || '')}`); if (!url || !row.metadata?.token) return null; return templates.passwordReset({ displayName, resetUrl: url, expiresInMinutes: 60 }); } if (row.notification_type === 'enrollment_confirmation') { const url = buildAppUrl('/') || '/'; return templates.enrollmentConfirmation({ displayName, programName: row.metadata?.program_name || 'your program', signInUrl: url }); } if (row.notification_type === 'password_changed') { return templates.passwordChanged({ displayName }); } // Phase 8 — registration-stage notifications. No token/public-origin dependency, so these // never fail on missing PUBLIC_APP_ORIGIN config the way activation/reset do. if (row.notification_type === 'registration_received') { return templates.registrationReceived({ displayName, offeringName: row.metadata?.offering_name || 'your offering' }); } if (row.notification_type === 'registration_confirmed') { return templates.registrationConfirmed({ displayName, offeringName: row.metadata?.offering_name || 'your offering' }); } if (row.notification_type === 'registration_needs_action') { return templates.registrationNeedsAction({ displayName, offeringName: row.metadata?.offering_name || 'your offering', reason: row.metadata?.reason }); } if (row.notification_type === 'agreement_required') { return templates.agreementRequired({ displayName, offeringName: row.metadata?.offering_name || 'your offering', agreementTitle: row.metadata?.agreement_title || 'a required agreement' }); } if (row.notification_type === 'camp_operational_update') { return templates.campOperationalUpdate({ displayName, offeringName: row.metadata?.offering_name || 'your offering', message: row.metadata?.message || '' }); } return null; } async function processOne(pool) { const row = await notifications.claimNext(pool); if (!row) return false; const { EMAIL_PROVIDER } = require('./email/provider'); const rendered = renderFor(row); if (!rendered) { await notifications.markFailedOrRetry(pool, row, { errorCode: 'template_or_config_error', retryable: false }); metrics.recordNotificationAttempt(EMAIL_PROVIDER, 'failed'); return true; } const provider = getEmailProvider(); try { const result = await provider.send({ to: row.recipient_email, fromName: process.env.EMAIL_FROM_NAME || 'RootNode Academy', fromAddress: process.env.EMAIL_FROM_ADDRESS, replyTo: process.env.EMAIL_REPLY_TO, subject: rendered.subject, text: rendered.text, html: rendered.html, }); // §20 — a dev-only preview (never in production, regardless of provider) so a test/admin // can inspect exactly what would have been sent, including the activation/reset URL, // through the dedicated dev-only route — never through the normal admin notification view. const isProduction = process.env.COOKIE_SECURE === '1'; const devPreview = (!isProduction && result.preview) ? result.preview : null; await notifications.markSent(pool, row.notification_id, result.providerMessageId, EMAIL_PROVIDER, devPreview); metrics.recordNotificationAttempt(EMAIL_PROVIDER, 'sent'); } catch (e) { const retryable = e.errorClass !== 'permanent'; const outcome = await notifications.markFailedOrRetry(pool, row, { errorCode: e.errorCode || 'unknown_error', retryable }); metrics.recordNotificationAttempt(EMAIL_PROVIDER, outcome === 'retrying' ? 'retry' : 'failed'); console.error(`[notification-worker] send failed (${row.notification_type}, ${outcome}): ${e.errorCode || e.message}`); } return true; } function startNotificationWorker(pool) { const tick = async () => { try { // Drain everything currently claimable before waiting out the next poll interval — // keeps a burst (e.g. a CSV import's worth of activations) from trickling out one per // POLL_MS when they could all go immediately. while (await processOne(pool)) { /* keep draining */ } } catch (e) { console.error('[notification-worker] tick failed:', e.message); } }; const handle = setInterval(tick, POLL_MS); handle.unref(); // never keeps the process alive on its own, matching sessions.js's sweeper tick(); // don't wait a full interval for the first pass return handle; } module.exports = { startNotificationWorker, processOne, renderFor, buildAppUrl };