Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130import { FETCH_TIMEOUT_MS, isPublicHttpsUrl } from '../net.ts';import type { Upstream } from '../upstream.ts';
/** The most a JSON answer or a blob may be. 5 MB is also the lexicon's per-blob cap. */export const MAX_JSON_BYTES = 5_000_000;export const MAX_BLOB_BYTES = 5_000_000;const MAX_REDIRECTS = 3;
export type UpstreamFailure = 'timeout' | 'too-large' | 'unreachable' | 'unsafe';
/** Why a call to a PDS, PLC, a knot, or microcosm failed. The gateway answers each with a 5xx, so the * SDK falls back to reading direct. */export class UpstreamError extends Error { readonly reason: UpstreamFailure; constructor(reason: UpstreamFailure, message: string) { super(message); this.reason = reason; }}
export const CORS = { 'access-control-allow-origin': '*', 'access-control-expose-headers': 'x-nib-cache',};
export function xrpcError(error: string, message: string, status: number): Response { return Response.json({ error, message }, { status, headers: { ...CORS, 'cache-control': 'no-store' } });}
export function failureResponse(error: unknown): Response { if (!(error instanceof UpstreamError)) throw error; if (error.reason === 'timeout') return xrpcError('UpstreamTimeout', error.message, 504); return xrpcError('UpstreamFailure', error.message, 502);}
/** * One request to an untrusted host. Redirects are followed by hand, at most 3, and each hop must be a * public `https:` URL like the first (`isPublicHttpsUrl`), so a PDS cannot bounce the worker inward. * `timeoutMs` covers the headers and the body. */export async function fetchUpstream(fetchFn: Upstream, url: string, init: RequestInit & { timeoutMs?: number } = {}): Promise<Response> { const { timeoutMs = FETCH_TIMEOUT_MS, ...rest } = init; const signal = AbortSignal.timeout(timeoutMs); let target = url; for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { if (!isPublicHttpsUrl(target)) throw new UpstreamError('unsafe', `refusing to fetch ${target}`); let res: Response; try { res = await fetchFn(target, { ...rest, signal, redirect: 'manual' }); } catch (error) { if (signal.aborted) throw new UpstreamError('timeout', `${new URL(target).host} did not answer in ${timeoutMs} ms`); throw new UpstreamError('unreachable', `${new URL(target).host}: ${(error as Error).message}`); } const location = res.headers.get('location'); if (res.status < 300 || res.status >= 400 || !location) return res; await res.body?.cancel(); target = new URL(location, target).toString(); } throw new UpstreamError('unsafe', `${url} redirected more than ${MAX_REDIRECTS} times`);}
/** The whole body, failing once it passes `limit` bytes, whatever Content-Length claimed. */export async function readLimited(res: Response, limit: number): Promise<Uint8Array> { if (Number(res.headers.get('content-length') ?? 0) > limit) { await res.body?.cancel(); throw new UpstreamError('too-large', `the answer is over ${limit} bytes`); } if (!res.body) return new Uint8Array(); const reader = res.body.getReader(); const chunks: Uint8Array[] = []; let size = 0; try { for (let next = await reader.read(); !next.done; next = await reader.read()) { size += next.value.byteLength; if (size > limit) { await reader.cancel(); throw new UpstreamError('too-large', `the answer is over ${limit} bytes`); } chunks.push(next.value); } } catch (error) { if (error instanceof UpstreamError) throw error; throw new UpstreamError('unreachable', `the answer broke off: ${(error as Error).message}`); } const bytes = new Uint8Array(size); let at = 0; for (const chunk of chunks) { bytes.set(chunk, at); at += chunk.byteLength; } return bytes;}
/** `body` passed through, erroring once more than `limit` bytes have gone by. */export function limitStream(body: ReadableStream<Uint8Array>, limit: number): ReadableStream<Uint8Array> { let size = 0; return body.pipeThrough( new TransformStream<Uint8Array, Uint8Array>({ transform(chunk, controller) { size += chunk.byteLength; if (size > limit) controller.error(new UpstreamError('too-large', `the answer is over ${limit} bytes`)); else controller.enqueue(chunk); }, }), );}
/** * An XRPC JSON query to `url`, answered as the upstream answered it: a 2xx or 4xx body that is JSON * and at most 5 MB. A 5xx, a non-JSON body, or a failure becomes a 502 or 504. */export async function proxyJson(fetchFn: Upstream, url: string): Promise<Response> { try { const res = await fetchUpstream(fetchFn, url, { headers: { accept: 'application/json' } }); if (res.status >= 500) { await res.body?.cancel(); return xrpcError('UpstreamFailure', `${new URL(url).host} answered ${res.status}`, 502); } const bytes = await readLimited(res, MAX_JSON_BYTES); try { JSON.parse(new TextDecoder().decode(bytes)); } catch { return xrpcError('UpstreamFailure', `${new URL(url).host} answered with something other than JSON`, 502); } return new Response(bytes, { status: res.status, headers: { ...CORS, 'content-type': 'application/json' } }); } catch (error) { return failureResponse(error); }}