import "server-only"; import { stat } from "node:fs/promises"; import { attachmentPath } from "@/lib/attachments"; import { randomUUID } from "node:crypto"; import { createElement } from "react"; import type { Announcement, Prisma } from "@prisma/client"; import { NewsEmail } from "@/emails/NewsEmail"; import { announcementRecipientWhere } from "@/lib/announcement-access"; import { prisma } from "@/lib/prisma"; import { sendAppEmail } from "@/lib/mail"; import { UserFacingError } from "@/lib/action-error"; export async function lockAnnouncement( tx: Prisma.TransactionClient, kitaId: string, id: string, ) { await tx.$queryRaw`SELECT id FROM announcements WHERE id = ${id} AND "kitaId" = ${kitaId} FOR UPDATE`; } export async function queueAnnouncementEmails( tx: Prisma.TransactionClient, announcement: Pick, now = new Date(), ) { const users = await tx.user.findMany({ where: { kitaId: announcement.kitaId, disabledAt: null, emailVerifiedAt: { not: null }, privacyPolicyAcceptedAt: { not: null }, AND: [ announcementRecipientWhere(announcement.audience), { OR: [{ lockedUntil: null }, { lockedUntil: { lte: now } }] }, ], }, select: { id: true }, }); if (!users.length) return 0; return ( await tx.announcementEmail.createMany({ data: users.map((user) => ({ id: randomUUID(), kitaId: announcement.kitaId, announcementId: announcement.id, revision: announcement.revision, userId: user.id, nextAttemptAt: now, })), skipDuplicates: true, }) ).count; } type Scope = { kitaId?: string; announcementId?: string; revision?: number }; export async function deliverDueAnnouncementEmails( scope: Scope = {}, send = sendAppEmail, now = new Date(), ) { // Recover a crash between completed file writes and the final readiness update. // Partial or missing files never release a message for sending. let preparationCursor: string | undefined; for (;;) { const unfinished = await prisma.announcement.findMany({ where: { kitaId: scope.kitaId, id: scope.announcementId, revision: scope.revision, emailReady: false, createdAt: { lte: new Date(now.getTime() - 5 * 60_000) }, ...(preparationCursor ? { AND: { id: { gt: preparationCursor } } } : {}), }, select: { id: true, attachments: { select: { id: true, sizeBytes: true } }, }, orderBy: { id: "asc" }, take: 12, }); for (const row of unfinished) { const complete = await Promise.all( row.attachments.map(async (file) => { try { return ( file.sizeBytes !== null && (await stat(attachmentPath(file.id))).size === file.sizeBytes ); } catch { return false; } }), ); if (complete.every(Boolean)) await prisma.announcement.updateMany({ where: { id: row.id, kitaId: scope.kitaId, emailReady: false }, data: { emailReady: true }, }); } if (unfinished.length < 12) break; preparationCursor = unfinished[unfinished.length - 1].id; } const jobs = await prisma.announcementEmail.findMany({ where: { ...scope, sentAt: null, skippedAt: null, nextAttemptAt: { lte: now }, announcement: { emailReady: true }, OR: [{ leaseUntil: null }, { leaseUntil: { lte: now } }], }, select: { id: true, kitaId: true, announcementId: true }, orderBy: [{ nextAttemptAt: "asc" }, { createdAt: "asc" }, { id: "asc" }], take: 12, }); const result = { sent: 0, failed: 0, skipped: 0 }; const deliver = async (candidate: (typeof jobs)[number]) => { const claimToken = randomUUID(); // Edits and claims share the publication lock. An active send blocks editing // until it finishes, so a retried version always has the same message body. // ReadCommitted plus the row lock and atomic claim is sufficient here; // recipient updates must not cause serialization conflicts for other jobs. const claimed = await prisma.$transaction( async (tx) => { await lockAnnouncement(tx, candidate.kitaId, candidate.announcementId); return tx.announcementEmail.updateMany({ where: { id: candidate.id, sentAt: null, skippedAt: null, nextAttemptAt: { lte: now }, announcement: { emailReady: true }, OR: [{ leaseUntil: null }, { leaseUntil: { lte: now } }], }, data: { claimToken, leaseUntil: new Date(now.getTime() + 5 * 60_000), attempts: { increment: 1 }, }, }); }, { isolationLevel: "ReadCommitted" }, ); if (!claimed.count) return; const finish = (data: Prisma.AnnouncementEmailUpdateManyMutationInput) => prisma.announcementEmail.updateMany({ where: { id: candidate.id, claimToken }, data: { ...data, claimToken: null, leaseUntil: null }, }); try { const job = await prisma.announcementEmail.findUniqueOrThrow({ where: { id: candidate.id }, include: { announcement: true, user: true }, }); const eligible = await prisma.user.count({ where: { id: job.userId, kitaId: job.kitaId, disabledAt: null, emailVerifiedAt: { not: null }, privacyPolicyAcceptedAt: { not: null }, AND: [ announcementRecipientWhere(job.announcement.audience), { OR: [{ lockedUntil: null }, { lockedUntil: { lte: now } }] }, ], }, }); if ( !eligible || job.announcement.revision !== job.revision || job.announcement.kitaId !== job.kitaId ) { await finish({ skippedAt: now, failed: false, lastError: "Empfänger oder Beitragsversion nicht mehr gültig.", }); result.skipped++; return; } const base = process.env.NEXT_PUBLIC_SITE_URL || process.env.AUTH_URL || process.env.NEXTAUTH_URL; if (!base) throw new Error("App-Adresse fehlt."); const site = new URL(base); if ( !["http:", "https:"].includes(site.protocol) || site.username || site.password ) throw new Error("App-Adresse ungültig."); const response = await send({ to: job.user.email, idempotencyKey: `announcement-email/${job.id}`, subject: `${job.revision > 1 ? "Aktualisierte" : "Neue"} Kita-Ankündigung: ${job.announcement.title}`, react: createElement(NewsEmail, { title: job.announcement.title, content: job.announcement.content, updated: job.revision > 1, dashboardUrl: new URL( "/dashboard/mitteilungen", site.origin, ).toString(), }), }); if (!response.success) throw new Error(response.error); await finish({ sentAt: now, failed: false, lastError: null }); result.sent++; } catch { const job = await prisma.announcementEmail.findUnique({ where: { id: candidate.id }, select: { attempts: true }, }); const delay = Math.min( 6 * 60 * 60_000, 60_000 * 2 ** Math.min(job?.attempts ?? 1, 9), ); await finish({ failed: true, lastError: "E-Mail-Versand fehlgeschlagen. Ein neuer Versuch folgt automatisch.", nextAttemptAt: new Date(now.getTime() + delay), }); result.failed++; } }; for (let i = 0; i < jobs.length; i += 4) await Promise.all(jobs.slice(i, i + 4).map(deliver)); return result; } export type AnnouncementActor = { id: string; kitaId: string; sessionVersion: number; }; export async function assertAnnouncementAdmin( tx: Prisma.TransactionClient, actor: AnnouncementActor, ) { const active = await tx.user.findFirst({ where: { id: actor.id, kitaId: actor.kitaId, sessionVersion: actor.sessionVersion, role: "ADMIN", disabledAt: null, emailVerifiedAt: { not: null }, privacyPolicyAcceptedAt: { not: null }, OR: [{ lockedUntil: null }, { lockedUntil: { lte: new Date() } }], }, select: { id: true }, }); if (!active) throw new UserFacingError( "Du darfst Mitteilungen nicht mehr verwalten. Bitte melde dich erneut an.", ); } export async function assertAnnouncementEditable( tx: Prisma.TransactionClient, kitaId: string, id: string, revision: number, ) { await lockAnnouncement(tx, kitaId, id); const announcement = await tx.announcement.findFirst({ where: { id, kitaId, revision, emailReady: true }, }); if (!announcement) throw new UserFacingError( "Der Beitrag wurde inzwischen geändert oder ist nicht verfügbar. Bitte lade die Seite neu.", ); if ( await tx.announcementEmail.count({ where: { kitaId, announcementId: id, leaseUntil: { gt: new Date() } }, }) ) throw new UserFacingError( "Für diesen Beitrag läuft gerade der E-Mail-Versand. Bitte versuche es gleich erneut.", ); return announcement; }