Something went wrong. Try again.
A fully featured CLI used as an alternative to AppView applications (Bsky.app/mu.social)
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166import { CliError, EXIT, XrpcError } from './errors.ts';import { setting } from './config.ts';
// The wire: XRPC calls and plain JSON fetches, with the two behaviours every// request needs and no caller should have to remember.//// * **Rate limits are waited out.** A 429 (or a 503 that says when) is asked// again once the time the server names has passed: `Retry-After`, else// atproto's `ratelimit-reset`, else a short backoff — up to four times, and// never for longer than the rate-limit-wait setting (default 60 seconds) in// one go. Each wait is said on stderr. Only requests that are safe to repeat// are repeated: reads, and writes the server refused before acting on.// * **--verbose** logs every request and its status to stderr.
export type Params = Record<string, string | number | boolean | readonly string[] | undefined>;
/** Extra headers for one request, computed per request: DPoP proofs need the method and URL. */export type Authorizer = (request: { method: string; url: string }) => Promise<Record<string, string>> | Record<string, string>;
export interface CallOptions { params?: Params | undefined; /** A JSON body, or raw bytes / a string with `encoding`. Present means POST. */ input?: unknown; encoding?: string | undefined; headers?: Record<string, string> | undefined; auth?: Authorizer | undefined; method?: 'GET' | 'POST' | undefined; signal?: AbortSignal | undefined;}
let verbose = false;export function setVerbose(on: boolean): void { verbose = on;}
function log(line: string): void { if (verbose) process.stderr.write(`* ${line}\n`);}
export function xrpcUrl(service: string, nsid: string, params?: Params): string { const url = new URL(`/xrpc/${nsid}`, service); for (const [key, value] of Object.entries(params ?? {})) { if (value === undefined) continue; if (Array.isArray(value)) for (const item of value) url.searchParams.append(key, String(item)); else url.searchParams.set(key, String(value)); } return url.toString();}
/** Call an XRPC method and parse its answer. */export async function xrpc<T = unknown>(service: string, nsid: string, options: CallOptions = {}): Promise<T> { const response = await xrpcResponse(service, nsid, options); const type = response.headers.get('content-type') ?? ''; if (type.includes('json')) return await response.json() as T; if (type.startsWith('text/')) return await response.text() as T; const bytes = new Uint8Array(await response.arrayBuffer()); return (bytes.length ? bytes : {}) as T;}
export async function xrpcResponse(service: string, nsid: string, options: CallOptions = {}): Promise<Response> { const method = options.method ?? (options.input !== undefined ? 'POST' : 'GET'); const url = xrpcUrl(service, nsid, options.params); const headers: Record<string, string> = { accept: 'application/json', ...options.headers }; let body: string | Uint8Array | undefined; if (method === 'POST' && options.input !== undefined) { if (options.input instanceof Uint8Array || typeof options.input === 'string') { body = options.input; headers['content-type'] = options.encoding ?? 'application/octet-stream'; } else { body = JSON.stringify(options.input); headers['content-type'] = 'application/json'; } }
const response = await patiently(nsid, async () => { const authHeaders = options.auth ? await options.auth({ method, url }) : {}; log(`${method} ${url}`); let result: Response; try { result = await fetch(url, { method, headers: { ...headers, ...authHeaders }, ...(body !== undefined ? { body } : {}), signal: options.signal ?? AbortSignal.timeout(120_000), }); } catch (cause) { const reason = (cause as Error).name === 'TimeoutError' ? 'timed out' : String((cause as Error & { cause?: Error }).cause?.message ?? (cause as Error).message); throw new CliError(`${nsid}: could not reach ${new URL(service).host}: ${reason}`, 'NetworkError'); } log(`${result.status} ${nsid}`); return result; });
if (!response.ok) throw await errorFrom(nsid, response); return response;}
/** The XRPC error a failed response carries. */export async function errorFrom(nsid: string, response: Response): Promise<XrpcError> { let code = `HTTP${response.status}`; let message: string | undefined; const text = await response.text().catch(() => ''); try { const parsed = JSON.parse(text) as { error?: string; message?: string; error_description?: string }; if (parsed.error) code = parsed.error; message = parsed.message ?? parsed.error_description; } catch { // An HTML error page from a proxy says nothing a message should repeat. message = /^\s*</.test(text) ? `HTTP ${response.status} ${response.statusText}`.trim() : text.trim().slice(0, 300) || undefined; } const error = new XrpcError(nsid, response.status, code, message ? `${nsid}: ${message}` : `${nsid}: ${code} (HTTP ${response.status})`); (error as XrpcError & { headers?: Headers }).headers = response.headers; return error;}
/** GET a JSON document (a DID document, OAuth metadata): not XRPC, same patience. */export async function fetchJson<T>(url: string, init: RequestInit = {}): Promise<T> { const response = await patiently(url, async () => { log(`${init.method ?? 'GET'} ${url}`); try { return await fetch(url, { signal: AbortSignal.timeout(30_000), ...init, headers: { accept: 'application/json', ...(init.headers as Record<string, string> | undefined) } }); } catch (cause) { throw new CliError(`could not reach ${new URL(url).host}: ${(cause as Error & { cause?: Error }).cause?.message ?? (cause as Error).message}`, 'NetworkError'); } }); if (!response.ok) { const error = new CliError(`${url}: ${response.status} ${response.statusText}`, `HTTP${response.status}`); (error as CliError & { status: number }).status = response.status; throw error; } return await response.json() as T;}
/** * Send a request, and send it again when the server says to slow down and * names a time worth waiting for. */export async function patiently(what: string, send: () => Promise<Response>): Promise<Response> { const limit = Number(setting('rate-limit-wait')) * 1000; for (let attempt = 0; ; attempt++) { const response = await send(); if (response.status !== 429 && !(response.status === 503 && response.headers.has('retry-after'))) return response; const wait = waitFor(response.headers, attempt); if (attempt >= 4 || wait > limit) { const until = new Date(Date.now() + wait).toISOString(); throw new CliError(`${what}: rate limited until ${until}`, 'RateLimitExceeded', EXIT.rateLimited); } process.stderr.write(`atcli: ${new URL(response.url || 'http://x').host || what} says to slow down; waiting ${Math.ceil(wait / 1000)}s\n`); await response.body?.cancel().catch(() => {}); await new Promise((resolve) => setTimeout(resolve, wait)); }}
function waitFor(headers: Headers, attempt: number): number { const retry = headers.get('retry-after'); if (retry) { const seconds = Number(retry); if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000); const date = Date.parse(retry); if (!Number.isNaN(date)) return Math.max(0, date - Date.now()); } const reset = Number(headers.get('ratelimit-reset')); if (reset > 0) return Math.max(0, reset * 1000 - Date.now()); return 1000 * 2 ** attempt;}