network
tcmts.examples.userinput.network · #dfsu3ha5 · ts file
// tcm: namespace userinput
// Reading the network: records (from Slingshot's cache, else the author's PDS), which records point at a record
// (Constellation's backlinks), identities and profiles. Everything read is kept for the page's life, and the same
// request made twice at once is made once: a board shows each person and each count many times.
import assert from "node:assert/strict";
import { parseAtUri, services } from "#codebase/userinput";
import type { Stored } from "#codebase/userinput";
/** An error from a service: what was asked, and what came back. */
export class ServiceError extends Error {
readonly status: number;
constructor(what: string, status: number, detail: string) {
super(`${what}: ${status} ${detail}`.trim());
this.status = status;
}
}
/** Requests in flight to each service, and the ones waiting their turn: a page asks a lot at once, and services
* limit how fast anyone may. */
export const lanes = new Map<string, { running: number; waiting: (() => void)[] }>();
/** How many requests to one service run at once. */
export const AT_ONCE = 24;
/**
* Fetch, a few at a time per service, trying again (after a pause that grows) when the service says it's being asked
* too fast (429) or is briefly unavailable (503).
*/
export async function politeFetch(url: string, init?: RequestInit): Promise<Response> {
const host = new URL(url).host;
const lane = lanes.get(host) ?? { running: 0, waiting: [] };
lanes.set(host, lane);
if (lane.running >= AT_ONCE) await new Promise<void>((r) => lane.waiting.push(r));
lane.running++;
try {
for (let attempt = 0; ; attempt++) {
const res = await fetch(url, init);
if ((res.status !== 429 && res.status !== 503) || attempt >= 4) return res;
const after = Number(res.headers.get("retry-after"));
await new Promise((r) => setTimeout(r, after > 0 ? after * 1000 : 400 * 2 ** attempt));
}
} finally {
lane.running--;
lane.waiting.shift()?.();
}
}
/** GET a service's XRPC method, as JSON. */
export async function xrpc<T>(base: string, method: string, params: Record<string, string | number | string[] | undefined>): Promise<T> {
const q = new URLSearchParams();
for (const [k, v] of Object.entries(params)) {
if (v === undefined) continue;
for (const one of Array.isArray(v) ? v : [v]) q.append(k, String(one));
}
const res = await politeFetch(`${base}/xrpc/${method}?${q}`);
if (!res.ok) throw new ServiceError(method, res.status, await res.text().catch(() => ""));
return (await res.json()) as T;
}
/** What's been read (or is being read), by request: for the page's life. */
export const memo = new Map<string, Promise<unknown>>();
/** `read`, once per key: a failure isn't kept, so it's tried again next time. */
export function once<T>(key: string, read: () => Promise<T>): Promise<T> {
const known = memo.get(key) as Promise<T> | undefined;
if (known) return known;
const p = read().catch((e: unknown) => {
memo.delete(key);
throw e;
});
memo.set(key, p);
return p;
}
/** Forget what was read about these keys' prefixes (after a write changes it). */
export function forget(...prefixes: string[]) {
for (const k of memo.keys()) if (prefixes.some((p) => k.startsWith(p))) memo.delete(k);
}
/**
* What userinput.app's index (its app view) has at a path (`board/<did>/<rkey>`): a board or a thread with all it
* shows, read once, in one request. Undefined when there's no index, it doesn't have it (yet), it's slow to answer,
* or it says it can't give all of it (`complete: false`): then the page reads the records from the network itself.
*/
export function indexed<T>(path: string): Promise<T | undefined> {
const base = services().index;
if (!base) return Promise.resolve(undefined);
return once(`index:${path}`, async () => {
try {
const res = await politeFetch(`${base}/api/${path}`, { cache: "no-store", signal: AbortSignal.timeout(8000) });
if (!res.ok) return undefined;
const body = (await res.json()) as T & { complete?: boolean };
return body.complete === false ? undefined : body;
} catch {
return undefined;
}
});
}
/**
* After a write, once the network has it: ask the index to read again what the page has read from it, and forget
* what it said, so the next read is what it has now.
*/
export async function reindex(): Promise<void> {
const base = services().index;
const paths = [...memo.keys()].filter((k) => k.startsWith("index:")).map((k) => k.slice("index:".length));
await Promise.all(
paths.map(async (p) => {
memo.delete(`index:${p}`);
await politeFetch(`${base}/api/${p}?repass=1`, { cache: "no-store", signal: AbortSignal.timeout(20000) }).catch(() => undefined);
}),
);
}
/** Who someone is: their DID, handle and the server holding their repo. */
export interface Identity {
did: string;
handle: string;
pds: string;
}
/** A handle or DID, resolved (through Slingshot). */
export function resolveIdentity(identifier: string): Promise<Identity> {
return once(`identity:${identifier}`, async () => {
const d = await xrpc<{ did: string; handle: string; pds: string }>(services().slingshot, "blue.microcosm.identity.resolveMiniDoc", { identifier });
return { did: d.did, handle: d.handle, pds: d.pds };
});
}
/** A record, by its `at://` URI: undefined if it's gone. */
export function getRecord<T>(uri: string): Promise<Stored<T> | undefined> {
return once(`record:${uri}`, async () => {
const at = parseAtUri(uri);
if (!at) return undefined;
try {
return await xrpc<Stored<T>>(services().slingshot, "com.atproto.repo.getRecord", { repo: at.did, collection: at.collection, rkey: at.rkey });
} catch (e) {
if (e instanceof ServiceError && (e.status === 400 || e.status === 404)) return undefined;
throw e;
}
});
}
/** Records, by URI, read together; the ones gone are left out. */
export async function getRecords<T>(uris: string[]): Promise<Stored<T>[]> {
const got = await Promise.all(uris.map((u) => getRecord<T>(u).catch(() => undefined)));
return got.filter((r): r is Stored<T> => !!r);
}
/** Every record of a collection in someone's repo, read from their PDS (newest first). */
export function listRecords<T>(did: string, collection: string, most = 1000): Promise<Stored<T>[]> {
return once(`list:${did}:${collection}`, async () => {
const { pds } = await resolveIdentity(did);
const out: Stored<T>[] = [];
let cursor: string | undefined;
do {
const page = await xrpc<{ records: Stored<T>[]; cursor?: string }>(pds, "com.atproto.repo.listRecords", { repo: did, collection, limit: 100, cursor });
out.push(...page.records);
cursor = page.records.length ? page.cursor : undefined;
} while (cursor && out.length < most);
return out;
});
}
/** A record pointing at another, as Constellation lists them. */
export interface Backlink {
did: string;
collection: string;
rkey: string;
}
/** Where a link is: the collection of the records that hold it, and the path to it in them (`space.uri`). */
export const source = (collection: string, path: string): string => `${collection}:${path}`;
/**
* The records (of `collection`, at `path`) pointing at `subject`: every one, page by page, up to `most`. `dids`
* keeps only the ones by those authors (moderation counts only from a space's owner and moderators).
*/
export function backlinks(subject: string, collection: string, path: string, opts: { dids?: string[]; most?: number } = {}): Promise<Backlink[]> {
const most = opts.most ?? 1000;
return once(`links:${subject}:${collection}:${path}:${opts.dids?.join(",") ?? ""}:${most}`, async () => {
const out: Backlink[] = [];
let cursor: string | undefined;
do {
const page = await xrpc<{ records: Backlink[]; cursor?: string | null }>(services().constellation, "blue.microcosm.links.getBacklinks", {
subject,
source: source(collection, path),
did: opts.dids,
limit: 100,
cursor,
});
out.push(...page.records);
cursor = page.cursor ?? undefined;
} while (cursor && out.length < most);
return out;
});
}
/** How many records (of `collection`, at `path`) point at `subject`. */
export function backlinkCount(subject: string, collection: string, path: string): Promise<number> {
return once(`count:${subject}:${collection}:${path}`, async () => {
const r = await xrpc<{ total: number }>(services().constellation, "blue.microcosm.links.getBacklinksCount", { subject, source: source(collection, path) });
return r.total;
});
}
/**
* How many records of each kind point at `target`, in one request: by `collection:path` (`app.userinput.upvote:
* subject.uri`). What's worth asking about one at a time (a discussion's statuses, its edits) is asked about only
* if there are any.
*/
export function linkCounts(target: string): Promise<Map<string, number>> {
return once(`counts:${target}`, async () => {
const res = await politeFetch(`${services().constellation}/links/all?target=${encodeURIComponent(target)}`, { headers: { accept: "application/json" } });
if (!res.ok) throw new ServiceError("links/all", res.status, await res.text().catch(() => ""));
const body = (await res.json()) as { links: Record<string, Record<string, { records: number }>> };
const out = new Map<string, number>();
for (const [collection, paths] of Object.entries(body.links ?? {}))
for (const [p, n] of Object.entries(paths)) out.set(`${collection}:${p.replace(/^\./, "")}`, n.records);
return out;
});
}
/** The `at://` URI of a backlink's record. */
export const linkUri = (l: Backlink): string => `at://${l.did}/${l.collection}/${l.rkey}`;
/** Someone as the network shows them. */
export interface Profile {
did: string;
handle: string;
displayName?: string;
avatar?: string;
}
/** Profiles waiting to be asked for, together (the appview takes 25 at a time). */
export const profileQueue = new Map<string, ((p: Profile) => void)[]>();
/** Ask for the queued profiles, 25 at a time, all at once. */
export async function flushProfiles() {
const dids = [...profileQueue.keys()];
const waiting = new Map(profileQueue);
profileQueue.clear();
const batches: string[][] = [];
for (let i = 0; i < dids.length; i += 25) batches.push(dids.slice(i, i + 25));
await Promise.all(
batches.map(async (batch) => {
let got: Profile[] = [];
try {
got = (await xrpc<{ profiles: Profile[] }>(services().appview, "app.bsky.actor.getProfiles", { actors: batch })).profiles;
} catch {
// (a profile that can't be read is shown by its DID)
}
for (const did of batch) {
const p = got.find((x) => x.did === did) ?? { did, handle: did };
for (const done of waiting.get(did) ?? []) done(p);
}
}),
);
}
/** Someone's profile: asked for with the others wanted at the same moment. */
export function getProfile(did: string): Promise<Profile> {
return once(`profile:${did}`, () => new Promise<Profile>((resolve) => {
const first = profileQueue.size === 0;
profileQueue.set(did, [...(profileQueue.get(did) ?? []), resolve]);
if (first) setTimeout(() => void flushProfiles(), 10);
}));
}
/** Handles starting with what's typed, for mentions and member pickers. */
export async function searchHandles(q: string): Promise<Profile[]> {
if (!q.trim()) return [];
const r = await xrpc<{ actors: Profile[] }>(services().typeahead, "app.bsky.actor.searchActorsTypeahead", { q, limit: 8 });
return r.actors;
}
/** Image sizes, as the image service resizes them. */
export const IMAGE_SIZES = {
icon: "w=192,h=192,fit=cover,q=95,f=auto",
card: "w=384,h=384,fit=cover,q=95,f=auto",
full: "w=1440,fit=scale-down,q=95,f=auto",
} as const;
/** The URL of an image a record holds (by its author's DID and the blob's CID), at a size. */
export const imageUrl = (did: string, cid: string, size: keyof typeof IMAGE_SIZES): string =>
`${services().images}/blob/${encodeURIComponent(did)}/${encodeURIComponent(cid)}/${IMAGE_SIZES[size]}`;
export const testOnceMakesARequestOnce = async () => {
memo.clear();
let calls = 0;
const read = () => once("k", async () => (calls++, 42));
assert.deepEqual(await Promise.all([read(), read()]), [42, 42]);
assert.equal(calls, 1);
// a failure isn't kept
let fail = true;
const flaky = () => once("f", async () => {
if (fail) throw new Error("no");
return 1;
});
await assert.rejects(flaky());
fail = false;
assert.equal(await flaky(), 1);
forget("k");
await read();
assert.equal(calls, 2);
memo.clear();
};
used by