Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175import { isRecord, parseSpecArg, UPSTREAM_BUDGET, walkClosure, type Loader } from "./closure.ts";import { CLOSURE_TYPE, HEARTBEAT_MS, MAX_SPECS, NSID, SERVICE_ID, SERVICE_TYPE, type ClosureRequest } from "./protocol.ts";
/** A string store with expiry: memory by default, or KV, Redis, the Workers Cache API. */export interface GatewayCache { get(key: string): Promise<string | undefined>; put(key: string, value: string, ttlSeconds: number): Promise<void>;}
export interface GatewayOptions { /** Upstream registry. Default https://registry.npmjs.org. */ registry?: string; /** Every upstream request goes through this. Default globalThis.fetch. */ fetch?: (input: string, init?: RequestInit) => Promise<Response>; cache?: GatewayCache; /** Sent upstream, so a registry can reach whoever runs this gateway. */ userAgent?: string; /** Serve `/.well-known/did.json` for this gateway's did:web, naming it with a * `#nibling_npm` service entry. */ didWeb?: boolean; /** How long the stream may go without a line before a `wait` line. Default HEARTBEAT_MS, * which clients count on; shorter only makes more lines. */ heartbeatMs?: number; /** The most upstream documents one walk loads. Default UPSTREAM_BUDGET (900), for a * Worker's subrequest limit; a gateway with no such limit can raise it. */ budget?: number;}
/** How long a packument, and a closure built from packuments, is reused: the registry's own * max-age, so a gateway is no staler than npm's cache. A version document never changes. */const PACKUMENT_TTL = 300;const VERSION_TTL = 31_536_000;
const CORS = { "access-control-allow-origin": "*", "access-control-allow-methods": "GET, OPTIONS", "access-control-max-age": "86400" };const CORGI = "application/vnd.npm.install-v1+json; q=1.0, application/json; q=0.8, */*";
export function memoryCache(maxEntries = 5000, now = () => Date.now()): GatewayCache { const entries = new Map<string, { value: string; expires: number }>(); return { async get(key) { const hit = entries.get(key); if (!hit || hit.expires <= now()) return undefined; return hit.value; }, async put(key, value, ttl) { entries.delete(key); entries.set(key, { value, expires: now() + ttl * 1000 }); for (const oldest of entries.keys()) { if (entries.size <= maxEntries) break; entries.delete(oldest); } }, };}
const error = (status: number, error: string, message: string) => Response.json({ error, message }, { status, headers: { ...CORS, "cache-control": "no-store" } });
/** The query's params as a request, or why they are not one. */export function parseClosureParams(params: URLSearchParams): ClosureRequest | string { const specs = [...new Set(params.getAll("spec"))]; if (!specs.length || specs.length > MAX_SPECS) return `pass 1–${MAX_SPECS} spec=<name>@<spec>`; if (specs.some((s) => !parseSpecArg(s))) return "a spec's package name is not valid"; const aliases = new Map<string, string>(); for (const pair of params.getAll("alias")) { const [pkg, sub] = pair.split(":"); if (!pkg || !sub || !parseSpecArg(pkg) || !parseSpecArg(sub)) return `alias=${pair} is not <package>:<substitute>`; aliases.set(pkg, sub); } const bindings = [...new Set(params.getAll("binding"))]; if (bindings.length > 10 || bindings.some((b) => !/^[a-z0-9-]{1,64}$/.test(b))) return "a binding suffix is not valid"; return { specs, aliases, bindings };}
/** The canonical form of a request, for its cache key: the same specs and rules in any order. */const requestKey = (r: ClosureRequest) => JSON.stringify([[...r.specs].sort(), [...(r.aliases ?? [])].sort(), [...(r.bindings ?? [])].sort()]);
/** A `fetch` handler that answers dev.nibling.npm.getClosure, for Deno.serve, Bun.serve, a * Worker's `export default { fetch }`, or Node through bin/serve.ts. */export function createGateway(options: GatewayOptions = {}): { fetch(request: Request): Promise<Response> } { const registry = (options.registry ?? "https://registry.npmjs.org").replace(/\/+$/, ""); const upstream = options.fetch ?? ((input, init) => globalThis.fetch(input, init)); const cache = options.cache ?? memoryCache(); const headers = (accept: string): Record<string, string> => ({ accept, ...(options.userAgent && { "user-agent": options.userAgent }) }); const docUrl = (name: string) => `${registry}/${name.replace("/", "%2f")}`;
// A 404 is kept as "absent" for the packument's max-age; other failures are not kept. const json = async (url: string, accept: string, ttl: number): Promise<Record<string, unknown> | "absent" | undefined> => { const key = `doc ${accept === CORGI ? "corgi" : "json"} ${url}`; const hit = await cache.get(key).catch(() => undefined); if (hit !== undefined) return hit === "absent" ? "absent" : JSON.parse(hit); const res = await upstream(url, { headers: headers(accept), signal: AbortSignal.timeout(10_000) }); if (res.status === 404) { await res.body?.cancel(); await cache.put(key, "absent", PACKUMENT_TTL).catch(() => undefined); return "absent"; } if (!res.ok) { await res.body?.cancel(); return undefined; } const text = await res.text(); const body: unknown = JSON.parse(text); if (!isRecord(body)) return undefined; await cache.put(key, text, ttl).catch(() => undefined); return body; }; const loader: Loader = { registry, packument: (name) => json(docUrl(name), CORGI, PACKUMENT_TTL), version: async (name, version) => { const doc = await json(`${docUrl(name)}/${version}`, "application/json", VERSION_TTL); return doc === "absent" ? undefined : doc; }, };
return { async fetch(request) { const url = new URL(request.url); if (request.method === "OPTIONS") return new Response(null, { status: 204, headers: CORS }); if (request.method !== "GET") return error(405, "MethodNotAllowed", `${request.method} is not supported`); if (options.didWeb && url.pathname === "/.well-known/did.json") { const did = `did:web:${encodeURIComponent(url.host)}`; return Response.json({ "@context": ["https://www.w3.org/ns/did/v1"], id: did, service: [{ id: SERVICE_ID, type: SERVICE_TYPE, serviceEndpoint: url.origin }] }, { headers: CORS }); } if (url.pathname !== `/xrpc/${NSID}`) return error(404, "MethodNotImplemented", `${url.pathname} is not served here`); const parsed = parseClosureParams(url.searchParams); if (typeof parsed === "string") return error(400, "InvalidRequest", parsed); const key = `closure ${requestKey(parsed)}`; const hit = await cache.get(key).catch(() => undefined); const head = (status: string) => ({ ...CORS, "content-type": CLOSURE_TYPE, "cache-control": `public, max-age=${PACKUMENT_TTL}`, "x-nib-cache": status }); if (hit !== undefined) return new Response(hit, { headers: head("hit") }); // Each line goes out as the walk finds it, with a `wait` line whenever the upstream keeps // it quiet for HEARTBEAT_MS. A walk that finishes is kept whole, without its waits; one // that fails ends without its `end` line, so the client knows to fetch the rest itself. const encoder = new TextEncoder(); const lines: string[] = []; const reading = new AbortController(); const body = new ReadableStream<Uint8Array>({ // Not awaited: a pending start() would hold the lines back until the walk ends. start(controller) { let last = Date.now(); const send = (text: string) => { last = Date.now(); if (!reading.signal.aborted) controller.enqueue(encoder.encode(text)); }; const every = Math.min(options.heartbeatMs ?? HEARTBEAT_MS, HEARTBEAT_MS); const heartbeat = setInterval(() => { if (Date.now() - last >= every) send('{"t":"wait"}\n'); }, every / 4); void (async () => { try { await walkClosure(loader, parsed, (line) => { const text = `${JSON.stringify(line)}\n`; lines.push(text); send(text); }, reading.signal, options.budget ?? UPSTREAM_BUDGET); if (!reading.signal.aborted) await cache.put(key, lines.join(""), PACKUMENT_TTL).catch(() => undefined); } catch { // The client has every line sent so far, and no `end`. } clearInterval(heartbeat); if (!reading.signal.aborted) controller.close(); })(); }, cancel() { reading.abort(); }, }); return new Response(body, { headers: head("miss") }); }, };}