Something went wrong. Try again.
Periodic backups of ATProto repos and blobs safekeeper.dummy.cafe
atproto backup
Something went wrong. Try again.
4.7 kB · 162 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163// Figures out the set of (did, currentPds) pairs we should attempt to back// up on a given cycle. This is the union of://// (1) every account on every enabled `pds`, discovered via listRepos// (2) every DID in `tracked_account`//// An account in both sets gets `discovery_source='both'`.
import { eq, sql } from 'drizzle-orm';import db from '../database';import { account, blocked_account, pds, tracked_account } from '../database/schema';import type { AccountStatus } from '../database/schema';import { iterateRepos, resolveAtprotoData, StreamXrpcError} from '../atproto/client';
export interface BackupTarget { did: string; pds_endpoint: string; handle: string | null; // The PDS-reported status from listRepos, when known. May be overridden by // per-target getRepoStatus later. initial_status: AccountStatus; // The rev the PDS just told us about — lets the runner skip accounts whose // repo hasn't advanced since the last archived rev. latest_rev: string | null; source: 'pds' | 'tracked' | 'both';}
// Translate the optional `active`/`status` shape that listRepos returns into// our internal enum.function mapStatus(active: boolean | undefined, status: string | undefined): AccountStatus { if (active === false) { switch (status) { case 'takendown': case 'suspended': case 'deleted': case 'deactivated': case 'desynchronized': case 'throttled': return status; default: return 'unknown'; } } return 'active';}
export async function collectTargets(): Promise<{ targets: BackupTarget[]; pdsErrors: { hostname: string; message: string }[];}> { const targets = new Map<string, BackupTarget>(); const pdsErrors: { hostname: string; message: string }[] = [];
// Any DID here has been explicitly wiped by an admin — we drop it from // targets before any upsert so the scheduler doesn't resurrect it. const blockedRows = await db .select({ did: blocked_account.did }) .from(blocked_account); const blocked = new Set(blockedRows.map((r) => r.did));
// --- (1) accounts on enabled PDSes --- const enabledPdses = await db.query.pds.findMany({ where: eq(pds.enabled, true) });
for (const row of enabledPdses) { const pdsEndpoint = row.hostname.includes('://') ? row.hostname : `https://${row.hostname}`; let listError: string | null = null; try { for await (const r of iterateRepos(pdsEndpoint)) { if (blocked.has(r.did)) continue; targets.set(r.did, { did: r.did, pds_endpoint: pdsEndpoint, handle: null, initial_status: mapStatus(r.active, r.status), latest_rev: r.rev ?? null, source: 'pds' }); } } catch (err) { listError = err instanceof StreamXrpcError ? `${err.status} ${err.body.slice(0, 120)}` : (err as Error).message; pdsErrors.push({ hostname: row.hostname, message: listError }); } await db .update(pds) .set({ last_listed_at: new Date(), last_list_error: listError }) .where(eq(pds.hostname, row.hostname)); }
// --- (2) explicitly tracked DIDs --- const tracked = await db.select({ did: tracked_account.did }).from(tracked_account); for (const { did } of tracked) { if (blocked.has(did)) continue; const existing = targets.get(did); if (existing) { existing.source = 'both'; continue; } // Resolve the tracked DID's current PDS (may differ from anything we've // listed). A failure here is recorded as an error but doesn't stop the // whole cycle. try { const data = await resolveAtprotoData(did); targets.set(did, { did, pds_endpoint: data.pds, handle: data.handle, initial_status: 'unknown', latest_rev: null, source: 'tracked' }); } catch (err) { pdsErrors.push({ hostname: `(tracked) ${did}`, message: (err as Error).message }); } }
// --- Upsert `account` rows so the dashboard reflects the set of known // accounts even before any backup work runs. We don't change status // for non-active-derived sources here — that happens in the runner // once we've confirmed via getRepoStatus. const now = new Date(); for (const t of targets.values()) { await db .insert(account) .values({ did: t.did, handle: t.handle, pds_endpoint: t.pds_endpoint, status: t.initial_status, last_status_checked_at: now, discovery_source: t.source }) .onConflictDoUpdate({ target: account.did, set: { handle: sql`coalesce(excluded.handle, ${account.handle})`, pds_endpoint: t.pds_endpoint, status: t.initial_status, last_status_checked_at: now, discovery_source: t.source } }); }
return { targets: Array.from(targets.values()), pdsErrors };}