Something went wrong. Try again.
An easy to set up and operate PDS admin panel. forked from brookie.blog/pds-operator
Something went wrong. Try again.
13 kB · 370 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371import { randomBytes } from "node:crypto";import WebSocket from "ws";import { decodeMultiple } from "cbor-x";
export interface RepoEntry { did: string; active?: boolean; status?: string; head?: string; rev?: string;}
export interface AccountInfo { did: string; handle: string; email?: string; indexedAt: string;}
export interface InviteCode { code: string; available: number; disabled: boolean; forAccount: string; createdBy: string; createdAt: string; uses: { usedBy: string; usedAt: string }[];}
export interface AdminAccount { did: string; handle: string; email?: string; indexedAt: string; status?: "active" | "takendown" | "deactivated"; avatar?: string; labels?: string[]; /** the invite code this account signed up with; forAccount is the code's owner */ invitedBy?: { code: string; forAccount?: string };}
export class PdsClient { private accessJwt: string | null = null; private refreshJwt: string | null = null; private sessionPromise: Promise<void> | null = null;
constructor( public readonly hostname: string, private adminPassword: string, /** * When set, admin calls sign in as this account (which must have admin rights on the * PDS) and send bearer tokens instead of `admin:<password>` basic auth. Needed for * PDS implementations without an admin password, e.g. tranquil-pds. adminPassword is * then this account's password — an app password works and sidesteps 2FA. */ private adminIdentifier?: string, ) {}
private async createSession(): Promise<void> { const res = await fetch(`https://${this.hostname}/xrpc/com.atproto.server.createSession`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ identifier: this.adminIdentifier, password: this.adminPassword }), }); if (!res.ok) { const body = await res.text().catch(() => ""); throw new Error(`PDS admin sign-in as ${this.adminIdentifier} failed: ${res.status} ${body}`); } const { accessJwt, refreshJwt } = (await res.json()) as { accessJwt: string; refreshJwt: string; }; this.accessJwt = accessJwt; this.refreshJwt = refreshJwt; }
private async updateSession(update: () => Promise<void>): Promise<void> { let promise = this.sessionPromise; if (!promise) { promise = update(); this.sessionPromise = promise; } try { await promise; } finally { if (this.sessionPromise === promise) this.sessionPromise = null; } }
private async ensureSession(): Promise<string> { if (!this.accessJwt) await this.updateSession(() => this.createSession()); return this.accessJwt!; }
/** Any failure falls back to a fresh sign-in — never leave an expired session cached. */ private async refreshSession(): Promise<boolean> { if (!this.refreshJwt) return false; try { const res = await fetch(`https://${this.hostname}/xrpc/com.atproto.server.refreshSession`, { method: "POST", headers: { Authorization: `Bearer ${this.refreshJwt}` }, }); if (!res.ok) return false; const { accessJwt, refreshJwt } = (await res.json()) as { accessJwt: string; refreshJwt: string; }; this.accessJwt = accessJwt; this.refreshJwt = refreshJwt; return true; } catch { return false; } }
private async renewSession(expiredAccessJwt: string): Promise<void> { if (this.accessJwt !== expiredAccessJwt) return; await this.updateSession(async () => { if (this.accessJwt !== expiredAccessJwt) return; if (await this.refreshSession()) return; this.accessJwt = null; this.refreshJwt = null; await this.createSession(); }); }
private async xrpc(path: string, opts: RequestInit = {}, retry = true): Promise<any> { const accessJwt = this.adminIdentifier ? await this.ensureSession() : null; const authorization = accessJwt ? `Bearer ${accessJwt}` : `Basic ${Buffer.from(`admin:${this.adminPassword}`).toString("base64")}`; const res = await fetch(`https://${this.hostname}/xrpc/${path}`, { ...opts, headers: { ...opts.headers, Authorization: authorization, "Content-Type": "application/json", }, }); if (!res.ok) { const body = await res.text().catch(() => ""); // expired access tokens are 401 on the reference PDS but 400 ExpiredToken on tranquil-pds const expired = res.status === 401 || (res.status === 400 && body.includes("ExpiredToken")); if (expired && this.adminIdentifier && retry) { await this.renewSession(accessJwt!); return this.xrpc(path, opts, false); } throw new Error(`PDS ${path} failed: ${res.status} ${body}`); } // some procedures (e.g. disableInviteCodes) return 200 with an empty body const text = await res.text(); return text ? JSON.parse(text) : {}; }
async listRepos(cursor?: string, limit = 500): Promise<{ repos: RepoEntry[]; cursor?: string }> { const params = new URLSearchParams({ limit: String(limit) }); if (cursor) params.set("cursor", cursor); return this.xrpc(`com.atproto.sync.listRepos?${params}`) as Promise<{ repos: RepoEntry[]; cursor?: string; }>; }
async listAllRepos(): Promise<RepoEntry[]> { const repos: RepoEntry[] = []; let cursor: string | undefined; do { const page = await this.listRepos(cursor); repos.push(...page.repos); cursor = page.repos.length > 0 ? page.cursor : undefined; } while (cursor); return repos; }
/** Admin account info (handle/email/indexedAt); `dids` capped at 100 per call by the PDS. */ async accountInfos(dids: string[]): Promise<AccountInfo[]> { if (dids.length === 0) return []; const params = new URLSearchParams(); for (const did of dids) params.append("dids", did); const { infos } = (await this.xrpc(`com.atproto.admin.getAccountInfos?${params}`)) as { infos: AccountInfo[]; }; return infos; }
async repoStatus(did: string): Promise<RepoEntry> { return this.xrpc( `com.atproto.sync.getRepoStatus?did=${encodeURIComponent(did)}`, ) as Promise<RepoEntry>; }
async getAccount(did: string): Promise<AdminAccount> { return this.xrpc(`com.atproto.admin.getAccountInfo?did=${encodeURIComponent(did)}`); }
async setAccountTakedown(did: string, takedown: boolean) { return this.xrpc("com.atproto.admin.updateSubjectStatus", { method: "POST", body: JSON.stringify({ subject: { $type: "com.atproto.admin.defs#repoRef", did }, takedown: takedown ? { applied: true } : { applied: false }, }), }); }
/** * Admin-authed raw fetch for sync endpoints, which stream bytes rather than JSON. * Admin credentials also reach takendown/deactivated repos. Retries 429/5xx with * backoff so the storage sweep can run at real concurrency; other 4xx throw. */ private async authedFetch(path: string, tries = 3): Promise<Response> { let lastErr: unknown; for (let attempt = 0; attempt < tries; attempt++) { const accessJwt = this.adminIdentifier ? await this.ensureSession() : null; const authorization = accessJwt ? `Bearer ${accessJwt}` : `Basic ${Buffer.from(`admin:${this.adminPassword}`).toString("base64")}`; let res: Response; try { res = await fetch(`https://${this.hostname}/xrpc/${path}`, { headers: { Authorization: authorization }, }); } catch (err) { lastErr = err; await new Promise((r) => setTimeout(r, 1000 * 2 ** attempt)); continue; } if (res.status === 429 || res.status >= 500) { lastErr = new Error(`PDS ${path} failed: ${res.status}`); void res.body?.cancel().catch(() => {}); const retryAfter = Number(res.headers.get("retry-after")); const delay = Number.isFinite(retryAfter) && retryAfter > 0 ? retryAfter * 1000 : 1000 * 2 ** attempt; await new Promise((r) => setTimeout(r, delay)); continue; } if (!res.ok) { const body = await res.text().catch(() => ""); throw new Error(`PDS ${path} failed: ${res.status} ${body}`); } return res; } throw lastErr ?? new Error(`PDS ${path} failed after ${tries} tries`); }
private async countBody(res: Response): Promise<number> { let bytes = 0; if (res.body) { for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) { bytes += chunk.length; } } return bytes; }
/** Size of the repo's current CAR export, a storage proxy for the record data. */ async repoCarBytes(did: string): Promise<number> { const res = await this.authedFetch(`com.atproto.sync.getRepo?did=${encodeURIComponent(did)}`); return this.countBody(res); }
async listBlobCids(did: string): Promise<string[]> { const cids: string[] = []; let cursor: string | undefined; do { const params = new URLSearchParams({ did, limit: "500" }); if (cursor) params.set("cursor", cursor); const res = await this.authedFetch(`com.atproto.sync.listBlobs?${params}`); const page = (await res.json()) as { cids: string[]; cursor?: string }; cids.push(...page.cids); cursor = page.cids.length > 0 ? page.cursor : undefined; } while (cursor); return cids; }
/** Blob size from the content-length header, body cancelled unread when possible. */ async blobBytes(did: string, cid: string): Promise<number> { const res = await this.authedFetch( `com.atproto.sync.getBlob?did=${encodeURIComponent(did)}&cid=${encodeURIComponent(cid)}`, ); const len = Number(res.headers.get("content-length")); if (Number.isFinite(len) && len > 0) { void res.body?.cancel().catch(() => {}); return len; } return this.countBody(res); }
async getInviteCodes(limit = 100): Promise<{ codes: InviteCode[] }> { return this.xrpc(`com.atproto.admin.getInviteCodes?sort=recent&limit=${limit}`) as Promise<{ codes: InviteCode[]; }>; }
async createInviteCode(useCount = 1): Promise<{ code: string }> { return this.xrpc("com.atproto.server.createInviteCode", { method: "POST", body: JSON.stringify({ useCount }), }) as Promise<{ code: string }>; }
async disableInviteCodes(codes: string[]) { return this.xrpc("com.atproto.admin.disableInviteCodes", { method: "POST", body: JSON.stringify({ codes }), }); }
async resetAccountPassword(did: string): Promise<{ password: string }> { const password = randomBytes(16).toString("base64url"); await this.xrpc("com.atproto.admin.updateAccountPassword", { method: "POST", body: JSON.stringify({ did, password }), }); return { password }; }
/** * No public "current max seq" endpoint exists on the PDS. Approximate by opening a * subscribeRepos socket starting at `fromSeq` (e.g. the relay's last-known seq) so the * server immediately sends buffered backfill frames rather than waiting on live traffic, * and taking the highest seq seen within the read window. When there's a large backlog * this returns "how far it got in windowMs," not necessarily the true live head — still * far more useful than an unbounded wait for the next real-time event. */ async approximateHeadSeq(fromSeq: number, windowMs = 3000): Promise<number | null> { return new Promise((resolve) => { const ws = new WebSocket( `wss://${this.hostname}/xrpc/com.atproto.sync.subscribeRepos?cursor=${fromSeq}`, ); let maxSeq: number | null = null; let opened = false;
const finish = () => { clearTimeout(timer); ws.terminate(); // Socket opened but no frames arrived: the PDS has nothing past fromSeq, so the // head is (at least) fromSeq — report "caught up" rather than "unknown". resolve(maxSeq ?? (opened ? fromSeq : null)); }; const timer = setTimeout(finish, windowMs);
ws.once("open", () => { opened = true; }); ws.on("message", (data: Buffer) => { try { const frame = readFrame(data); if (typeof frame.body?.seq === "number") { maxSeq = maxSeq == null ? frame.body.seq : Math.max(maxSeq, frame.body.seq); } } catch { // ignore undecodable frames } });
ws.once("error", finish); }); }}
/** * Each event-stream frame is two concatenated CBOR values: a header ({op, t}) then a body. * Shared by subscribeRepos and subscribeLabels consumers. */export function readFrame(buf: Buffer): { header?: { op?: number; t?: string }; body?: any } { const values = decodeMultiple(buf) as unknown[] | undefined; return { header: values?.[0] as { op?: number; t?: string }, body: values?.[1] };}