Email: show the real sender; retry webhooks and notifications separately (#31)

The settings and admin pages showed notifications@diffinery.com whenever
MAIL_FROM was empty, while sending used notifications@ the public host. Both
now use one senderAddress() helper, which also matches render-config's
--mail-from default for self-hosted installs.

An event's webhooks and notifications used to run in one Promise.all, so a
failed notification retried the whole event and sent every webhook again with
new delivery ids (and the reverse sent emails twice). Each now runs as its own
stage; a failed stage goes back on the events queue alone with backoff. Once a
hook may have been called, bookkeeping errors no longer fail the stage.


Claude-Session: https://claude.ai/code/session_0179NAjfJDTGSXDDZ1nPq8nW

Co-authored-by: Claude <noreply@anthropic.com>

Canberk Demirci authored 13 days ago· GitHub committed 13 days agosigned1 parent 0f675dacommit 6c101ba5822568504aacd891ded4a1e103e066d3Browse files

7 files changed+137−11

Files (7)

apps/diffinery/src/app.ts+24−4
apps/diffinery/src/lib/stages.ts+43−0
apps/diffinery/src/mail/api.ts+2−2
apps/diffinery/src/mail/send.ts+4−1
apps/diffinery/src/mail/web.tsx+3−3
apps/diffinery/test/unit/email.test.ts+9−1
apps/diffinery/test/unit/stages.test.ts+52−0

modifiedapps/diffinery/src/app.ts+24−4View file

addedapps/diffinery/src/lib/stages.ts+43−0View file

modifiedapps/diffinery/src/mail/api.ts+2−2View file

modifiedapps/diffinery/src/mail/send.ts+4−1View file

modifiedapps/diffinery/src/mail/web.tsx+3−3View file

modifiedapps/diffinery/test/unit/email.test.ts+9−1View file

addedapps/diffinery/test/unit/stages.test.ts+52−0View file

		@@ -9,6 +9,7 @@
9	9	import { SESSION_COOKIE } from "./auth/sessions.js";
10	10	import { newId } from "./lib/util.js";
11	11	import { deliverWebhook, type WebhookDeps, type WebhookRetry } from "./lib/webhooks.js";
	12	+import { runStages, type Stage, type StageRetry } from "./lib/stages.js";
12	13	import { apiNotFound, errorResponse, loadRepo, repoStub } from "./http/common.js";
13	14	import { git } from "./http/git.js";
14	15	import { runner } from "./http/runner.js";
		@@ -145,7 +146,10 @@
145	146	const hooks = (await stub.webhooks()).filter((h) => h.active && (h.events.includes("*") \|\| h.events.includes(GITHUB_EVENT[ev.type])));
146	147	if (!hooks.length) return;
147	148	const body = JSON.stringify(webhookPayload(env, ev));
148		- await Promise.all(hooks.map((h) => deliverWebhook(webhookDeps(env), { type: "webhook_retry", repoId: ev.repoId, hookId: h.id, event: GITHUB_EVENT[ev.type], diffineryEvent: ev.type, body, delivery: newId(), attempt: 0 }, h)));
	149	+ // Each hook retries on its own (deliverWebhook requeues it). Once a hook may have been sent to,
	150	+ // an error must not fail the stage, or the stage retry would send every hook again.
	151	+ const sent = await Promise.allSettled(hooks.map((h) => deliverWebhook(webhookDeps(env), { type: "webhook_retry", repoId: ev.repoId, hookId: h.id, event: GITHUB_EVENT[ev.type], diffineryEvent: ev.type, body, delivery: newId(), attempt: 0 }, h)));
	152	+ for (const r of sent) if (r.status === "rejected") console.error("webhook delivery bookkeeping failed", ev.type, r.reason);
149	153	}
150	154
151	155	const webhookDeps = (env: Env): WebhookDeps => ({
		@@ -167,9 +171,24 @@
167	171	if (ev.type === "pull_request" && ev.action === "merged") {
168	172	await dir.audit({ actorId: ev.actor.id, actorLogin: ev.actor.login, onBehalfOf: ev.actor.onBehalfOf, action: "pull.merge", target: `${ev.owner}/${ev.repo}#${ev.number}`, at: ev.at });
169	173	}
170		- await Promise.all([deliverWebhooks(env, ev), notifyEvent(env, dir, ev)]);
	174	+ await runEventStages(env, dir, ev, ["webhooks", "notify"]);
171	175	}
172	176
	177	+type RepoEvent = Exclude<DiffineryEvent, { type: "audit" }>;
	178	+
	179	+/** Webhooks and notifications run independently; a failed one is retried alone (see lib/stages.ts). */
	180	+function runEventStages(env: Env, dir: Directory, ev: RepoEvent, stages: Stage[], attempt = 0): Promise<void> {
	181	+ return runStages(
	182	+ ev,
	183	+ stages,
	184	+ (stage) => (stage === "webhooks" ? deliverWebhooks(env, ev) : notifyEvent(env, dir, ev)),
	185	+ async (r, delaySeconds) => {
	186	+ await env.EVENTS.send(r as unknown as DiffineryEvent, { delaySeconds });
	187	+ },
	188	+ attempt,
	189	+ );
	190	+}
	191	+
173	192	export async function queue(batch: MessageBatch<DiffineryEvent \| MailJob>, env: Env): Promise<void> {
174	193	// Email has its own queue (diffinery-mail): its jobs can hold sign-in and invite links, so they stay out of the event archive.
175	194	if (batch.queue.startsWith("diffinery-mail")) return mailQueue(batch as MessageBatch<MailJob>, env);
		@@ -180,12 +199,13 @@
180	199	const dir = new Directory(env);
181	200	// Keep an append-only copy of every event in R2 for long-term audit retention.
182	201	const day = new Date().toISOString().slice(0, 10).replaceAll("-", "/");
183		- const fresh = batch.messages.filter((m) => (m.body as { type: string }).type !== "webhook_retry");
	202	+ const fresh = batch.messages.filter((m) => !["webhook_retry", "stage_retry"].includes((m.body as { type: string }).type));
184	203	if (fresh.length) await env.BUCKET.put(`events/${day}/${Date.now()}-${newId()}.jsonl`, fresh.map((m) => JSON.stringify(m.body)).join("\n")).catch((e) => console.error("event archive failed", e));
185	204	for (const msg of batch.messages) {
186	205	try {
187		- const body = msg.body as DiffineryEvent \| WebhookRetry;
	206	+ const body = msg.body as DiffineryEvent \| WebhookRetry \| StageRetry<RepoEvent>;
188	207	if (body.type === "webhook_retry") await deliverWebhook(webhookDeps(env), body);
	208	+ else if (body.type === "stage_retry") await runEventStages(env, dir, body.event, [body.stage], body.attempt);
189	209	else await handleEvent(env, dir, body);
190	210	msg.ack();
191	211	} catch (e) {

		@@ -1,0 +1,43 @@
	1	+/**
	2	+ * An event fans out to independent side effects (webhooks, notifications).
	3	+ * Each runs on its own: one that fails is put back on the events queue alone,
	4	+ * with a delay, so retrying it never repeats the others (webhooks going out
	5	+ * again with new delivery ids, or emails sent twice).
	6	+ */
	7	+export type Stage = "webhooks" \| "notify";
	8	+
	9	+export interface StageRetry<E> {
	10	+ type: "stage_retry";
	11	+ stage: Stage;
	12	+ event: E;
	13	+ /** Attempts made so far. */
	14	+ attempt: number;
	15	+}
	16	+
	17	+// Delays after the first, second, ... failure: about 1.5 hours in all.
	18	+export const STAGE_RETRY_DELAYS = [30, 120, 600, 1800, 3600];
	19	+
	20	+/** Run `stages` for `event`; never throws. Failures are requeued one stage at a time. */
	21	+export async function runStages<E>(
	22	+ event: E,
	23	+ stages: Stage[],
	24	+ run: (stage: Stage) => Promise<void>,
	25	+ requeue: (r: StageRetry<E>, delaySeconds: number) => Promise<void>,
	26	+ attempt = 0,
	27	+): Promise<void> {
	28	+ await Promise.all(
	29	+ stages.map(async (stage) => {
	30	+ try {
	31	+ await run(stage);
	32	+ } catch (e) {
	33	+ const delay = STAGE_RETRY_DELAYS[attempt];
	34	+ if (delay === undefined) {
	35	+ console.error(`event ${stage} failed ${attempt + 1} times; giving up`, e);
	36	+ return;
	37	+ }
	38	+ console.error(`event ${stage} failed; retrying it alone in ${delay}s`, e);
	39	+ await requeue({ type: "stage_retry", stage, event, attempt: attempt + 1 }, delay).catch((q) => console.error(`could not queue the ${stage} retry`, q));
	40	+ }
	41	+ }),
	42	+ );
	43	+}

		@@ -6,7 +6,7 @@
6	6	import { HttpError, now } from "../lib/util.js";
7	7	import { MailStore, mergePrefs, type WatchLevel } from "./prefs.js";
8	8	import { sampleJobs } from "./samples.js";
9		-import { deliver } from "./send.js";
	9	+import { deliver, senderAddress } from "./send.js";
10	10
11	11	/**
12	12	* Email over the REST API: repository watching in GitHub's shape
		@@ -126,7 +126,7 @@
126	126	siteAdmin(c, "read");
127	127	const store = new MailStore(c.env.DB);
128	128	const [day, last] = await Promise.all([store.counts(new Date(Date.now() - 86400_000).toISOString()), store.recent({ limit: 1 })]);
129		- return c.json({ sending: !!c.env.EMAIL, queue: !!c.env.MAIL, from: c.env.MAIL_FROM \|\| null, last_24h: day, last: last[0] ? { ...last[0], dev_copy: undefined } : null });
	129	+ return c.json({ sending: !!c.env.EMAIL, queue: !!c.env.MAIL, from: senderAddress(c.env), last_24h: day, last: last[0] ? { ...last[0], dev_copy: undefined } : null });
130	130	});
131	131
132	132	/**

		@@ -30,6 +30,9 @@
30	30
31	31	export class RetryableMailError extends Error {}
32	32
	33	+/** The address every email is sent from: MAIL_FROM, or notifications@ the public host. */
	34	+export const senderAddress = (env: Pick<Env, "MAIL_FROM" \| "PUBLIC_URL">): string => env.MAIL_FROM \|\| `notifications@${new URL(env.PUBLIC_URL).hostname}`;
	35	+
33	36	const clean = (s: string) => s.replace(/[\r\n]+/g, " ").trim();
34	37	const listIdLabel = (s: string) => s.toLowerCase().replace(/[^a-z0-9-]+/g, "-").replace(/^-+\|-+$/g, "") \|\| "x";
35	38
		@@ -83,7 +86,7 @@
83	86	headers["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click";
84	87	}
85	88	return {
86		- from: { name: email.fromName ? `${clean(email.fromName)} (Diffinery)` : "Diffinery", email: env.MAIL_FROM \|\| `notifications@${host}` },
	89	+ from: { name: email.fromName ? `${clean(email.fromName)} (Diffinery)` : "Diffinery", email: senderAddress(env) },
87	90	to: to.name ? { name: clean(to.name), email: to.email } : to.email,
88	91	subject: clean(email.subject).slice(0, 900),
89	92	html: toHtml(email, ctx),

		@@ -10,7 +10,7 @@
10	10	import { Form, Time, WATCH_OPTIONS, act, flash, post, prm, render, repoUrl, type WebApp } from "../web/ui.js";
11	11	import { MailStore, WATCH_LEVELS, type EmailLogRow, type NotificationPrefs, type WatchLevel } from "./prefs.js";
12	12	import { sampleJobs } from "./samples.js";
13		-import { buildMessage, deliver, enqueueMail } from "./send.js";
	13	+import { buildMessage, deliver, enqueueMail, senderAddress } from "./send.js";
14	14	import type { Category, MailJob } from "./types.js";
15	15	import { readUnsubToken, type UnsubScope, type UnsubToken } from "./unsubscribe.js";
16	16
		@@ -127,7 +127,7 @@
127	127	<>
128	128	<h1>Notifications</h1>
129	129	<p class="muted">
130		- Emails go to <b>{me.email ?? "no address"}</b> from {c.env.MAIL_FROM \|\| "notifications@diffinery.com"}. Security and account emails always send.
	130	+ Emails go to <b>{me.email ?? "no address"}</b> from {senderAddress(c.env)}. Security and account emails always send.
131	131	</p>
132	132	<Form action="/settings/notifications" t={t} class="box">
133	133	<table class="tbl">
		@@ -386,7 +386,7 @@
386	386	<dt>Queue</dt>
387	387	<dd>{c.env.MAIL ? "diffinery-mail" : <span class="bad">not bound: emails are dropped</span>}</dd>
388	388	<dt>From</dt>
389		- <dd>{c.env.MAIL_FROM \|\| "notifications@diffinery.com"}</dd>
	389	+ <dd>{senderAddress(c.env)}</dd>
390	390	<dt>Last 24 hours</dt>
391	391	<dd>
392	392	<span class="ok">{counts.sent} sent</span> · <span class={counts.failed ? "bad" : "faint"}>{counts.failed} failed</span> · <span class="faint">{counts.skipped} skipped</span>

		@@ -3,7 +3,7 @@
3	3	import { mentions } from "../../src/mail/notify.js";
4	4	import { DEFAULT_PREFS, mergePrefs, wantsEmail } from "../../src/mail/prefs.js";
5	5	import { sampleJobs } from "../../src/mail/samples.js";
6		-import { buildMessage } from "../../src/mail/send.js";
	6	+import { buildMessage, senderAddress } from "../../src/mail/send.js";
7	7	import { compose, toHtml, toText } from "../../src/mail/templates.js";
8	8	import { categoryOf, isMandatory, type MailJob } from "../../src/mail/types.js";
9	9	import { readUnsubToken, unsubToken } from "../../src/mail/unsubscribe.js";
		@@ -66,6 +66,14 @@
66	66	});
67	67	});
68	68
	69	+describe("sender address", () => {
	70	+ it("uses MAIL_FROM, or notifications@ the public host", () => {
	71	+ expect(senderAddress(env)).toBe("notifications@diffinery.example");
	72	+ expect(senderAddress({ PUBLIC_URL: "https://git.example.com", MAIL_FROM: "" } as Env)).toBe("notifications@git.example.com");
	73	+ expect(senderAddress({ PUBLIC_URL: "https://git.example.com" } as Env)).toBe("notifications@git.example.com");
	74	+ });
	75	+});
	76	+
69	77	describe("mentions", () => {
70	78	it("finds @logins outside code", () => {
71	79	expect(mentions("Hey @Octocat and @rivera-2, see `@notme` and\n```\n@alsonot\n```\nmail me@example.com")).toEqual(["octocat", "rivera-2"]);

		@@ -1,0 +1,52 @@
	1	+import { describe, expect, it, vi } from "vitest";
	2	+import { STAGE_RETRY_DELAYS, runStages, type Stage, type StageRetry } from "../../src/lib/stages.js";
	3	+
	4	+const quiet = () => vi.spyOn(console, "error").mockImplementation(() => {});
	5	+
	6	+describe("event stages", () => {
	7	+ it("retries only the stage that failed", async () => {
	8	+ quiet();
	9	+ const ran: Stage[] = [];
	10	+ const queued: Array<[StageRetry<string>, number]> = [];
	11	+ const run = async (s: Stage) => {
	12	+ ran.push(s);
	13	+ if (s === "notify") throw new Error("D1 hiccup");
	14	+ };
	15	+ await runStages("ev", ["webhooks", "notify"], run, async (r, d) => void queued.push([r, d]));
	16	+ expect(ran.sort()).toEqual(["notify", "webhooks"]);
	17	+ expect(queued).toEqual([[{ type: "stage_retry", stage: "notify", event: "ev", attempt: 1 }, STAGE_RETRY_DELAYS[0]]]);
	18	+
	19	+ // The retry runs notifications alone: webhooks are not sent again.
	20	+ ran.length = 0;
	21	+ queued.length = 0;
	22	+ await runStages("ev", ["notify"], async (s) => void ran.push(s), async (r, d) => void queued.push([r, d]), 1);
	23	+ expect(ran).toEqual(["notify"]);
	24	+ expect(queued).toEqual([]);
	25	+ });
	26	+
	27	+ it("backs off and eventually gives up without throwing", async () => {
	28	+ quiet();
	29	+ const delays: number[] = [];
	30	+ const fail = async () => {
	31	+ throw new Error("down");
	32	+ };
	33	+ for (let attempt = 0; attempt <= STAGE_RETRY_DELAYS.length; attempt++) await runStages("ev", ["webhooks"], fail, async (_r, d) => void delays.push(d), attempt);
	34	+ expect(delays).toEqual(STAGE_RETRY_DELAYS);
	35	+ });
	36	+
	37	+ it("does not throw when the retry can't be queued", async () => {
	38	+ quiet();
	39	+ await expect(
	40	+ runStages(
	41	+ "ev",
	42	+ ["notify"],
	43	+ async () => {
	44	+ throw new Error("x");
	45	+ },
	46	+ async () => {
	47	+ throw new Error("queue down");
	48	+ },
	49	+ ),
	50	+ ).resolves.toBeUndefined();
	51	+ });
	52	+});

codician/diffinerypublic

Email: show the real sender; retry webhooks and notifications separately (#31)