| 1 | import { Hono } from "hono"; |
| 2 | import { getCookie } from "hono/cookie"; |
| 3 | import type { DiffineryEvent } from "@diffinery/shared"; |
| 4 | import { cronMatches } from "@diffinery/workflow-yaml"; |
| 5 | import type { Env } from "./env.js"; |
| 6 | import { credentialFrom, resolvePrincipal, type AppVars } from "./auth/principal.js"; |
| 7 | import { Directory } from "./lib/directory.js"; |
| 8 | import { BOOKMARK_COOKIE, BOOKMARK_HEADER, d1Session, sessionConstraint } from "./lib/d1.js"; |
| 9 | import { SESSION_COOKIE } from "./auth/sessions.js"; |
| 10 | import { newId } from "./lib/util.js"; |
| 11 | import { deliverWebhook, type WebhookDeps, type WebhookRetry } from "./lib/webhooks.js"; |
| 12 | import { runStages, type Stage, type StageRetry } from "./lib/stages.js"; |
| 13 | import { apiNotFound, errorResponse, loadRepo, repoStub } from "./http/common.js"; |
| 14 | import { git } from "./http/git.js"; |
| 15 | import { runner } from "./http/runner.js"; |
| 16 | import { api, wellKnown } from "./http/api.js"; |
| 17 | import { avatars } from "./http/avatars.js"; |
| 18 | import { mailApi } from "./mail/api.js"; |
| 19 | import { files } from "./http/files.js"; |
| 20 | import { nodeId, numericId } from "./http/gh.js"; |
| 21 | import { mcp } from "./mcp/server.js"; |
| 22 | import { registry } from "./registry/http.js"; |
| 23 | import { packagesApi } from "./registry/api.js"; |
| 24 | import { RegistryStore } from "./registry/store.js"; |
| 25 | import { mailCron, notifyEvent } from "./mail/notify.js"; |
| 26 | import { mailQueue } from "./mail/send.js"; |
| 27 | import type { MailJob } from "./mail/types.js"; |
| 28 | |
| 29 | export const app = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 30 | |
| 31 | app.onError((e) => errorResponse(e)); |
| 32 | app.notFound((c) => (c.req.path.startsWith("/api/") ? apiNotFound() : c.text("404 Not Found", 404))); |
| 33 | |
| 34 | // Security headers on every response, then identity: tokens, Access JWT, or anonymous. |
| 35 | app.use("*", async (c, next) => { |
| 36 | // Browser sessions ride on cookies, so refuse state-changing requests that come from another site. |
| 37 | // Token-authenticated calls (git, CLI, MCP, runners) carry no ambient credentials and are exempt. |
| 38 | if (!["GET", "HEAD", "OPTIONS"].includes(c.req.method) && !credentialFrom(c.req.raw)) { |
| 39 | const site = c.req.header("sec-fetch-site"); |
| 40 | const origin = c.req.header("origin"); |
| 41 | const ours = new URL(c.env.PUBLIC_URL).origin; |
| 42 | if ((site && site !== "same-origin" && site !== "none") || (!site && origin && origin !== ours)) { |
| 43 | return c.json({ message: "cross-site request refused" }, 403); |
| 44 | } |
| 45 | } |
| 46 | // One D1 session per request: replica reads that still see this client's earlier writes (see lib/d1.ts). |
| 47 | const token = !!credentialFrom(c.req.raw); |
| 48 | const cookie = getCookie(c, BOOKMARK_COOKIE); |
| 49 | const d1 = d1Session(c.env, sessionConstraint({ header: c.req.header(BOOKMARK_HEADER), cookie, token })); |
| 50 | c.env = d1.env; |
| 51 | const dir = new Directory(c.env); |
| 52 | c.set("dir", dir); |
| 53 | // Avatar images are public and on every page many times over: skip the session lookup for them. |
| 54 | const avatar = (c.req.method === "GET" || c.req.method === "HEAD") && c.req.path.startsWith("/api/avatars/"); |
| 55 | c.set("principal", avatar ? { type: "anonymous" } : await resolvePrincipal(c.req.raw, c.env, dir)); |
| 56 | await next(); |
| 57 | if (c.res.status === 101) return; |
| 58 | const h = c.res.headers; |
| 59 | const bookmark = d1.bookmark(); |
| 60 | if (bookmark) { |
| 61 | h.set(BOOKMARK_HEADER, bookmark); |
| 62 | // Browsers: remember it once they have written something or are signed in. |
| 63 | const browser = !token && (!["GET", "HEAD"].includes(c.req.method) || cookie !== undefined || getCookie(c, SESSION_COOKIE) !== undefined); |
| 64 | if (browser && !avatar && bookmark !== cookie) h.append("set-cookie", `${BOOKMARK_COOKIE}=${bookmark}; Max-Age=86400; Path=/; HttpOnly; Secure; SameSite=Lax`); |
| 65 | } |
| 66 | h.set("strict-transport-security", "max-age=63072000; includeSubDomains; preload"); |
| 67 | h.set("x-content-type-options", "nosniff"); |
| 68 | h.set("referrer-policy", "strict-origin-when-cross-origin"); |
| 69 | if (!h.has("x-frame-options")) h.set("x-frame-options", "DENY"); |
| 70 | h.set("permissions-policy", "camera=(), microphone=(), geolocation=()"); |
| 71 | }); |
| 72 | |
| 73 | app.get("/healthz", (c) => c.json({ ok: true })); |
| 74 | |
| 75 | app.route("/", avatars); |
| 76 | app.route("/", wellKnown); |
| 77 | app.route("/", git); |
| 78 | app.route("/", runner); |
| 79 | app.route("/", mailApi); |
| 80 | app.route("/", registry); |
| 81 | app.route("/", packagesApi); |
| 82 | app.route("/", api); |
| 83 | app.route("/", files); |
| 84 | app.route("/", mcp); |
| 85 | |
| 86 | // Live updates: authorise in the Worker, then hand the WebSocket to the repository's Durable Object. |
| 87 | app.get("/:owner/:repo/live", async (c) => { |
| 88 | if (c.req.header("upgrade") !== "websocket") return c.text("expected websocket", 426); |
| 89 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo")); |
| 90 | return stub.fetch(c.req.raw); |
| 91 | }); |
| 92 | |
| 93 | app.get("/:owner/:repo/actions/runs/:runId/artifacts/:name", async (c) => { |
| 94 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo")); |
| 95 | const name = c.req.param("name"); |
| 96 | if (!(await stub.artifacts(c.req.param("runId"))).some((a) => a.name === name)) return c.json({ message: "artifact not found" }, 404); |
| 97 | const obj = await c.env.BUCKET.get(await stub.artifactKeyForRun(c.req.param("runId"), name)); |
| 98 | if (!obj) return c.json({ message: "artifact expired" }, 410); |
| 99 | return new Response(obj.body, { |
| 100 | headers: { |
| 101 | "content-type": "application/gzip", |
| 102 | "content-length": String(obj.size), |
| 103 | "content-disposition": `attachment; filename="${name.replace(/[^\w.-]+/g, "_")}.tar.gz"`, |
| 104 | }, |
| 105 | }); |
| 106 | }); |
| 107 | |
| 108 | // ---------------------------------------------------------------- events queue |
| 109 | |
| 110 | const GITHUB_EVENT: Record<DiffineryEvent["type"], string> = { |
| 111 | push: "push", |
| 112 | pull_request: "pull_request", |
| 113 | issues: "issues", |
| 114 | issue_comment: "issue_comment", |
| 115 | pull_request_review: "pull_request_review", |
| 116 | workflow_run: "workflow_run", |
| 117 | workflow_dispatch: "workflow_dispatch", |
| 118 | release: "release", |
| 119 | status: "status", |
| 120 | deployment: "deployment", |
| 121 | deployment_status: "deployment_status", |
| 122 | repository_dispatch: "repository_dispatch", |
| 123 | audit: "audit", |
| 124 | }; |
| 125 | |
| 126 | function webhookPayload(env: Env, ev: Exclude<DiffineryEvent, { type: "audit" }>) { |
| 127 | const repository = { id: numericId(ev.repoId), node_id: nodeId("Repository", ev.repoId), name: ev.repo, full_name: `${ev.owner}/${ev.repo}`, owner: { login: ev.owner }, html_url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}`, clone_url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}.git` }; |
| 128 | const sender = { id: ev.actor.id, login: ev.actor.login, type: ev.actor.kind === "agent" ? "Bot" : ev.actor.kind === "system" ? "Bot" : "User", on_behalf_of: ev.actor.onBehalfOf }; |
| 129 | const { type: _t, repoId: _r, owner: _o, repo: _n, actor: _a, ...rest } = ev; |
| 130 | const action = "action" in ev ? (ev.action === "merged" ? "closed" : ev.action) : undefined; |
| 131 | const extra: Record<string, unknown> = {}; |
| 132 | if (ev.type === "pull_request") { |
| 133 | extra.pull_request = { number: ev.number, html_url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/pull/${ev.number}`, head: { ref: ev.headRef, sha: ev.headSha }, base: { ref: ev.baseRef }, merged: ev.action === "merged" }; |
| 134 | } |
| 135 | if (ev.type === "issues" || ev.type === "issue_comment") extra.issue = { number: ev.number, html_url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/issues/${ev.number}` }; |
| 136 | // GitHub-shaped events (release, status, deployment, ...) carry their payload ready-made. |
| 137 | if ("payload" in ev) { |
| 138 | const { payload, at: _at, ...head } = rest as typeof rest & { payload: Record<string, unknown>; at: string }; |
| 139 | return { ...head, ...payload, action, repository, sender }; |
| 140 | } |
| 141 | return { ...rest, action, repository, sender, ...extra }; |
| 142 | } |
| 143 | |
| 144 | async function deliverWebhooks(env: Env, ev: Exclude<DiffineryEvent, { type: "audit" }>): Promise<void> { |
| 145 | const stub = repoStub(env, ev.repoId); |
| 146 | const hooks = (await stub.webhooks()).filter((h) => h.active && (h.events.includes("*") || h.events.includes(GITHUB_EVENT[ev.type]))); |
| 147 | if (!hooks.length) return; |
| 148 | const body = JSON.stringify(webhookPayload(env, ev)); |
| 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); |
| 153 | } |
| 154 | |
| 155 | const webhookDeps = (env: Env): WebhookDeps => ({ |
| 156 | hook: async (repoId, hookId) => (await repoStub(env, repoId).webhooks()).find((h) => h.id === hookId && h.active) ?? null, |
| 157 | record: (repoId, hookId, status) => repoStub(env, repoId).recordDelivery(hookId, status), |
| 158 | requeue: async (r, delaySeconds) => { |
| 159 | await env.EVENTS.send(r as unknown as DiffineryEvent, { delaySeconds }); |
| 160 | }, |
| 161 | }); |
| 162 | |
| 163 | async function handleEvent(env: Env, dir: Directory, ev: DiffineryEvent): Promise<void> { |
| 164 | if (ev.type === "audit") { |
| 165 | await dir.audit({ actorId: ev.actor.id, actorLogin: ev.actor.login, onBehalfOf: ev.actor.onBehalfOf, action: ev.action, target: ev.target, detail: ev.detail, at: ev.at }); |
| 166 | return; |
| 167 | } |
| 168 | if (ev.type === "push") { |
| 169 | await dir.audit({ actorId: ev.actor.id, actorLogin: ev.actor.login, onBehalfOf: ev.actor.onBehalfOf, action: ev.forced ? "git.force_push" : ev.deleted ? "git.delete_ref" : "git.push", target: `${ev.owner}/${ev.repo}:${ev.ref}`, detail: { before: ev.before, after: ev.after }, at: ev.at }); |
| 170 | } |
| 171 | if (ev.type === "pull_request" && ev.action === "merged") { |
| 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 }); |
| 173 | } |
| 174 | await runEventStages(env, dir, ev, ["webhooks", "notify"]); |
| 175 | } |
| 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 | |
| 192 | export async function queue(batch: MessageBatch<DiffineryEvent | MailJob>, env: Env): Promise<void> { |
| 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. |
| 194 | if (batch.queue.startsWith("diffinery-mail")) return mailQueue(batch as MessageBatch<MailJob>, env); |
| 195 | return events(batch as MessageBatch<DiffineryEvent>, env); |
| 196 | } |
| 197 | |
| 198 | async function events(batch: MessageBatch<DiffineryEvent>, env: Env): Promise<void> { |
| 199 | const dir = new Directory(env); |
| 200 | // Keep an append-only copy of every event in R2 for long-term audit retention. |
| 201 | const day = new Date().toISOString().slice(0, 10).replaceAll("-", "/"); |
| 202 | const fresh = batch.messages.filter((m) => !["webhook_retry", "stage_retry"].includes((m.body as { type: string }).type)); |
| 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)); |
| 204 | for (const msg of batch.messages) { |
| 205 | try { |
| 206 | const body = msg.body as DiffineryEvent | WebhookRetry | StageRetry<RepoEvent>; |
| 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); |
| 209 | else await handleEvent(env, dir, body); |
| 210 | msg.ack(); |
| 211 | } catch (e) { |
| 212 | console.error("event handling failed", msg.body.type, e); |
| 213 | msg.retry({ delaySeconds: Math.min(300, 10 * 2 ** msg.attempts) }); |
| 214 | } |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | // ---------------------------------------------------------------- cron: workflow schedules |
| 219 | |
| 220 | export async function scheduled(controller: ScheduledController, env: Env, ctx?: ExecutionContext): Promise<void> { |
| 221 | // Container registry housekeeping: abandoned uploads and bytes no image uses any more. |
| 222 | const cleanup = new RegistryStore(env).cleanup().catch((e) => console.error("registry cleanup failed", e)); |
| 223 | if (ctx) ctx.waitUntil(cleanup); |
| 224 | else await cleanup; |
| 225 | // The trigger runs every 5 minutes (GitHub's minimum schedule interval); fire anything due in the window. |
| 226 | const end = new Date(controller.scheduledTime); |
| 227 | end.setUTCSeconds(0, 0); |
| 228 | const minutes = Array.from({ length: 5 }, (_, i) => new Date(end.getTime() - i * 60_000)); |
| 229 | const { results } = await env.DB.prepare("SELECT s.repo_id, s.workflow_path, s.cron FROM schedules s JOIN repos r ON r.id = s.repo_id WHERE r.archived = 0").all<{ repo_id: string; workflow_path: string; cron: string }>(); |
| 230 | await Promise.all([ |
| 231 | ...results.map(async (s) => { |
| 232 | try { |
| 233 | if (minutes.some((m) => cronMatches(s.cron, m))) await repoStub(env, s.repo_id).scheduleTick(s.workflow_path, s.cron); |
| 234 | } catch (e) { |
| 235 | console.error("schedule failed", s.repo_id, s.workflow_path, e); |
| 236 | } |
| 237 | }), |
| 238 | // Token expiry reminders and email log pruning. |
| 239 | mailCron(env, new Directory(env), end).catch((e) => console.error("mail cron failed", e)), |
| 240 | ]); |
| 241 | } |