| 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 { hmacSha256Hex, newId } from "./lib/util.js"; |
| 11 | import { errorResponse, loadRepo, repoStub } from "./http/common.js"; |
| 12 | import { git } from "./http/git.js"; |
| 13 | import { runner } from "./http/runner.js"; |
| 14 | import { api, wellKnown } from "./http/api.js"; |
| 15 | import { mcp } from "./mcp/server.js"; |
| 16 | import { registry } from "./registry/http.js"; |
| 17 | import { packagesApi } from "./registry/api.js"; |
| 18 | import { RegistryStore } from "./registry/store.js"; |
| 19 | |
| 20 | export const app = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 21 | |
| 22 | app.onError((e) => errorResponse(e)); |
| 23 | |
| 24 | // Security headers on every response, then identity: tokens, Access JWT, or anonymous. |
| 25 | app.use("*", async (c, next) => { |
| 26 | // Browser sessions ride on cookies, so refuse state-changing requests that come from another site. |
| 27 | // Token-authenticated calls (git, CLI, MCP, runners) carry no ambient credentials and are exempt. |
| 28 | if (!["GET", "HEAD", "OPTIONS"].includes(c.req.method) && !credentialFrom(c.req.raw)) { |
| 29 | const site = c.req.header("sec-fetch-site"); |
| 30 | const origin = c.req.header("origin"); |
| 31 | const ours = new URL(c.env.PUBLIC_URL).origin; |
| 32 | if ((site && site !== "same-origin" && site !== "none") || (!site && origin && origin !== ours)) { |
| 33 | return c.json({ message: "cross-site request refused" }, 403); |
| 34 | } |
| 35 | } |
| 36 | // One D1 session per request: replica reads that still see this client's earlier writes (see lib/d1.ts). |
| 37 | const token = !!credentialFrom(c.req.raw); |
| 38 | const cookie = getCookie(c, BOOKMARK_COOKIE); |
| 39 | const d1 = d1Session(c.env, sessionConstraint({ header: c.req.header(BOOKMARK_HEADER), cookie, token })); |
| 40 | c.env = d1.env; |
| 41 | const dir = new Directory(c.env); |
| 42 | c.set("dir", dir); |
| 43 | c.set("principal", await resolvePrincipal(c.req.raw, c.env, dir)); |
| 44 | await next(); |
| 45 | if (c.res.status === 101) return; |
| 46 | const h = c.res.headers; |
| 47 | const bookmark = d1.bookmark(); |
| 48 | if (bookmark) { |
| 49 | h.set(BOOKMARK_HEADER, bookmark); |
| 50 | // Browsers: remember it once they have written something or are signed in. |
| 51 | const browser = !token && (!["GET", "HEAD"].includes(c.req.method) || cookie !== undefined || getCookie(c, SESSION_COOKIE) !== undefined); |
| 52 | if (browser && bookmark !== cookie) h.append("set-cookie", `${BOOKMARK_COOKIE}=${bookmark}; Max-Age=86400; Path=/; HttpOnly; Secure; SameSite=Lax`); |
| 53 | } |
| 54 | h.set("strict-transport-security", "max-age=63072000; includeSubDomains; preload"); |
| 55 | h.set("x-content-type-options", "nosniff"); |
| 56 | h.set("referrer-policy", "strict-origin-when-cross-origin"); |
| 57 | if (!h.has("x-frame-options")) h.set("x-frame-options", "DENY"); |
| 58 | h.set("permissions-policy", "camera=(), microphone=(), geolocation=()"); |
| 59 | }); |
| 60 | |
| 61 | app.get("/healthz", (c) => c.json({ ok: true })); |
| 62 | |
| 63 | app.route("/", wellKnown); |
| 64 | app.route("/", git); |
| 65 | app.route("/", runner); |
| 66 | app.route("/", registry); |
| 67 | app.route("/", packagesApi); |
| 68 | app.route("/", api); |
| 69 | app.route("/", mcp); |
| 70 | |
| 71 | // Live updates: authorise in the Worker, then hand the WebSocket to the repository's Durable Object. |
| 72 | app.get("/:owner/:repo/live", async (c) => { |
| 73 | if (c.req.header("upgrade") !== "websocket") return c.text("expected websocket", 426); |
| 74 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo")); |
| 75 | return stub.fetch(c.req.raw); |
| 76 | }); |
| 77 | |
| 78 | app.get("/:owner/:repo/actions/runs/:runId/artifacts/:name", async (c) => { |
| 79 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo")); |
| 80 | const name = c.req.param("name"); |
| 81 | if (!(await stub.artifacts(c.req.param("runId"))).some((a) => a.name === name)) return c.json({ message: "artifact not found" }, 404); |
| 82 | const obj = await c.env.BUCKET.get(await stub.artifactKeyForRun(c.req.param("runId"), name)); |
| 83 | if (!obj) return c.json({ message: "artifact expired" }, 410); |
| 84 | return new Response(obj.body, { |
| 85 | headers: { |
| 86 | "content-type": "application/gzip", |
| 87 | "content-length": String(obj.size), |
| 88 | "content-disposition": `attachment; filename="${name.replace(/[^\w.-]+/g, "_")}.tar.gz"`, |
| 89 | }, |
| 90 | }); |
| 91 | }); |
| 92 | |
| 93 | // ---------------------------------------------------------------- events queue |
| 94 | |
| 95 | const GITHUB_EVENT: Record<DiffineryEvent["type"], string> = { |
| 96 | push: "push", |
| 97 | pull_request: "pull_request", |
| 98 | issues: "issues", |
| 99 | issue_comment: "issue_comment", |
| 100 | pull_request_review: "pull_request_review", |
| 101 | workflow_run: "workflow_run", |
| 102 | workflow_dispatch: "workflow_dispatch", |
| 103 | audit: "audit", |
| 104 | }; |
| 105 | |
| 106 | function webhookPayload(env: Env, ev: Exclude<DiffineryEvent, { type: "audit" }>) { |
| 107 | const repository = { id: 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` }; |
| 108 | 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 }; |
| 109 | const { type: _t, repoId: _r, owner: _o, repo: _n, actor: _a, ...rest } = ev; |
| 110 | const action = "action" in ev ? (ev.action === "merged" ? "closed" : ev.action) : undefined; |
| 111 | const extra: Record<string, unknown> = {}; |
| 112 | if (ev.type === "pull_request") { |
| 113 | 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" }; |
| 114 | } |
| 115 | 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}` }; |
| 116 | return { ...rest, action, repository, sender, ...extra }; |
| 117 | } |
| 118 | |
| 119 | async function deliverWebhooks(env: Env, ev: Exclude<DiffineryEvent, { type: "audit" }>): Promise<void> { |
| 120 | const stub = repoStub(env, ev.repoId); |
| 121 | const hooks = (await stub.webhooks()).filter((h) => h.active && (h.events.includes("*") || h.events.includes(GITHUB_EVENT[ev.type]))); |
| 122 | if (!hooks.length) return; |
| 123 | const body = JSON.stringify(webhookPayload(env, ev)); |
| 124 | await Promise.all( |
| 125 | hooks.map(async (h) => { |
| 126 | const delivery = newId(); |
| 127 | let status = 0; |
| 128 | try { |
| 129 | const res = await fetch(h.url, { |
| 130 | method: "POST", |
| 131 | headers: { |
| 132 | "content-type": "application/json", |
| 133 | "user-agent": "Diffinery-Hookshot/1", |
| 134 | "x-github-event": GITHUB_EVENT[ev.type], |
| 135 | "x-github-delivery": delivery, |
| 136 | "x-diffinery-event": ev.type, |
| 137 | "x-hub-signature-256": `sha256=${await hmacSha256Hex(h.secret, body)}`, |
| 138 | }, |
| 139 | body, |
| 140 | redirect: "manual", |
| 141 | signal: AbortSignal.timeout(10_000), |
| 142 | }); |
| 143 | status = res.status; |
| 144 | } catch { |
| 145 | status = 0; |
| 146 | } |
| 147 | await stub.recordDelivery(h.id, status); |
| 148 | }), |
| 149 | ); |
| 150 | } |
| 151 | |
| 152 | /** People who should hear about an event: participants, assignees and requested reviewers, minus the actor. */ |
| 153 | async function notifyParticipants(env: Env, dir: Directory, ev: DiffineryEvent): Promise<void> { |
| 154 | if (ev.type === "workflow_run") { |
| 155 | if (ev.action !== "completed" || ev.conclusion !== "failure" || ev.actor.kind === "system") return; |
| 156 | await dir.notify([ev.actor.id], { repoId: ev.repoId, kind: "run_failed", title: `Workflow run failed in ${ev.owner}/${ev.repo}`, url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/actions/runs/${ev.runId}` }); |
| 157 | return; |
| 158 | } |
| 159 | if (ev.type !== "pull_request" && ev.type !== "issues" && ev.type !== "issue_comment" && ev.type !== "pull_request_review") return; |
| 160 | // Queue removals already notify through the merge queue's comment on the pull request. |
| 161 | if (ev.type === "pull_request" && (ev.action === "synchronize" || ev.action === "enqueued" || ev.action === "dequeued")) return; |
| 162 | const { issue, comments, reviews } = await repoStub(env, ev.repoId).issue(ev.number); |
| 163 | const ids = new Set<string>([issue.author.id, ...comments.map((c) => c.author.id), ...reviews.map((r) => r.author.id)]); |
| 164 | for (const login of [...issue.assignees, ...(issue.pull?.requestedReviewers ?? [])]) { |
| 165 | const a = await dir.accountByLogin(login); |
| 166 | if (a) ids.add(a.id); |
| 167 | } |
| 168 | ids.delete(ev.actor.id); |
| 169 | if (!ids.size) return; |
| 170 | const kind = issue.kind === "pull" ? "pull" : "issues"; |
| 171 | const verb = ev.type === "issue_comment" ? "commented on" : ev.type === "pull_request_review" ? `reviewed (${ev.state.toLowerCase()})` : ev.action; |
| 172 | await dir.notify([...ids].slice(0, 100), { |
| 173 | repoId: ev.repoId, |
| 174 | kind: ev.type, |
| 175 | title: `${ev.actor.login} ${verb} ${ev.owner}/${ev.repo}#${ev.number}: ${issue.title}`, |
| 176 | url: `${env.PUBLIC_URL}/${ev.owner}/${ev.repo}/${kind}/${ev.number}`, |
| 177 | }); |
| 178 | } |
| 179 | |
| 180 | async function handleEvent(env: Env, dir: Directory, ev: DiffineryEvent): Promise<void> { |
| 181 | if (ev.type === "audit") { |
| 182 | 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 }); |
| 183 | return; |
| 184 | } |
| 185 | if (ev.type === "push") { |
| 186 | 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 }); |
| 187 | } |
| 188 | if (ev.type === "pull_request" && ev.action === "merged") { |
| 189 | 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 }); |
| 190 | } |
| 191 | await Promise.all([deliverWebhooks(env, ev), notifyParticipants(env, dir, ev)]); |
| 192 | } |
| 193 | |
| 194 | export async function queue(batch: MessageBatch<DiffineryEvent>, env: Env): Promise<void> { |
| 195 | const dir = new Directory(env); |
| 196 | // Keep an append-only copy of every event in R2 for long-term audit retention. |
| 197 | const day = new Date().toISOString().slice(0, 10).replaceAll("-", "/"); |
| 198 | await env.BUCKET.put(`events/${day}/${Date.now()}-${newId()}.jsonl`, batch.messages.map((m) => JSON.stringify(m.body)).join("\n")).catch((e) => console.error("event archive failed", e)); |
| 199 | for (const msg of batch.messages) { |
| 200 | try { |
| 201 | await handleEvent(env, dir, msg.body); |
| 202 | msg.ack(); |
| 203 | } catch (e) { |
| 204 | console.error("event handling failed", msg.body.type, e); |
| 205 | msg.retry({ delaySeconds: Math.min(300, 10 * 2 ** msg.attempts) }); |
| 206 | } |
| 207 | } |
| 208 | } |
| 209 | |
| 210 | // ---------------------------------------------------------------- cron: workflow schedules |
| 211 | |
| 212 | export async function scheduled(controller: ScheduledController, env: Env, ctx?: ExecutionContext): Promise<void> { |
| 213 | // Container registry housekeeping: abandoned uploads and bytes no image uses any more. |
| 214 | const cleanup = new RegistryStore(env).cleanup().catch((e) => console.error("registry cleanup failed", e)); |
| 215 | if (ctx) ctx.waitUntil(cleanup); |
| 216 | else await cleanup; |
| 217 | // The trigger runs every 5 minutes (GitHub's minimum schedule interval); fire anything due in the window. |
| 218 | const end = new Date(controller.scheduledTime); |
| 219 | end.setUTCSeconds(0, 0); |
| 220 | const minutes = Array.from({ length: 5 }, (_, i) => new Date(end.getTime() - i * 60_000)); |
| 221 | 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 }>(); |
| 222 | await Promise.all( |
| 223 | results.map(async (s) => { |
| 224 | try { |
| 225 | if (minutes.some((m) => cronMatches(s.cron, m))) await repoStub(env, s.repo_id).scheduleTick(s.workflow_path, s.cron); |
| 226 | } catch (e) { |
| 227 | console.error("schedule failed", s.repo_id, s.workflow_path, e); |
| 228 | } |
| 229 | }), |
| 230 | ); |
| 231 | } |