Repository Public
1import { sql } from "../db";2import { env, nowMs } from "../env";3import { manageTokenFor } from "../crypto";4import { invitation } from "../ics";5import { MailNotConfigured, sendMail } from "../mail/transport";6import { renderMail, type Locale, type MailContext } from "../mail/templates";7import type { NotificationKind } from "./notify";89const BACKOFF_MIN = [1, 5, 30, 120, 360]; // then dead10const ADMIN_KINDS = new Set<string>(["admin_new_request", "admin_new_booking", "admin_visitor_cancelled", "admin_request_withdrawn", "admin_reschedule_request", "admin_proposal_accepted"]);11const WITH_INVITE = new Set<string>(["booking_confirmed", "booking_rescheduled"]);1213interface JobRow { id: string; user_id: string; kind: NotificationKind; dedup_key: string; to_email: string; locale: Locale; request_id: string | null; booking_id: string | null; payload: Record<string, unknown>; attempts: number }1415interface Loaded { ctx: MailContext; ics: { filename: string; content: string; method: "REQUEST" | "CANCEL" } | null; replyTo: string | null; skip?: string }1617/** The message is rendered from the CURRENT state of the database, never from a snapshot taken when it was queued. */18async function load(job: JobRow): Promise<Loaded> {19 const [o] = await sql<{ display_name: string; email: string; public_name: string; timezone: string; brand_color: string; reply_to: string | null; notify_email: string | null }[]>`20 SELECT u.display_name, u.email::text, s.public_name, s.timezone, s.brand_color, s.reply_to::text, s.notify_email::text FROM users u JOIN settings s ON s.user_id = u.id WHERE u.id = ${job.user_id}`;21 const ownerName = o.public_name || o.display_name;22 const admin = ADMIN_KINDS.has(job.kind);23 const base: MailContext = {24 locale: job.locale, kind: job.kind, ownerName, brandColor: o.brand_color, typeTitle: "", visitorName: "", start: null, end: null, tz: o.timezone,25 locationKind: "video", locationText: "", instructions: "", message: "", manageUrl: null, adminUrl: `${env.appUrl}/admin`,26 };27 if (job.kind === "test") return { ctx: base, ics: null, replyTo: null };2829 const [r] = job.request_id ? await sql<{ id: string; status: string; starts_at: Date; ends_at: Date; visitor_tz: string; decline_message: string; proposed_starts_at: Date | null; proposed_ends_at: Date | null; proposal_message: string; reschedule_of_booking_id: string | null; event_type_id: string; contact_id: string | null }[]>`30 SELECT id, status, starts_at, ends_at, visitor_tz, decline_message, proposed_starts_at, proposed_ends_at, proposal_message, reschedule_of_booking_id, event_type_id, contact_id FROM booking_requests WHERE id = ${job.request_id}` : [];31 const [b] = job.booking_id ? await sql<{ id: string; status: string; starts_at: Date; ends_at: Date; visitor_tz: string; location_kind: string; location_text: string; ics_uid: string; ics_sequence: number; event_type_id: string | null; contact_id: string | null; cancelled_by: string | null; title: string }[]>`32 SELECT id, status, starts_at, ends_at, visitor_tz, location_kind, location_text, ics_uid, ics_sequence, event_type_id, contact_id, cancelled_by, title FROM bookings WHERE id = ${job.booking_id}` : [];33 if (!r && !b) return { ctx: base, ics: null, replyTo: null, skip: "subject no longer exists" };3435 const typeId = b?.event_type_id ?? r?.event_type_id ?? null;36 const [et] = typeId ? await sql<{ title: string; title_en: string; location_kind: string; instructions: string; instructions_en: string; reminder_hours: number | null }[]>`SELECT title, title_en, location_kind, instructions, instructions_en, reminder_hours FROM event_types WHERE id = ${typeId}` : [];37 const contactId = b?.contact_id ?? r?.contact_id ?? null;38 const [c] = contactId ? await sql<{ name: string; email: string }[]>`SELECT name, email::text FROM contacts WHERE id = ${contactId}` : [];39 const en = job.locale === "en";40 const ctx: MailContext = {41 ...base,42 typeTitle: (en && et?.title_en) || et?.title || b?.title || "",43 visitorName: c?.name ?? "",44 tz: admin ? o.timezone : (b?.visitor_tz ?? r?.visitor_tz ?? o.timezone),45 locationKind: b?.location_kind ?? et?.location_kind ?? "video",46 instructions: (en && et?.instructions_en) || et?.instructions || "",47 hoursBefore: et?.reminder_hours ?? null,48 };49 // Visitor link: rebuilt from the token row id; only the hash is stored.50 if (!admin && r) {51 const [t] = await sql<{ id: string }[]>`52 SELECT t.id FROM booking_tokens t JOIN booking_requests q ON q.id = t.request_id53 WHERE (q.id = ${r.id} OR (${b?.id ?? null}::uuid IS NOT NULL AND q.booking_id = ${b?.id ?? null}))54 AND t.revoked_at IS NULL AND t.expires_at > now() ORDER BY t.created_at LIMIT 1`;55 if (t) ctx.manageUrl = `${env.appUrl}/manage/${manageTokenFor(t.id)}`;56 }5758 let ics: Loaded["ics"] = null;59 const mkIcs = (status: "CONFIRMED" | "CANCELLED") => b && c ? {60 filename: status === "CANCELLED" ? "annulation.ics" : "invitation.ics", method: status === "CANCELLED" ? "CANCEL" as const : "REQUEST" as const,61 content: invitation({62 uid: b.ics_uid, sequence: b.ics_sequence, start: b.starts_at, end: b.ends_at, status,63 summary: `${ctx.typeTitle} — ${ownerName}`, location: b.location_text || undefined,64 description: ctx.manageUrl ?? undefined, organizerName: ownerName, organizerEmail: o.reply_to ?? o.notify_email ?? o.email, attendeeName: c.name, attendeeEmail: c.email,65 }),66 } : null;6768 switch (job.kind) {69 case "request_received": case "admin_new_request": case "request_expired": case "request_declined": case "admin_request_withdrawn":70 ctx.start = r!.starts_at; ctx.end = r!.ends_at;71 if (job.kind === "request_declined") ctx.message = r!.decline_message;72 if ((job.kind === "request_received" || job.kind === "admin_new_request") && r!.status !== "pending") return { ctx, ics, replyTo: null, skip: `request is now ${r!.status}` };73 break;74 case "proposal":75 if (r!.status !== "reschedule_requested" || !r!.proposed_starts_at) return { ctx, ics, replyTo: null, skip: "proposal no longer open" };76 ctx.start = r!.proposed_starts_at; ctx.end = r!.proposed_ends_at; ctx.otherStart = r!.starts_at; ctx.message = r!.proposal_message;77 break;78 case "reschedule_declined":79 if (!b) return { ctx, ics, replyTo: null, skip: "booking missing" };80 ctx.start = b.starts_at; ctx.end = b.ends_at; ctx.message = r?.decline_message ?? "";81 break;82 case "admin_reschedule_request":83 ctx.start = r!.starts_at; ctx.end = r!.ends_at; ctx.otherStart = b?.starts_at ?? null;84 if (r!.status !== "pending") return { ctx, ics, replyTo: null, skip: `request is now ${r!.status}` };85 break;86 case "booking_confirmed": case "booking_rescheduled": case "admin_new_booking": case "admin_proposal_accepted": case "reminder":87 if (!b || b.status !== "confirmed") return { ctx, ics, replyTo: null, skip: "booking is not confirmed any more" };88 if (job.kind === "reminder" && (job.payload.sequence !== b.ics_sequence || b.starts_at.getTime() <= nowMs())) return { ctx, ics, replyTo: null, skip: "reminder is obsolete" };89 // A newer change supersedes an unsent confirmation: only the latest SEQUENCE is announced.90 if (job.kind === "booking_rescheduled" && !job.dedup_key.endsWith(`:${b.ics_sequence}`)) return { ctx, ics, replyTo: null, skip: "superseded by a later change" };91 ctx.start = b.starts_at; ctx.end = b.ends_at; ctx.locationText = b.location_text;92 if (WITH_INVITE.has(job.kind)) ics = mkIcs("CONFIRMED");93 break;94 case "booking_cancelled": case "admin_visitor_cancelled":95 if (!b) return { ctx, ics, replyTo: null, skip: "booking missing" };96 ctx.start = b.starts_at; ctx.end = b.ends_at; ctx.cancelledBy = b.cancelled_by ?? undefined;97 if (job.kind === "booking_cancelled") ics = mkIcs("CANCELLED");98 break;99 }100 return { ctx, ics, replyTo: admin ? (c?.email ?? null) : (o.reply_to ?? null) };101}102103export interface OutboxStats { sent: number; failed: number; dead: number; unconfigured: number; skipped: number }104105/**106 * Sends what is due. A job is "sent" only when the transport returned a message id. Failures are retried with107 * backoff, then marked dead; without a configured transport the job is parked as "unconfigured" (visible in the108 * admin area, re-queued from there) — it is never reported as sent.109 */110export async function processOutbox(limit = 20): Promise<OutboxStats> {111 const stats: OutboxStats = { sent: 0, failed: 0, dead: 0, unconfigured: 0, skipped: 0 };112 // Recover jobs left "sending" by a process that died.113 await sql`UPDATE notification_jobs SET status = 'queued' WHERE status = 'sending' AND locked_at < now() - interval '5 minutes'`;114 const jobs = await sql<JobRow[]>`115 UPDATE notification_jobs SET status = 'sending', locked_at = now(), attempts = attempts + 1116 WHERE id IN (SELECT id FROM notification_jobs WHERE status = 'queued' AND run_at <= ${new Date(nowMs())} ORDER BY run_at LIMIT ${limit} FOR UPDATE SKIP LOCKED)117 RETURNING id, user_id, kind, dedup_key, to_email::text, locale, request_id, booking_id, payload, attempts`;118 for (const job of jobs) {119 try {120 const { ctx, ics, replyTo, skip } = await load(job);121 if (skip) { await sql`UPDATE notification_jobs SET status = 'cancelled', last_error = ${skip} WHERE id = ${job.id}`; stats.skipped++; continue; }122 const m = renderMail(ctx);123 const res = await sendMail({ to: job.to_email, subject: m.subject, text: m.text, html: m.html, replyTo, ics, idempotencyKey: `spcal-${job.id}` });124 await sql`UPDATE notification_jobs SET status = 'sent', sent_at = now(), transport = ${res.transport}, provider_message_id = ${res.messageId}, last_error = NULL WHERE id = ${job.id}`;125 stats.sent++;126 } catch (e) {127 if (e instanceof MailNotConfigured) {128 await sql`UPDATE notification_jobs SET status = 'unconfigured', attempts = attempts - 1, last_error = ${e.message} WHERE id = ${job.id}`;129 stats.unconfigured++; continue;130 }131 const msg = (e instanceof Error ? e.message : String(e)).slice(0, 500);132 const delay = BACKOFF_MIN[job.attempts - 1];133 if (delay === undefined) { await sql`UPDATE notification_jobs SET status = 'dead', last_error = ${msg} WHERE id = ${job.id}`; stats.dead++; }134 else { await sql`UPDATE notification_jobs SET status = 'queued', run_at = ${new Date(nowMs() + delay * 60_000)}, last_error = ${msg} WHERE id = ${job.id}`; stats.failed++; }135 }136 }137 return stats;138}139140/** Admin action: put parked or dead jobs back in the queue (after configuring the transport, or fixing an address). */141export async function requeue(userId: string, ids?: string[]): Promise<number> {142 const at = new Date(nowMs());143 const rows = ids?.length144 ? await sql`UPDATE notification_jobs SET status = 'queued', run_at = ${at}, attempts = 0, last_error = NULL WHERE user_id = ${userId} AND id = ANY(${ids}) AND status IN ('unconfigured', 'dead', 'failed') RETURNING id`145 : await sql`UPDATE notification_jobs SET status = 'queued', run_at = ${at}, attempts = 0, last_error = NULL WHERE user_id = ${userId} AND status IN ('unconfigured', 'dead') RETURNING id`;146 return rows.length;147}148