Something went wrong. Try again.
Periodic backups of ATProto repos and blobs safekeeper.dummy.cafe
atproto backup
Something went wrong. Try again.
4.6 kB · 158 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159// ATProto client surface for safekeeper.//// Uses the official @atproto packages:// - @atproto/identity -> IdResolver handles handle ↔ DID resolution and// DID-doc → PDS endpoint extraction (plc + did:web).// - @atproto/api -> AtpAgent gives us typed, lexicon-generated calls// for every `com.atproto.*` endpoint.//// Streaming concession: AtpAgent's `getRepo` / `getBlob` buffer the whole// response into a `Uint8Array`. For backup workloads where accounts can be// GB-scale, we fall back to raw `fetch` for those two endpoints so the body// can be piped straight into storage.
import { AtpAgent } from '@atproto/api';import { IdResolver, MemoryCache } from '@atproto/identity';import { UpstreamError } from '../../error';
// Shared resolver, cached for 6 hours. All PDS/handle lookups go through it.const didCache = new MemoryCache(6 * 60 * 60 * 1000);export const idResolver = new IdResolver({ didCache });
// Small pool of agents keyed by PDS endpoint so we don't rebuild an XrpcClient// for every call.const agentPool = new Map<string, AtpAgent>();function agent(pds: string): AtpAgent { const key = normalisePds(pds); let a = agentPool.get(key); if (!a) { a = new AtpAgent({ service: key }); agentPool.set(key, a); } return a;}
function normalisePds(pds: string): string { return pds.endsWith('/') ? pds.slice(0, -1) : pds;}
// --- Identity ---
// Resolves a DID to its current atproto identity (handle + pds + signingKey).// Throws if the DID can't be resolved or has no atproto PDS service entry.export async function resolveAtprotoData(did: string) { try { return await idResolver.did.resolveAtprotoData(did); } catch (err) { throw new UpstreamError( `could not resolve DID ${did}: ${(err as Error).message}` ); }}
// Resolve a handle to a DID (DNS TXT or /.well-known fallback).export async function resolveHandleToDid(handle: string): Promise<string> { const did = await idResolver.handle.resolve(handle); if (!did) throw new UpstreamError(`could not resolve handle ${handle}`); return did;}
// --- Sync endpoints (typed via AtpAgent) ---
export async function* iterateRepos( pds: string, limit = 1000): AsyncGenerator<{ did: string; head: string; rev: string; active?: boolean; status?: string;}> { const a = agent(pds); let cursor: string | undefined; do { const res = await a.com.atproto.sync.listRepos({ limit, cursor }); for (const r of res.data.repos) yield r; cursor = res.data.cursor; } while (cursor);}
export async function getRepoStatus(pds: string, did: string) { const res = await agent(pds).com.atproto.sync.getRepoStatus({ did }); return res.data;}
export async function* iterateBlobs( pds: string, did: string, since?: string, limit = 1000): AsyncGenerator<string> { const a = agent(pds); let cursor: string | undefined; do { const res = await a.com.atproto.sync.listBlobs({ did, since, limit, cursor }); for (const c of res.data.cids) yield c; cursor = res.data.cursor; } while (cursor);}
export async function describeRepo(pds: string, repo: string) { const res = await agent(pds).com.atproto.repo.describeRepo({ repo }); return res.data;}
// --- Streaming blob/repo fetches ---
// Returns the raw Response so callers can pipe `.body` straight into storage.// Throws on non-2xx; surfaces `Retry-After` seconds when available.export async function streamRepo( pds: string, did: string, since?: string): Promise<Response> { const url = new URL('/xrpc/com.atproto.sync.getRepo', normalisePds(pds)); url.searchParams.set('did', did); if (since) url.searchParams.set('since', since); return expectOk(await fetch(url), url);}
export async function streamBlob( pds: string, did: string, cid: string): Promise<Response> { const url = new URL('/xrpc/com.atproto.sync.getBlob', normalisePds(pds)); url.searchParams.set('did', did); url.searchParams.set('cid', cid); return expectOk(await fetch(url), url);}
export class StreamXrpcError extends Error { constructor( public readonly endpoint: string, public readonly status: number, public readonly body: string, public readonly retryAfterSeconds: number | null ) { super(`XRPC ${endpoint} -> ${status}: ${body.slice(0, 200)}`); }}
async function expectOk(res: Response, url: URL): Promise<Response> { if (res.ok) return res; const body = await res.text().catch(() => ''); const ra = res.headers.get('retry-after'); throw new StreamXrpcError( url.pathname, res.status, body, ra ? Number.parseInt(ra, 10) || null : null );}