path: root/worker.js

// The worker routes every request to the one container and answers the
// container's outbound calls from bindings. Everything about git is syncd's.
import { DurableObject, WorkerEntrypoint } from "cloudflare:workers";

// syncd fronts git-forge, so this is the only port the worker talks to.
const PORT = 8081;

// syncd answers this prefix itself; refused below, so only the alarm asks.
const SYNCD = "/.syncd/";
const INFLIGHT = SYNCD + "inflight";

// None of these resolve: all three are intercepted and served from bindings.
const R2_HOST = "r2.forge.internal";
const MAIL_HOST = "mail.forge.internal";
// Cloudflare does not forward container stdout to observability, so syncd sends
// its own and git-forge's here.
const LOG_HOST = "logs.forge.internal";

const BOOT_TRIES = 40;
const BOOT_INTERVAL_MS = 250;

// An alarm is the idle timer.
const SLEEP_AFTER_MS = 10 * 60 * 1000;

// oidcEnv picks out the client secrets by prefix: the provider keys are in
// forge.json, not here.
function oidcEnv(env) {
	const out = {};
	for (const [name, value] of Object.entries(env)) {
		if (name.startsWith("GIT_FORGE_OIDC") && typeof value === "string") {
			out[name] = value;
		}
	}
	// Names, never values.
	console.log("container env: oidc variables forwarded:", Object.keys(out).join(", ") || "none");
	return out;
}

export default {
	fetch(request, env) {
		// One name, so one container globally.
		return env.FORGE.get(env.FORGE.idFromName("forge")).fetch(request);
	},
};

export class Forge extends DurableObject {
	async fetch(request) {
		const url = new URL(request.url);
		// Only the alarm may ask, and it does not come through here.
		if (url.pathname.startsWith(SYNCD)) {
			return new Response("not found\n", { status: 404 });
		}
		// base_url from the request, as https: git-forge refuses a form post
		// whose Origin disagrees with it, and the scheme puts Secure on every
		// cookie for the life of the container.
		await this.boot(url.origin.replace(/^http:/, "https:"));
		await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS);
		// The container listens on http; the rest passes through untouched.
		return await this.ctx.container
			.getTcpPort(PORT)
			.fetch(request.url.replace(/^https:/, "http:"), request);
	}

	// SIGTERM rather than destroy: syncd drains git-forge and flushes once more
	// on the way out.
	async alarm() {
		if (!this.ctx.container.running) {
			console.log("alarm: already stopped");
			return;
		}
		const busy = await this.inflight();
		if (busy > 0) {
			console.log(`alarm: ${busy} in flight, staying up`);
			await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS);
			return;
		}
		console.log("alarm: idle, stopping the container");
		this.ctx.container.signal(15);
		// Armed again, so a container that did not stop is signalled once more.
		await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS);
	}

	// Asked of syncd: a fetch here resolves at the headers, minutes before a
	// clone's last byte.
	async inflight() {
		try {
			const r = await this.ctx.container.getTcpPort(PORT).fetch(`http://forge${INFLIGHT}`);
			return parseInt(await r.text(), 10) || 0;
		} catch (e) {
			// Unreachable counts as idle, so a wedged container is stopped.
			console.log("alarm: asked what was in flight and got:", e);
			return 0;
		}
	}

	// blockConcurrencyWhile makes this once for two requests arriving together.
	async boot(origin) {
		if (this.ctx.container.running) return;
		await this.ctx.blockConcurrencyWhile(async () => {
			if (this.ctx.container.running) return;

			// Wired before start, so no early outbound call escapes to the
			// internet. Absent on an old runtime.
			if (this.ctx.exports?.Outbound === undefined) {
				throw new Error(
					"ctx.exports.Outbound is undefined, so outbound interception cannot be wired: " +
						"raise compatibility_date in wrangler.jsonc",
				);
			}
			const proxy = this.ctx.exports.Outbound({ props: {} });
			await this.ctx.container.interceptOutboundHttp(R2_HOST, proxy);
			await this.ctx.container.interceptOutboundHttp(LOG_HOST, proxy);
			// https because git-forge validates mail.mailgun.base_url as https.
			await this.ctx.container.interceptOutboundHttps(MAIL_HOST, proxy);

			// No datadir: syncd tells git-forge where it is.
			this.ctx.container.start({
				enableInternet: true, // OIDC, and mail if it goes out directly
				env: {
					GIT_FORGE_BASE_URL: this.env.GIT_FORGE_BASE_URL || origin,
					// Every request arrives from syncd otherwise; the edge
					// overwrites this header.
					GIT_FORGE_CLIENT_IP_HEADER: "CF-Connecting-IP",
					// A secret, not a file in a datadir that does not survive a
					// restart.
					GIT_FORGE_COOKIE_KEY: this.env.GIT_FORGE_COOKIE_KEY ?? "",
					SYNCD_R2: `http://${R2_HOST}`,
					SYNCD_LOGS: `http://${LOG_HOST}/`,
					// git-forge requires a key for the mailgun driver; the
					// handler below ignores it.
					GIT_FORGE_MAIL_MAILGUN_API_KEY: "intercepted",
					...oidcEnv(this.env),
				},
			});
			// Not awaited: it exists so a container that dies is logged.
			this.ctx.container.monitor().catch((e) => console.log("container exited:", e));

			await this.waitReady();
		});
	}

	// waitReady polls git-forge's health endpoint through syncd, so a pass
	// proves both are up.
	async waitReady() {
		const port = this.ctx.container.getTcpPort(PORT);
		for (let i = 0; i < BOOT_TRIES; i++) {
			try {
				const r = await port.fetch("http://forge/healthz");
				if (r.ok) return;
			} catch {
				// Not listening yet.
			}
			await new Promise((r) => setTimeout(r, BOOT_INTERVAL_MS));
		}
		throw new Error("container did not become ready");
	}
}

