Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

413 lines · 23.5 KBRaw History
1/**
2 * Who hears about what. GitHub's rules, applied to Diffinery's events:
3 *
4 * - Participating: you opened the thread, commented or reviewed, are assigned
5 * or asked to review, were @mentioned, or subscribed by hand.
6 * - Watching: "All activity" on a repository adds every new issue, pull
7 * request, comment, review and state change there.
8 * - Ignoring a repository or muting a thread stops them (a direct @mention,
9 * assignment or review request in the same event still gets through a mute).
10 * - Your own actions never notify you, unless you turn on "Include your own updates" for email.
11 * - Workflow runs notify whoever triggered them (an agent's owner for agents;
12 * watchers for scheduled runs), for failures unless they ask for every run,
13 * and not while they ignore the repository.
14 * - Security emails (sign-ins, passkeys, tokens, agents, push protection) can't be turned off.
15 *
16 * Every recipient must still be able to read the repository.
17 */
18import type { DiffineryEvent } from "@diffinery/shared";
19import type { Env } from "../env.js";
20import { repoStub } from "../http/common.js";
21import type { Account, Directory, Repo } from "../lib/directory.js";
22import { sha256 } from "../lib/util.js";
23import { MailStore, type NotificationPrefs } from "./prefs.js";
24import { enqueueMail } from "./send.js";
25import type { JobInfo, RunInfo } from "../repo/actions.js";
26import type { MailJob, Reason, ThreadData } from "./types.js";
27
28type ThreadEvent = Extract<DiffineryEvent, { type: "pull_request" | "issues" | "issue_comment" | "pull_request_review" }>;
29
30/** Logins @mentioned in Markdown, ignoring code. */
31export function mentions(md: string | undefined | null): string[] {
32 if (!md) return [];
33 const text = md.replace(/```[\s\S]*?```/g, " ").replace(/~~~[\s\S]*?~~~/g, " ").replace(/`[^`\n]*`/g, " ");
34 const out = new Set<string>();
35 for (const m of text.matchAll(/(?:^|[^\w`/@.-])@([A-Za-z0-9](?:[A-Za-z0-9-]{0,38}))(?![\w@-]|\.\w)/g)) {
36 out.add(m[1]!.toLowerCase());
37 if (out.size >= 50) break;
38 }
39 return [...out];
40}
41
42/** Reasons in priority order: the first that applies is the one the email gives. */
43const PRIORITY: Reason[] = ["review_requested", "assign", "mention", "author", "comment", "manual", "state_change", "subscribed"];
44const DIRECT: Reason[] = ["review_requested", "assign", "mention"];
45
46class Recipients {
47 readonly map = new Map<string, Reason>();
48 add(id: string | undefined | null, reason: Reason): void {
49 if (!id) return;
50 const cur = this.map.get(id);
51 if (!cur || PRIORITY.indexOf(reason) < PRIORITY.indexOf(cur)) this.map.set(id, reason);
52 }
53}
54
55async function accountsByLogin(dir: Directory, logins: string[]): Promise<Account[]> {
56 const out: Account[] = [];
57 for (const l of [...new Set(logins.map((x) => x.toLowerCase()))].slice(0, 60)) {
58 const a = await dir.accountByLogin(l);
59 if (a) out.push(a);
60 }
61 return out;
62}
63
64/** Recipients who may still read the repository, with their accounts. */
65async function readable(dir: Directory, repo: Repo, ids: string[]): Promise<Account[]> {
66 const out: Account[] = [];
67 for (const id of ids.slice(0, 200)) {
68 const a = await dir.account(id);
69 if (!a || a.kind === "org" || a.status !== "active") continue;
70 if (await dir.roleFor(a, repo)) out.push(a);
71 }
72 return out;
73}
74
75async function threadEvent(env: Env, dir: Directory, ev: ThreadEvent): Promise<void> {
76 if (ev.type === "pull_request" && (ev.action === "enqueued" || ev.action === "dequeued" || ev.action === "edited" || ev.action === "auto_merge_enabled" || ev.action === "auto_merge_disabled")) return;
77 if (ev.type === "issues" && ev.action === "edited") return;
78 const repo = await dir.repoById(ev.repoId);
79 if (!repo) return;
80 const stub = repoStub(env, ev.repoId);
81 const { issue, comments, reviews } = await stub.issue(ev.number);
82 const isPull = issue.kind === "pull";
83 const threadUrl = `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/${isPull ? "pull" : "issues"}/${ev.number}`;
84 const d: ThreadData = { repo: `${ev.owner}/${ev.repo}`, repoId: ev.repoId, number: ev.number, kind: issue.kind, title: issue.title, event: "opened", actor: ev.actor.login, url: threadUrl, base: issue.pull?.base, head: issue.pull?.head, intent: issue.pull?.intent ?? null, labels: issue.labels };
85 const r = new Recipients();
86 let text: string | undefined;
87 let participantsOnly = false;
88
89 if ((ev.type === "issues" || ev.type === "pull_request") && ev.action === "opened") {
90 d.event = "opened";
91 d.body = issue.body;
92 d.assignees = issue.assignees;
93 d.reviewers = issue.pull?.requestedReviewers;
94 text = issue.body;
95 for (const a of await accountsByLogin(dir, issue.assignees)) r.add(a.id, "assign");
96 for (const a of await accountsByLogin(dir, issue.pull?.requestedReviewers ?? [])) r.add(a.id, "review_requested");
97 if (isPull) d.commits = (await stub.pullCommits(ev.number).catch(() => [])).slice(-10).map((c) => ({ sha: c.sha, message: c.message }));
98 } else if (ev.type === "issue_comment") {
99 const cm = comments.find((c) => c.id === ev.commentId);
100 if (!cm) return;
101 d.event = "comment";
102 d.body = cm.body;
103 d.url = `${threadUrl}#c-${cm.id}`;
104 text = cm.body;
105 } else if (ev.type === "pull_request_review") {
106 const rv = [...reviews].reverse().find((x) => x.author.id === ev.actor.id);
107 d.event = "review";
108 d.state = ev.state;
109 const lineComments = rv ? comments.filter((c) => c.reviewId === rv.id) : [];
110 d.body = [rv?.body ?? "", ...lineComments.slice(0, 20).map((c) => `**${c.path}${c.line ? `:${c.line}` : ""}**\n\n${c.body}`)].filter((x) => x.trim()).join("\n\n---\n\n");
111 text = d.body;
112 } else if (ev.action === "closed" || ev.action === "merged" || ev.action === "reopened") {
113 if (ev.type === "pull_request" && ev.action === "closed" && issue.pull?.merged) return; // the "merged" event covers it
114 d.event = ev.action;
115 d.stateReason = issue.stateReason;
116 } else if (ev.type === "pull_request" && ev.action === "synchronize") {
117 d.event = "pushed";
118 d.commits = (await stub.pullCommits(ev.number).catch(() => [])).slice(-10).map((c) => ({ sha: c.sha, message: c.message }));
119 participantsOnly = true;
120 } else if (ev.action === "assigned") {
121 d.event = "assigned";
122 d.assignees = ev.assignees ?? [];
123 for (const a of await accountsByLogin(dir, d.assignees)) r.add(a.id, "assign");
124 participantsOnly = true;
125 } else if (ev.type === "pull_request" && ev.action === "review_requested") {
126 d.event = "review_requested";
127 d.reviewers = ev.reviewers ?? [];
128 for (const a of await accountsByLogin(dir, d.reviewers)) r.add(a.id, "review_requested");
129 participantsOnly = true;
130 } else return;
131
132 for (const a of await accountsByLogin(dir, mentions(text))) r.add(a.id, "mention");
133 r.add(issue.author.id, "author");
134 for (const c of comments) r.add(c.author.id, "comment");
135 for (const rv of reviews) r.add(rv.author.id, "comment");
136 // Assignees and requested reviewers take part in the thread from then on.
137 if (d.event !== "assigned") for (const a of await accountsByLogin(dir, issue.assignees)) r.add(a.id, "assign");
138 if (d.event !== "review_requested") for (const a of await accountsByLogin(dir, issue.pull?.requestedReviewers ?? [])) r.add(a.id, "review_requested");
139
140 const store = new MailStore(env.DB);
141 const [subs, watch] = await Promise.all([store.threadSubs(ev.repoId, ev.number), store.watchers(ev.repoId)]);
142 for (const [id, on] of subs) if (on) r.add(id, "manual");
143 // Assignments, review requests and pushes go to the people involved, not to every watcher.
144 if (!participantsOnly) for (const id of watch.all) r.add(id, "subscribed");
145 const direct = new Set([...(ev.type === "pull_request" && ev.action === "review_requested" ? d.reviewers ?? [] : []), ...(ev.action === "assigned" ? d.assignees ?? [] : []), ...mentions(text)].map((x) => x.toLowerCase()));
146
147 const candidates = [...r.map.keys()].filter((id) => !watch.ignore.has(id));
148 const accounts = await readable(dir, repo, candidates);
149 const prefs = await store.prefsFor(accounts.map((a) => a.id));
150 const web = new Map<Reason, string[]>();
151 const jobs: MailJob[] = [];
152 for (const a of accounts) {
153 let reason = r.map.get(a.id)!;
154 if (subs.get(a.id) === false && !(DIRECT.includes(reason) && direct.has(a.login.toLowerCase()))) continue;
155 const p = prefs.get(a.id)!;
156 const own = a.id === ev.actor.id;
157 if (own) reason = "your_activity";
158 const group: keyof NotificationPrefs = reason === "subscribed" ? "watching" : "participating";
159 if (!own && p[group].web) web.set(reason, [...(web.get(reason) ?? []), a.id]);
160 if (a.kind === "user" && (!own || p.ownActivity)) jobs.push({ kind: "thread", data: d, to: { accountId: a.id }, reason, at: ev.at });
161 }
162 const verb: Record<ThreadData["event"], string> = { opened: "opened", comment: "commented on", review: "reviewed", closed: "closed", merged: "merged", reopened: "reopened", assigned: "assigned", review_requested: "requested review on", pushed: "pushed to" };
163 const title = `${ev.actor.login} ${d.event === "review" ? `reviewed (${(ev as { state?: string }).state?.toLowerCase().replace(/_/g, " ")})` : verb[d.event]} ${d.repo}#${ev.number}: ${issue.title}`;
164 const webIds = [...web.values()].flat();
165 await Promise.all([webIds.length ? dir.notify(webIds.slice(0, 200), { repoId: ev.repoId, kind: ev.type, title, url: d.url }) : null, enqueueMail(env, jobs)]);
166}
167
168async function workflowRun(env: Env, dir: Directory, ev: Extract<DiffineryEvent, { type: "workflow_run" }>): Promise<void> {
169 if (ev.action !== "completed") return;
170 const failed = ev.conclusion === "failure" || ev.conclusion === "timed_out";
171 if (!failed && ev.conclusion !== "success") return;
172 const repo = await dir.repoById(ev.repoId);
173 if (!repo) return;
174 const store = new MailStore(env.DB);
175 const watch = await store.watchers(ev.repoId);
176 let ids: string[] = [];
177 if (ev.actor.kind === "user") ids = [ev.actor.id];
178 else if (ev.actor.kind === "agent" && ev.actor.onBehalfOf) {
179 const owner = await dir.accountByLogin(ev.actor.onBehalfOf);
180 if (owner?.kind === "user") ids = [owner.id];
181 } else ids = watch.all;
182 // Ignoring a repository silences its runs too, even the ones you started.
183 const accounts = await readable(dir, repo, ids.filter((id) => !watch.ignore.has(id)));
184 if (!accounts.length) return;
185 const prefs = await store.prefsFor(accounts.map((a) => a.id));
186 const url = `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/actions/runs/${ev.runId}`;
187 const webIds = failed ? accounts.filter((a) => prefs.get(a.id)!.actions.web).map((a) => a.id) : [];
188 const mailTo = accounts.filter((a) => a.kind === "user" && (failed || !prefs.get(a.id)!.actions.failuresOnly));
189 let jobs: MailJob[] = [];
190 if (mailTo.length) {
191 // The RPC return type is too deep for TypeScript to instantiate; it is exactly getRun()'s.
192 const run = (await (repoStub(env, ev.repoId).run as unknown as (id: string) => Promise<unknown>)(ev.runId)) as RunInfo & { jobs: JobInfo[] };
193 const data = {
194 repo: `${ev.owner}/${ev.repo}`,
195 repoId: ev.repoId,
196 runId: ev.runId,
197 runNumber: run.number,
198 workflow: run.workflowName,
199 title: run.title,
200 conclusion: ev.conclusion ?? "",
201 event: run.event,
202 branch: run.headBranch || run.ref.replace(/^refs\/(heads|tags)\//, ""),
203 sha: run.headSha || run.sha,
204 actor: run.actor.login,
205 url,
206 jobs: run.jobs.map((j) => ({ name: j.name, conclusion: j.conclusion, url: `${url}/job/${j.id}`, failure: j.failure?.message ?? null })),
207 };
208 jobs = mailTo.map((a) => ({ kind: "run" as const, data, to: { accountId: a.id }, reason: "ci_activity" as const, at: ev.at }));
209 }
210 await Promise.all([webIds.length ? dir.notify(webIds, { repoId: ev.repoId, kind: "run_failed", title: `Workflow run failed in ${ev.owner}/${ev.repo}`, url }) : null, enqueueMail(env, jobs)]);
211}
212
213/** Web notifications and emails for one event from the events queue. */
214export async function notifyEvent(env: Env, dir: Directory, ev: DiffineryEvent): Promise<void> {
215 if (ev.type === "workflow_run") return workflowRun(env, dir, ev);
216 if (ev.type === "deployment_review" && ev.action === "requested") return deploymentReview(env, dir, ev);
217 if (ev.type === "pull_request" || ev.type === "issues" || ev.type === "issue_comment" || ev.type === "pull_request_review") return threadEvent(env, dir, ev);
218}
219
220/**
221 * A run's jobs wait on an environment's required reviewers: each reviewer who can still read the
222 * repository gets a notification and an email (unless they turned Actions email off).
223 */
224async function deploymentReview(env: Env, dir: Directory, ev: Extract<DiffineryEvent, { payload: unknown }>): Promise<void> {
225 const p = ev.payload as { environment?: string; workflow_run?: { id?: string; run_number?: number; name?: string; display_title?: string; head_sha?: string; head_branch?: string }; reviewers?: Array<{ reviewer?: { id?: string } }> };
226 const repo = await dir.repoById(ev.repoId);
227 const run = p.workflow_run;
228 if (!repo || !run?.id || !p.environment) return;
229 const ids = (p.reviewers ?? []).map((r) => r.reviewer?.id).filter((x): x is string => typeof x === "string" && x !== ev.actor.id);
230 const accounts = (await readable(dir, repo, ids)).filter((a) => a.kind === "user");
231 if (!accounts.length) return;
232 const url = `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/actions/runs/${run.id}`;
233 const prefs = await new MailStore(env.DB).prefsFor(accounts.map((a) => a.id));
234 const data = { repo: `${ev.owner}/${ev.repo}`, environment: p.environment, workflow: run.name ?? "", runNumber: run.run_number ?? 0, title: run.display_title ?? "", branch: run.head_branch ?? "", sha: run.head_sha ?? "", requestor: ev.actor.login, url };
235 await Promise.all([
236 dir.notify(
237 accounts.filter((a) => prefs.get(a.id)?.actions.web !== false).map((a) => a.id),
238 { repoId: ev.repoId, kind: "deployment_review", title: `Deployment to ${p.environment} in ${ev.owner}/${ev.repo} is waiting for your review`, url },
239 ),
240 enqueueMail(env, accounts.map((a) => ({ kind: "deployment_review_requested" as const, data, to: { accountId: a.id }, reason: "review_requested" as const, at: ev.at }))),
241 ]);
242}
243
244// ---------------------------------------------------------------- account and security emails
245
246/** People who administer a repository: its owner (or the org's owners and admins) and admin collaborators. */
247export async function repoAdmins(dir: Directory, repo: Repo): Promise<string[]> {
248 const ids = new Set<string>();
249 if (repo.owner_kind === "org") for (const m of await dir.orgMembers(repo.owner_id)) if (m.role === "owner" || m.role === "admin") ids.add(m.id);
250 if (repo.owner_kind === "user") ids.add(repo.owner_id);
251 for (const c of await dir.collaborators(repo.id)) if (c.role === "admin" && c.account.kind === "user") ids.add(c.account.id);
252 return [...ids];
253}
254
255/** The people an account's security emails go to: the person, or an agent's owner (or its org's owners). */
256export async function humansFor(dir: Directory, accountId: string | undefined | null): Promise<string[]> {
257 if (!accountId) return [];
258 const a = await dir.account(accountId);
259 if (!a) return [];
260 if (a.kind === "user") return [a.id];
261 if (a.kind === "agent" && a.owner_id) {
262 const o = await dir.account(a.owner_id);
263 if (o?.kind === "user") return [o.id];
264 if (o?.kind === "org") return (await dir.orgMembers(o.id)).filter((m) => m.role === "owner" || m.role === "admin").map((m) => m.id);
265 }
266 return [];
267}
268
269export interface AuditEntry {
270 actorId?: string;
271 actorLogin?: string;
272 action: string;
273 target?: string;
274 detail?: unknown;
275 at?: string;
276}
277
278/** A sign-in device: the passkey (when there is one) plus the browser and OS. */
279export async function deviceFingerprint(device: string, passkeyId?: string | null): Promise<string> {
280 return (await sha256(`${passkeyId ?? "-"}|${device}`)).slice(0, 32);
281}
282
283/** Security emails for audited actions. Called for every audit entry; ignores the rest. */
284export async function mailForAudit(env: Env, dir: Directory, e: AuditEntry): Promise<void> {
285 const d = (e.detail && typeof e.detail === "object" ? e.detail : {}) as Record<string, unknown>;
286 const s = (k: string) => (typeof d[k] === "string" ? (d[k] as string) : undefined);
287 const at = e.at ?? new Date().toISOString();
288 const to = (ids: string[], body: Omit<MailJob, "to" | "at" | "reason">, reason: Reason = "security_alert") => ids.map((id) => ({ ...body, to: { accountId: id }, reason, at }) as MailJob);
289 const store = new MailStore(env.DB);
290 let jobs: MailJob[] = [];
291 switch (e.action) {
292 case "session.create": {
293 if (!e.actorId) return;
294 const device = s("device") ?? "an unknown browser";
295 const method = (s("method") ?? "passkey") as "passkey" | "access" | "link";
296 const fresh = await store.seeDevice(e.actorId, await deviceFingerprint(device, s("passkeyId")), device);
297 if (!fresh && method === "passkey") return;
298 jobs = to([e.actorId], { kind: "signin_new", data: { method, device, ip: s("ip"), location: s("location"), passkey: s("passkey") } });
299 break;
300 }
301 case "passkey.add":
302 if (!e.actorId) return;
303 if (s("device")) await store.seeDevice(e.actorId, await deviceFingerprint(s("device")!, s("passkeyId")), s("device")!);
304 jobs = to([e.actorId], { kind: "passkey_added", data: { name: s("name") ?? "passkey", via: s("via") === "signin-link" ? "signin-link" : "settings" } });
305 break;
306 case "passkey.remove":
307 if (!e.actorId) return;
308 jobs = to([e.actorId], { kind: "passkey_removed", data: { name: s("name") ?? "passkey" } });
309 break;
310 case "agent.create":
311 if (!e.actorId || !e.target) return;
312 jobs = to([e.actorId], { kind: "agent_created", data: { agent: e.target } });
313 break;
314 case "push.secret_bypassed":
315 case "api.secret_bypassed":
316 case "push.blocked_secret":
317 case "api.blocked_secret": {
318 const [owner, name] = (e.target ?? "").split(":")[0]!.split("/");
319 if (!owner || !name) return;
320 const repo = await dir.repo(owner, name);
321 if (!repo) return;
322 const kinds = Array.isArray(d.kinds) ? [...new Set(d.kinds as string[])] : [];
323 const refs = Array.isArray(d.refs) ? (d.refs as string[]) : undefined;
324 const pusher = await humansFor(dir, e.actorId);
325 // A commit made without git (web editor, API, MCP) names the files instead of refs.
326 const paths = Array.isArray(d.paths) ? (d.paths as string[]) : [];
327 if (e.action === "push.blocked_secret") jobs = to(pusher, { kind: "push_blocked", data: { repo: `${owner}/${name}`, kinds, refs } });
328 else if (e.action === "api.blocked_secret") jobs = to(pusher, { kind: "push_blocked", data: { repo: `${owner}/${name}`, kinds, paths } });
329 else jobs = to([...new Set([...pusher, ...(await repoAdmins(dir, repo))])], { kind: "push_bypassed", data: { repo: `${owner}/${name}`, pusher: e.actorLogin ?? "someone", kinds, reason: s("reason") ?? "", refs } });
330 break;
331 }
332 case "invite.accept":
333 // A new account made from an invite (passkey sign-up) gets a welcome email.
334 if (e.actorId && s("method") === "passkey") {
335 if (s("device")) await store.seeDevice(e.actorId, await deviceFingerprint(s("device")!, s("passkeyId")), s("device")!);
336 jobs = to([e.actorId], { kind: "welcome", data: { login: e.actorLogin ?? "" } }, "account");
337 }
338 break;
339 case "repo.delete": {
340 const [owner, name] = (e.target ?? "").split("/");
341 const ownerAcc = owner ? await dir.accountByLogin(owner) : null;
342 if (!ownerAcc || !name) return;
343 // The person who deleted it already knows.
344 const ids = (ownerAcc.kind === "org" ? (await dir.orgMembers(ownerAcc.id)).filter((m) => m.role === "owner" || m.role === "admin").map((m) => m.id) : [ownerAcc.id]).filter((id) => id !== e.actorId);
345 jobs = to(ids, { kind: "repo_deleted", data: { repo: `${owner}/${name}`, actor: e.actorLogin ?? "someone" } }, "admin");
346 break;
347 }
348 case "repo.transfer": {
349 // The admins of both owners: the old one's lose the repository, the new one's gain it.
350 const from = s("from");
351 const dest = s("to");
352 if (!from || !dest) return;
353 const admins = async (login: string) => {
354 const acc = await dir.accountByLogin(login);
355 if (!acc) return [];
356 return acc.kind === "org" ? (await dir.orgMembers(acc.id)).filter((m) => m.role === "owner" || m.role === "admin").map((m) => m.id) : acc.kind === "user" ? [acc.id] : [];
357 };
358 const ids = [...new Set([...(await admins(from.split("/")[0]!)), ...(await admins(dest.split("/")[0]!))])].filter((id) => id !== e.actorId);
359 jobs = to(ids, { kind: "repo_transferred", data: { from, to: dest, actor: e.actorLogin ?? "someone" } }, "admin");
360 break;
361 }
362 default:
363 return;
364 }
365 await enqueueMail(env, jobs);
366}
367
368/** A token was created for a person or an agent (from the web, the API or the CLI). */
369export async function tokenCreated(env: Env, dir: Directory, account: Account, t: { name: string; scopes: string[] }): Promise<void> {
370 const at = new Date().toISOString();
371 const ids = await humansFor(dir, account.id);
372 await enqueueMail(env, ids.map((id) => ({ kind: "token_created" as const, data: { name: t.name, scopes: t.scopes, agent: account.kind === "agent" ? account.login : undefined }, to: { accountId: id }, reason: "security_alert" as const, at })));
373}
374
375/** A repository was created: tell its creator (unless a script did it with a token) and the org's other owners and admins; the creator watches it. */
376export async function repoCreated(env: Env, dir: Directory, repo: Repo, creator: Account, opts: { viaToken: boolean }): Promise<void> {
377 const store = new MailStore(env.DB);
378 const human = creator.kind === "user";
379 if (human && (await store.prefs(creator.id)).autoWatch) await store.setWatch(creator.id, repo.id, "all");
380 const data = { repo: `${repo.owner_login}/${repo.name}`, creator: creator.login, visibility: repo.visibility, defaultBranch: repo.default_branch, description: repo.description };
381 const at = new Date().toISOString();
382 const jobs: MailJob[] = [];
383 if (human && !opts.viaToken) jobs.push({ kind: "repo_created", data, to: { accountId: creator.id }, reason: "your_activity", at });
384 if (repo.owner_kind === "org") {
385 for (const m of await dir.orgMembers(repo.owner_id)) {
386 if ((m.role === "owner" || m.role === "admin") && m.id !== creator.id && m.id !== creator.owner_id) jobs.push({ kind: "repo_created", data, to: { accountId: m.id }, reason: "admin", at });
387 }
388 }
389 await enqueueMail(env, jobs);
390}
391
392/** Daily-ish jobs from the cron trigger: token expiry reminders and log pruning. */
393export async function mailCron(env: Env, dir: Directory, at: Date): Promise<void> {
394 const store = new MailStore(env.DB);
395 const now = at.toISOString();
396 const soon = new Date(at.getTime() + 7 * 86400_000).toISOString();
397 const { results } = await env.DB.prepare(
398 "SELECT id, account_id, name, prefix, expires_at FROM tokens WHERE revoked_at IS NULL AND expiry_notified_at IS NULL AND expires_at > ? AND expires_at <= ? AND julianday(expires_at) - julianday(created_at) > 8 LIMIT 50",
399 )
400 .bind(now, soon)
401 .all<{ id: string; account_id: string; name: string; prefix: string; expires_at: string }>();
402 const jobs: MailJob[] = [];
403 for (const t of results) {
404 const a = await dir.account(t.account_id);
405 if (a && a.status === "active") {
406 const ids = await humansFor(dir, a.id);
407 for (const id of ids) jobs.push({ kind: "token_expiring", data: { name: t.name, prefix: t.prefix, expiresAt: t.expires_at, agent: a.kind === "agent" ? a.login : undefined }, to: { accountId: id }, reason: "security_alert", at: now });
408 }
409 await env.DB.prepare("UPDATE tokens SET expiry_notified_at = ? WHERE id = ?").bind(now, t.id).run();
410 }
411 await enqueueMail(env, jobs);
412 if (at.getUTCHours() === 3 && at.getUTCMinutes() < 5) await store.prune();
413}