| 1 | import { Hono } from "hono"; |
| 2 | import { getCookie } from "hono/cookie"; |
| 3 | import type { DiffineryEvent } from "@diffinery/shared"; |
| 4 | import type { Env } from "./env.js"; |
| 5 | import { credentialFrom, resolvePrincipal, type AppVars } from "./auth/principal.js"; |
| 6 | import { Directory } from "./lib/directory.js"; |
| 7 | import { getFirst } from "./lib/storage-keys.js"; |
| 8 | import { BOOKMARK_COOKIE, BOOKMARK_HEADER, d1Session, sessionConstraint } from "./lib/d1.js"; |
| 9 | import { SESSION_COOKIE } from "./auth/sessions.js"; |
| 10 | import { newId, now } from "./lib/util.js"; |
| 11 | import { bucketsFor, overLimit, tooMany } from "./lib/rate-limit.js"; |
| 12 | import { deliverWebhook, type WebhookDeps, type WebhookRetry } from "./lib/webhooks.js"; |
| 13 | import { runStages, type Stage, type StageRetry } from "./lib/stages.js"; |
| 14 | import { apiNotFound, errorResponse, loadRepo, repoStub } from "./http/common.js"; |
| 15 | import { git } from "./http/git.js"; |
| 16 | import { runner } from "./http/runner.js"; |
| 17 | import { actionsRuntime } from "./http/actions-runtime.js"; |
| 18 | import { api, wellKnown } from "./http/api.js"; |
| 19 | import { avatars } from "./http/avatars.js"; |
| 20 | import { mailApi } from "./mail/api.js"; |
| 21 | import { files } from "./http/files.js"; |
| 22 | import { webhookPayload } from "./lib/webhook-payload.js"; |
| 23 | import { mcp } from "./mcp/server.js"; |
| 24 | import { registry } from "./registry/http.js"; |
| 25 | import { packagesApi } from "./registry/api.js"; |
| 26 | import { RegistryStore } from "./registry/store.js"; |
| 27 | import { mailCron, notifyEvent } from "./mail/notify.js"; |
| 28 | import { schedulerStub } from "./lib/schedules.js"; |
| 29 | import { mailQueue } from "./mail/send.js"; |
| 30 | import { onReviewEvent } from "./agentic/reviews.js"; |
| 31 | import { recordActivity } from "./agentic/today.js"; |
| 32 | import { modelProxy } from "./agentic/proxy.js"; |
| 33 | import type { MailJob } from "./mail/types.js"; |
| 34 | |
| 35 | export const app = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 36 | |
| 37 | app.onError((e) => errorResponse(e)); |
| 38 | app.notFound((c) => (c.req.path.startsWith("/api/") ? apiNotFound() : c.text("404 Not Found", 404))); |
| 39 | |
| 40 | // The model proxy for agent tasks: its own task-scoped token, not an account credential (agentic/proxy.ts). |
| 41 | app.all("/-/model/:provider/*", (c) => modelProxy(c)); |
| 42 | |
| 43 | // Security headers on every response, then identity: tokens, Access JWT, or anonymous. |
| 44 | app.use("*", async (c, next) => { |
| 45 | // Browser sessions ride on cookies, so refuse state-changing requests that come from another site. |
| 46 | // Token-authenticated calls (git, CLI, MCP, runners) carry no ambient credentials and are exempt. |
| 47 | if (!["GET", "HEAD", "OPTIONS"].includes(c.req.method) && !credentialFrom(c.req.raw)) { |
| 48 | const site = c.req.header("sec-fetch-site"); |
| 49 | const origin = c.req.header("origin"); |
| 50 | const ours = new URL(c.env.PUBLIC_URL).origin; |
| 51 | if ((site && site !== "same-origin" && site !== "none") || (!site && origin && origin !== ours)) { |
| 52 | return c.json({ message: "cross-site request refused" }, 403); |
| 53 | } |
| 54 | } |
| 55 | // One D1 session per request: replica reads that still see this client's earlier writes (see lib/d1.ts). |
| 56 | const token = !!credentialFrom(c.req.raw); |
| 57 | const cookie = getCookie(c, BOOKMARK_COOKIE); |
| 58 | const d1 = d1Session(c.env, sessionConstraint({ header: c.req.header(BOOKMARK_HEADER), cookie, token })); |
| 59 | c.env = d1.env; |
| 60 | const dir = new Directory(c.env); |
| 61 | c.set("dir", dir); |
| 62 | // The Actions runtime services authenticate with their own signed tokens and URLs (http/actions-runtime.ts). |
| 63 | const runtime = c.req.path.startsWith("/twirp/") || c.req.path.startsWith("/_runtime/"); |
| 64 | // Avatar images are public and on every page many times over: skip the session lookup for them. |
| 65 | const avatar = (c.req.method === "GET" || c.req.method === "HEAD") && c.req.path.startsWith("/api/avatars/"); |
| 66 | c.set("principal", runtime || avatar ? { type: "anonymous" } : await resolvePrincipal(c.req.raw, c.env, dir)); |
| 67 | const limited = await overLimit(c.env, bucketsFor(c.req.method, c.req.path, c.get("principal"), c.req.header("cf-connecting-ip") ?? "unknown")); |
| 68 | if (limited) c.res = tooMany(c, limited); |
| 69 | else await next(); |
| 70 | if (c.res.status === 101) return; |
| 71 | const h = c.res.headers; |
| 72 | const bookmark = d1.bookmark(); |
| 73 | if (bookmark) { |
| 74 | h.set(BOOKMARK_HEADER, bookmark); |
| 75 | // Browsers: remember it once they have written something or are signed in. |
| 76 | const browser = !token && (!["GET", "HEAD"].includes(c.req.method) || cookie !== undefined || getCookie(c, SESSION_COOKIE) !== undefined); |
| 77 | if (browser && !avatar && bookmark !== cookie) h.append("set-cookie", `${BOOKMARK_COOKIE}=${bookmark}; Max-Age=86400; Path=/; HttpOnly; Secure; SameSite=Lax`); |
| 78 | } |
| 79 | h.set("strict-transport-security", "max-age=63072000; includeSubDomains; preload"); |
| 80 | h.set("x-content-type-options", "nosniff"); |
| 81 | h.set("referrer-policy", "strict-origin-when-cross-origin"); |
| 82 | if (!h.has("x-frame-options")) h.set("x-frame-options", "DENY"); |
| 83 | h.set("permissions-policy", "camera=(), microphone=(), geolocation=()"); |
| 84 | }); |
| 85 | |
| 86 | app.get("/healthz", (c) => c.json({ ok: true })); |
| 87 | |
| 88 | app.route("/", avatars); |
| 89 | app.route("/", wellKnown); |
| 90 | app.route("/", git); |
| 91 | app.route("/", runner); |
| 92 | app.route("/", mailApi); |
| 93 | app.route("/", registry); |
| 94 | app.route("/", packagesApi); |
| 95 | app.route("/", actionsRuntime); |
| 96 | app.route("/", api); |
| 97 | app.route("/", files); |
| 98 | app.route("/", mcp); |
| 99 | |
| 100 | // Live updates: authorise in the Worker, then hand the WebSocket to the repository's Durable Object. |
| 101 | app.get("/:owner/:repo/live", async (c) => { |
| 102 | if (c.req.header("upgrade") !== "websocket") return c.text("expected websocket", 426); |
| 103 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), "read", { follow: true }); |
| 104 | return stub.fetch(c.req.raw); |
| 105 | }); |
| 106 | |
| 107 | app.get("/:owner/:repo/actions/runs/:runId/artifacts/:name", async (c) => { |
| 108 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo")); |
| 109 | // By name, or by id (the artifact-url output of upload-artifact). |
| 110 | const name = c.req.param("name"); |
| 111 | const all = await stub.artifacts(c.req.param("runId")); |
| 112 | const artifact = all.find((a) => a.name === name) ?? (/^\d+$/.test(name) ? all.find((a) => a.id === Number(name)) : undefined); |
| 113 | if (!artifact) return c.json({ message: "artifact not found" }, 404); |
| 114 | const obj = await getFirst(c.env.BUCKET, artifact.r2Keys); |
| 115 | if (!obj) return c.json({ message: "artifact expired" }, 410); |
| 116 | return new Response(obj.body, { |
| 117 | headers: { |
| 118 | "content-type": artifact.contentType, |
| 119 | "content-length": String(obj.size), |
| 120 | "content-disposition": `attachment; filename="${artifact.fileName.replace(/[^\w.-]+/g, "_")}"`, |
| 121 | }, |
| 122 | }); |
| 123 | }); |
| 124 | |
| 125 | // ---------------------------------------------------------------- events queue |
| 126 | |
| 127 | const GITHUB_EVENT: Record<DiffineryEvent["type"], string> = { |
| 128 | push: "push", |
| 129 | pull_request: "pull_request", |
| 130 | issues: "issues", |
| 131 | issue_comment: "issue_comment", |
| 132 | pull_request_review: "pull_request_review", |
| 133 | workflow_run: "workflow_run", |
| 134 | workflow_dispatch: "workflow_dispatch", |
| 135 | release: "release", |
| 136 | status: "status", |
| 137 | deployment: "deployment", |
| 138 | deployment_status: "deployment_status", |
| 139 | repository_dispatch: "repository_dispatch", |
| 140 | audit: "audit", |
| 141 | }; |
| 142 | |
| 143 | async function deliverWebhooks(env: Env, ev: Exclude<DiffineryEvent, { type: "audit" }>): Promise<void> { |
| 144 | const stub = repoStub(env, ev.repoId); |
| 145 | const hooks = (await stub.webhooks()).filter((h) => h.active && (h.events.includes("*") || h.events.includes(GITHUB_EVENT[ev.type]))); |
| 146 | if (!hooks.length) return; |
| 147 | const body = JSON.stringify(webhookPayload(env.PUBLIC_URL, ev, await new Directory(env).repoById(ev.repoId))); |
| 148 | // Each hook retries on its own (deliverWebhook requeues it). Once a hook may have been sent to, |
| 149 | // an error must not fail the stage, or the stage retry would send every hook again. |
| 150 | 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))); |
| 151 | for (const r of sent) if (r.status === "rejected") console.error("webhook delivery bookkeeping failed", ev.type, r.reason); |
| 152 | } |
| 153 | |
| 154 | const webhookDeps = (env: Env): WebhookDeps => ({ |
| 155 | ownHost: new URL(env.PUBLIC_URL).hostname, |
| 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 | // The Today view finds repositories by their latest activity (agentic/today.ts). |
| 169 | await recordActivity(env, ev); |
| 170 | if (ev.type === "push") { |
| 171 | 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 }); |
| 172 | } |
| 173 | if (ev.type === "pull_request" && ev.action === "merged") { |
| 174 | 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 }); |
| 175 | } |
| 176 | // Agent code review (agentic/reviews.ts): opt-in per repository, best effort, never retried. |
| 177 | await onReviewEvent(env, dir, ev).catch((e) => console.error("agent review event failed", e)); |
| 178 | await runEventStages(env, dir, ev, ["webhooks", "notify"]); |
| 179 | } |
| 180 | |
| 181 | type RepoEvent = Exclude<DiffineryEvent, { type: "audit" }>; |
| 182 | |
| 183 | /** Webhooks and notifications run independently; a failed one is retried alone (see lib/stages.ts). */ |
| 184 | function runEventStages(env: Env, dir: Directory, ev: RepoEvent, stages: Stage[], attempt = 0): Promise<void> { |
| 185 | return runStages( |
| 186 | ev, |
| 187 | stages, |
| 188 | (stage) => (stage === "webhooks" ? deliverWebhooks(env, ev) : notifyEvent(env, dir, ev)), |
| 189 | async (r, delaySeconds) => { |
| 190 | await env.EVENTS.send(r as unknown as DiffineryEvent, { delaySeconds }); |
| 191 | }, |
| 192 | attempt, |
| 193 | ); |
| 194 | } |
| 195 | |
| 196 | export async function queue(batch: MessageBatch<DiffineryEvent | MailJob>, env: Env): Promise<void> { |
| 197 | // Email has its own queue (diffinery-mail): its jobs can hold sign-in and invite links, so they stay out of the event archive. |
| 198 | if (batch.queue.startsWith("diffinery-mail")) return mailQueue(batch as MessageBatch<MailJob>, env); |
| 199 | return events(batch as MessageBatch<DiffineryEvent>, env); |
| 200 | } |
| 201 | |
| 202 | async function events(batch: MessageBatch<DiffineryEvent>, env: Env): Promise<void> { |
| 203 | const dir = new Directory(env); |
| 204 | // Keep an append-only copy of every event in R2 for long-term audit retention. |
| 205 | const day = new Date().toISOString().slice(0, 10).replaceAll("-", "/"); |
| 206 | const fresh = batch.messages.filter((m) => !["webhook_retry", "stage_retry"].includes((m.body as { type: string }).type)); |
| 207 | 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)); |
| 208 | for (const msg of batch.messages) { |
| 209 | try { |
| 210 | const body = msg.body as DiffineryEvent | WebhookRetry | StageRetry<RepoEvent>; |
| 211 | if (body.type === "webhook_retry") await deliverWebhook(webhookDeps(env), body); |
| 212 | else if (body.type === "stage_retry") await runEventStages(env, dir, body.event, [body.stage], body.attempt); |
| 213 | else await handleEvent(env, dir, body); |
| 214 | msg.ack(); |
| 215 | } catch (e) { |
| 216 | console.error("event handling failed", msg.body.type, e); |
| 217 | msg.retry({ delaySeconds: Math.min(300, 10 * 2 ** msg.attempts) }); |
| 218 | } |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | // ---------------------------------------------------------------- cron: housekeeping |
| 223 | |
| 224 | export async function scheduled(controller: ScheduledController, env: Env, ctx?: ExecutionContext): Promise<void> { |
| 225 | // Container registry housekeeping: abandoned uploads and bytes no image uses any more. |
| 226 | const cleanup = new RegistryStore(env).cleanup().catch((e) => console.error("registry cleanup failed", e)); |
| 227 | if (ctx) ctx.waitUntil(cleanup); |
| 228 | else await cleanup; |
| 229 | const at = new Date(controller.scheduledTime); |
| 230 | at.setUTCSeconds(0, 0); |
| 231 | await Promise.all([ |
| 232 | // Workflow schedules fire from the scheduler's own alarm (src/scheduler.ts); this trigger only makes sure it is set. |
| 233 | schedulerStub(env).ensure().catch((e) => console.error("scheduler alarm check failed", e)), |
| 234 | // Token expiry reminders and email log pruning. |
| 235 | mailCron(env, new Directory(env), at).catch((e) => console.error("mail cron failed", e)), |
| 236 | // Expire abandoned draft bulk imports after 7 days, and null their stored GitHub token (Low 5). |
| 237 | env.DB.prepare("UPDATE imports SET status = 'cancelled', token_ciphertext = NULL, error = 'draft expired', finished_at = ?, updated_at = ? WHERE status = 'draft' AND created_at < ?") |
| 238 | .bind(now(), now(), new Date(Date.now() - 7 * 86400_000).toISOString()) |
| 239 | .run() |
| 240 | .catch((e) => console.error("draft import expiry failed", e)), |
| 241 | ]); |
| 242 | } |