// Outbound answers the intercepted hosts from outside the container, where the
// bindings live.
export class Outbound extends WorkerEntrypoint {
	fetch(request) {
		const url = new URL(request.url);
		if (url.hostname === LOG_HOST) return this.log(request);
		if (url.hostname === MAIL_HOST) return this.mail(request);
		if (url.hostname === R2_HOST) return this.object(request, url);
		// Named above, not defaulted: a new interception has to say what
		// serves it.
		return new Response("no binding serves this host\n", { status: 404 });
	}

	// One console.log per line, as written.
	async log(request) {
		const body = await request.text();
		for (const line of body.split("\n")) {
			if (line !== "") console.log(line);
		}
		return new Response(null, { status: 204 });
	}

	// Four verbs, all syncd needs. Keys are the path; a get on / with a prefix
	// is a list.
	async object(request, url) {
		const key = decodeURIComponent(url.pathname.slice(1));
		const bucket = this.env.BUCKET;

		if (request.method === "GET" && key === "") {
			const out = await bucket.list({
				prefix: url.searchParams.get("prefix") ?? undefined,
				cursor: url.searchParams.get("cursor") || undefined,
				limit: 1000,
			});
			return Response.json({
				keys: out.objects.map((o) => o.key),
				cursor: out.truncated ? out.cursor : "",
			});
		}
		if (key === "") return new Response("no key", { status: 400 });

		switch (request.method) {
			case "GET": {
				const obj = await bucket.get(key);
				if (obj === null) return new Response("not found", { status: 404 });
				return new Response(obj.body, {
					headers: { "content-length": String(obj.size) },
				});
			}
			case "PUT": {
				// Streamed: a packfile can be larger than a worker's memory.
				const length = request.headers.get("content-length");
				if (length === null) return new Response("content-length required", { status: 411 });
				await bucket.put(key, request.body, {
					httpMetadata: { contentType: "application/octet-stream" },
				});
				return new Response(null, { status: 204 });
			}
			case "DELETE":
				await bucket.delete(key);
				return new Response(null, { status: 204 });
		}
		return new Response("bad method", { status: 405 });
	}

	// git-forge's mailgun driver is one form-encoded post, so answering that
	// shape gives it any sender.
	async mail(request) {
		const form = await request.formData();
		const to = form.get("to");
		const subject = form.get("subject") ?? "";
		const text = form.get("text") ?? "";
		const from = form.get("from") ?? "";

		if (this.env.SEND_EMAIL === undefined) {
			// Success, or git-forge's spool retries six times.
			console.log(`mail: no sender bound, dropping message to ${to}: ${subject}`);
			return Response.json({ id: "dropped", message: "no sender bound" });
		}

		// The binding wants a bare address; git-forge sends "Name <addr>".
		const addr = (s) => {
			const m = /<([^>]+)>/.exec(s);
			return m ? m[1] : s.trim();
		};

		const message = {
			from: addr(from),
			to: addr(to),
			subject,
			text,
		};
		const unsub = form.get("h:List-Unsubscribe");
		if (unsub) message.headers = { "List-Unsubscribe": unsub };

		try {
			await this.env.SEND_EMAIL.send(message);
		} catch (e) {
			// A 5xx is what makes git-forge's spool retry.
			console.log("mail: send failed:", e);
			return new Response(String(e), { status: 502 });
		}
		return Response.json({ id: "sent", message: "Queued. Thank you." });
	}
}