lambda.ts
12.8 kB · typescript · 349 lines
1import type { S3Client } from "bun";2import { createHash } from "node:crypto";3import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmSync, symlinkSync, writeFileSync } from "node:fs";4import { join } from "node:path";5import { client, getText, need, putBytes } from "./s3.ts";67/* WHERE */89const CACHE_DIR = process.env.BUN_INSTALL_CACHE_DIR || "/tmp/bun/cache";10const LAYER_DIR = process.env.NODE_LAYER_DIR || "/opt/node";11const SHELF_DIR = "/tmp/shelf";12const SHELF_ENV = "MRLY_SHELF";13const BACKOFF = [1000, 3000, 9000];14const SHA = /^[0-9a-f]{7,40}$/;1516export type Runner = (cmd: string[], cwd: string) => Promise<string>;1718export type Config = {19 source: string;20 head: string;21 dir: string;22 agent: string;23 bucket: string;24 folder: string;25 shelf?: string;26 sources?: string[];27 prepare?: (site: string, run: Runner) => Promise<string>;28 hash?: (site: string) => string;29 local?: string;30};3132export type Wake = { source: string; on: "source" | "shelf" | ""; sha: string };3334export type Head = { sha: string; etag: string; shelf: string; shelfEtag: string; data: string };3536export type Mark = { sha: string; etag: string };3738export type Reply = {39 ok: boolean;40 status: number;41 json: () => Promise<unknown>;42 text: () => Promise<string>;43 headers: { get: (name: string) => string | null };44};4546export type Get = (url: string, init: { headers: Record<string, string> }) => Promise<Reply>;4748/* CLOCK */4950let mark = Date.now();5152function log(line: string) {53 const now = Date.now();54 console.log(`${line} ${now - mark}ms`);55 mark = now;56}5758const short = (sha: string) => sha.slice(0, 7) || "none";5960/* RETRY */6162async function retry<T>(label: string, work: () => Promise<T>): Promise<T> {63 for (let attempt = 0; ; attempt++) {64 try {65 return await work();66 } catch (error) {67 if (attempt === BACKOFF.length) throw error;68 const wait = BACKOFF[attempt] ?? 1000;69 log(`${label} failed, retry ${attempt + 1} in ${wait}ms: ${String((error as Error).message ?? error).slice(0, 160)}`);70 await Bun.sleep(wait);71 }72 }73}7475/* JSON */7677function json(text: string): Record<string, unknown> {78 try {79 const value = JSON.parse(text);80 return value && typeof value === "object" ? (value as Record<string, unknown>) : {};81 } catch {82 return {};83 }84}8586async function payload(request?: Request): Promise<string> {87 if (!request) return "";88 try {89 return await request.text();90 } catch {91 return "";92 }93}9495/* MODULES */9697export type Modules = "layer" | "stale" | "absent";9899export function modules(site: string, layer = LAYER_DIR): Modules {100 const lock = createHash("sha256").update(readFileSync(join(site, "bun.lock"))).digest("hex");101 const held = join(layer, "bun.lock.sha256");102 if (!existsSync(held)) return "absent";103 return readFileSync(held, "utf8").trim() === lock ? "layer" : "stale";104}105106/* SPAWN */107108async function spawn(cmd: string[], cwd: string, env: Record<string, string>): Promise<string> {109 const child = Bun.spawn(cmd, { cwd, env, stdout: "pipe", stderr: "pipe" });110 const [out, err] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text()]);111 const code = await child.exited;112 if (code !== 0) throw new Error(`${cmd.join(" ")}: exit ${code}\n${(err || out).trim().slice(-1500)}`);113 return out;114}115116/* CORE */117118export function lambda(config: Config) {119 const SOURCES = new Set(config.sources ?? ["push", "schedule", "manual"]);120 const DRY = Boolean(config.local) && process.env.DRY === "1";121 if (DRY) process.env.DEV = "1";122 const HOLD = process.env.DRY_DIR ?? config.local ?? "";123 const SRC = config.local ? (process.env.SRC ?? "") : "";124125 const shelfRepo = () => (config.shelf ? (process.env[config.shelf] ?? "").trim() : "");126127 /* EVENT */128129 function readEvent(text: string): Wake {130 const outer = json(text);131 const body = typeof outer.body === "string" ? json(outer.body) : outer;132 const word = typeof body.source === "string" ? body.source.trim() : "";133 const repo = typeof body.repo === "string" ? body.repo.trim() : "";134 const named = typeof body.sha === "string" ? body.sha.trim().toLowerCase() : "";135 const shelf = shelfRepo();136 return {137 source: SOURCES.has(word) ? word : "",138 on: repo === config.source ? "source" : repo && repo === shelf ? "shelf" : "",139 sha: SHA.test(named) ? named : "",140 };141 }142143 /* HEAD */144145 const held = () => join(HOLD, "head");146147 async function readHead(s3: S3Client | null): Promise<Head> {148 const text = s3 ? await getText(s3, config.head) : existsSync(held()) ? readFileSync(held(), "utf8") : null;149 const lines = (text ?? "").trim().split("\n").map((one) => one.trim());150 const rest = lines.slice(2);151 return {152 sha: lines[0] ?? "",153 etag: lines[1] ?? "",154 shelf: config.shelf ? (rest.shift() ?? "") : "",155 shelfEtag: config.shelf ? (rest.shift() ?? "") : "",156 data: config.hash ? (rest.shift() ?? "") : "",157 };158 }159160 async function writeHead(s3: S3Client | null, head: Head): Promise<void> {161 const lines = [head.sha, head.etag];162 if (config.shelf) lines.push(head.shelf, head.shelfEtag);163 if (config.hash) lines.push(head.data);164 const body = `${lines.join("\n")}\n`;165 if (!s3) {166 mkdirSync(HOLD, { recursive: true });167 writeFileSync(held(), body);168 return;169 }170 await putBytes(s3, config.head, body, { type: "text/plain", cacheControl: "no-store" });171 }172173 /* GITHUB */174175 const headers = () => ({ "user-agent": config.agent, accept: "application/vnd.github+json" });176177 async function commit(slug: string, etag: string, get: Get = fetch): Promise<Mark | null> {178 return retry(`github ${slug}`, async () => {179 const sent: Record<string, string> = headers();180 if (etag) sent["if-none-match"] = etag;181 const res = await get(`https://api.github.com/repos/${slug}/commits/main`, { headers: sent });182 if (res.status === 304) return null;183 if (!res.ok) throw new Error(`github ${slug} ${res.status}: ${(await res.text()).slice(0, 200)}`);184 const body = (await res.json()) as { sha?: string };185 if (!body.sha) throw new Error(`github ${slug}: the commit carries no sha`);186 return { sha: body.sha, etag: res.headers.get("etag") ?? "" };187 });188 }189190 async function onMain(slug: string, sha: string, get: Get = fetch): Promise<boolean> {191 try {192 const res = await get(`https://api.github.com/repos/${slug}/compare/${sha}...main`, { headers: headers() });193 if (!res.ok) {194 log(`refused ${short(sha)} on ${slug}: compare ${res.status}`);195 return false;196 }197 const body = (await res.json()) as { status?: string };198 const state = typeof body?.status === "string" ? body.status : "";199 if (state === "ahead" || state === "identical") return true;200 log(`refused ${short(sha)} on ${slug}: ${state || "no status"}, not an ancestor of main`);201 return false;202 } catch (error) {203 log(`refused ${short(sha)} on ${slug}: compare failed, ${String((error as Error)?.message ?? error).slice(0, 160)}`);204 return false;205 }206 }207208 async function unpack(slug: string, ref: string, into: string): Promise<void> {209 if (ref !== "main" && !SHA.test(ref)) throw new Error(`codeload ${slug}: ${ref} is not a sha`);210 await retry(`codeload ${slug} ${short(ref)}`, async () => {211 const stage = `${into}.stage`;212 rmSync(stage, { recursive: true, force: true });213 rmSync(into, { recursive: true, force: true });214 const res = await fetch(`https://codeload.github.com/${slug}/tar.gz/${ref}`, { headers: { "user-agent": config.agent } });215 if (!res.ok) throw new Error(`codeload ${slug} ${ref}: ${res.status}`);216 const files = await new Bun.Archive(new Uint8Array(await res.arrayBuffer())).extract(stage);217 const [top] = readdirSync(stage);218 if (!files || !top) throw new Error(`codeload ${slug} ${ref}: the tarball is empty`);219 renameSync(join(stage, top), into);220 rmSync(stage, { recursive: true, force: true });221 });222 }223224 /* PLAN */225226 function complete(stored: Head): boolean {227 return Boolean(stored.sha) && (!shelfRepo() || Boolean(stored.shelf));228 }229230 function seen(wake: Wake, stored: Head): string {231 if (config.hash || !wake.sha || !complete(stored)) return "";232 if (wake.on === "source" && wake.sha === stored.sha) return short(wake.sha);233 if (wake.on === "shelf" && wake.sha === stored.shelf) return `shelf ${short(wake.sha)}`;234 return "";235 }236237 async function shelfMark(wake: Wake, stored: Head, get: Get): Promise<Mark> {238 const slug = shelfRepo();239 if (!slug) return { sha: "", etag: "" };240 if (wake.on === "shelf" && wake.sha && (await onMain(slug, wake.sha, get))) return { sha: wake.sha, etag: "" };241 return (await commit(slug, stored.shelf ? stored.shelfEtag : "", get)) ?? { sha: stored.shelf, etag: stored.shelfEtag };242 }243244 async function sourceMark(wake: Wake, stored: Head, get: Get): Promise<Mark> {245 if (wake.on === "source" && wake.sha && (await onMain(config.source, wake.sha, get))) return { sha: wake.sha, etag: "" };246 return (await commit(config.source, stored.sha ? stored.etag : "", get)) ?? { sha: stored.sha, etag: stored.etag };247 }248249 async function freshen(wake: Wake, stored: Head, get: Get = fetch): Promise<Head> {250 const source = await sourceMark(wake, stored, get);251 const shelf = await shelfMark(wake, stored, get);252 return { sha: source.sha, etag: source.etag, shelf: shelf.sha, shelfEtag: shelf.etag, data: stored.data };253 }254255 /* CHILD */256257 function childEnv(): Record<string, string> {258 const env: Record<string, string> = {259 ...(process.env as Record<string, string>),260 HOME: "/tmp",261 BUN_INSTALL_CACHE_DIR: CACHE_DIR,262 };263 if (shelfRepo()) env[SHELF_ENV] = join(SHELF_DIR, "research");264 else delete env[SHELF_ENV];265 return env;266 }267268 const run: Runner = (cmd, cwd) => spawn(cmd, cwd, childEnv());269270 async function install(site: string): Promise<string> {271 const state = modules(site);272 if (state === "layer") {273 const target = join(site, "node_modules");274 rmSync(target, { recursive: true, force: true });275 symlinkSync(join(LAYER_DIR, "node_modules"), target);276 return "modules layer";277 }278 await run([process.execPath, "install", "--frozen-lockfile"], site);279 return state === "stale" ? "modules layer stale, installed" : "modules installed";280 }281282 /* BUILD */283284 async function build(s3: S3Client | null, next: Head, stored: Head): Promise<string> {285 if (SRC) log(`source ${SRC}`);286 else {287 await unpack(config.source, next.sha, config.dir);288 log(`source ${short(next.sha)}`);289 }290 const slug = shelfRepo();291 if (slug) {292 await unpack(slug, next.shelf || "main", SHELF_DIR);293 if (!existsSync(join(SHELF_DIR, "research", "README.md"))) throw new Error("shelf: no research/README.md in the tarball");294 log(`shelf ${short(next.shelf)}`);295 }296 const site = join(SRC || config.dir, config.folder);297 log(await install(site));298 if (config.prepare) {299 const said = await config.prepare(site, run);300 if (said) log(said);301 }302 if (config.hash) {303 next.data = config.hash(site);304 log(`snapshot ${next.data || "none"}`);305 if (next.sha === stored.sha && next.data === stored.data) {306 if (next.etag !== stored.etag) await writeHead(s3, next);307 return `unchanged ${short(next.sha)}, data same`;308 }309 }310 const out = await run([process.execPath, "run", "push"], site);311 const line = out.split("\n").map((one) => one.trim()).find((one) => one.startsWith("push:")) ?? "push: no count line";312 log(line);313 await writeHead(s3, next);314 log(`head ${short(next.sha)}${config.shelf ? ` ${short(next.shelf)}` : ""}`);315 return line;316 }317318 /* HANDLER */319320 async function once(wake: Wake): Promise<string> {321 const began = Date.now();322 mark = began;323 const s3 = DRY ? null : client(need(config.bucket));324 const stored = await readHead(s3);325 const known = seen(wake, stored);326 if (known) {327 log(`seen ${known}`);328 return `seen ${known}`;329 }330 const next = await freshen(wake, stored);331 if (!next.sha) throw new Error(`${config.source}: no sha in the event, in the head or from github`);332 if (!config.hash && next.sha === stored.sha && next.shelf === stored.shelf) {333 if (next.etag !== stored.etag || next.shelfEtag !== stored.shelfEtag) await writeHead(s3, next);334 log(`unchanged ${short(next.sha)}`);335 return `unchanged ${short(next.sha)}`;336 }337 log(`${wake.source || "poll"} ${short(next.sha)}${config.shelf ? ` shelf ${short(next.shelf)}` : ""}`);338 const line = await build(s3, next, stored);339 const total = Date.now() - began;340 console.log(`done ${short(next.sha)} ${total}ms`);341 return `${line} in ${total}ms`;342 }343344 async function handler(request?: Request): Promise<Response> {345 return new Response(`${await once(readEvent(await payload(request)))}\n`);346 }347348 return { fetch: handler, once, readEvent, readHead, writeHead, onMain, seen, freshen };349}