Add announcement editing and reliable per-recipient mail delivery
This commit is contained in:
t.indorf
@@ -0,0 +1,286 @@
|
||||
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<Announcement, "id" | "kitaId" | "audience" | "revision">,
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user