Something went wrong. Try again.
Periodic backups of ATProto repos and blobs safekeeper.dummy.cafe
atproto backup
Something went wrong. Try again.
7.8 kB · 299 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300// Orchestrates one backup cycle: enumerate targets, then for each account// snapshot the repo + sync new blobs, respecting account status.
import { and, eq } from 'drizzle-orm';import db from '../database';import { account, backup_run, blob_backup, repo_backup} from '../database/schema';import type { AccountStatus } from '../database/schema';import { describeRepo, getRepoStatus, iterateBlobs, streamBlob, streamRepo, StreamXrpcError} from '../atproto/client';import { storage } from '../storage';import { collectTargets, type BackupTarget } from './targets';import { pruneAccountBackups } from './retention';
// Account is "healthy" — i.e. we should keep backing it up and pruning old// snapshots. Non-active statuses freeze the archive in place.export function isHealthy(status: AccountStatus): boolean { return status === 'active';}
// ISO-ish timestamp safe for filenames (no colons).function fsTimestamp(d: Date): string { return d .toISOString() .replace(/\.\d{3}Z$/, 'Z') .replace(/:/g, '-');}
interface RunError { did: string; stage: string; message: string;}
export interface RunSummary { accountsConsidered: number; accountsBackedUp: number; repoSnapshotsTaken: number; blobsFetched: number; bytesWritten: number; errors: RunError[];}
async function backupOneAccount( target: BackupTarget, errors: RunError[]): Promise<{ tookSnapshot: boolean; blobsFetched: number; bytesWritten: number;}> { const out = { tookSnapshot: false, blobsFetched: 0, bytesWritten: 0 }; const now = new Date(); const s = storage();
// --- Confirm status via getRepoStatus (authoritative for this cycle). --- let status: AccountStatus = target.initial_status; let currentRev = target.latest_rev ?? undefined; try { const st = await getRepoStatus(target.pds_endpoint, target.did); if (st.active === false) { const knownStatuses: AccountStatus[] = [ 'takendown', 'suspended', 'deleted', 'deactivated', 'desynchronized', 'throttled' ]; status = (knownStatuses as string[]).includes(st.status ?? '') ? (st.status as AccountStatus) : 'unknown'; } else { status = 'active'; } currentRev = st.rev ?? currentRev; } catch (err) { errors.push({ did: target.did, stage: 'getRepoStatus', message: (err as Error).message }); // Fall through with whatever we had from listRepos. }
await db .update(account) .set({ status, last_status_checked_at: now }) .where(eq(account.did, target.did));
// --- If not healthy, stop here. No new snapshots, no pruning. --- if (!isHealthy(status)) return out;
const existing = await db.query.account.findFirst({ where: eq(account.did, target.did) });
// --- Refresh handle every cycle. listRepos doesn't return it, and handles // are mutable (domain transfers, re-points), so we can't rely on the // stored value being current. describeRepo on the same PDS is cheap and // authoritative. On failure we leave the previous value in place. try { const desc = await describeRepo(target.pds_endpoint, target.did); if (desc.handle && desc.handle !== existing?.handle) { await db .update(account) .set({ handle: desc.handle }) .where(eq(account.did, target.did)); } } catch { // non-fatal; retry next cycle. }
const previousRev = existing?.last_archived_rev;
// --- Repo snapshot --- if (currentRev && currentRev === previousRev) { // No-op: the PDS reports the same rev we already archived. } else { try { const stampedRev = currentRev ?? fsTimestamp(now); const storagePath = `repos/${target.did}/${fsTimestamp(now)}_${stampedRev}.car`; const res = await streamRepo(target.pds_endpoint, target.did); if (!res.body) throw new Error('getRepo returned no body'); await s.put(storagePath, res.body); const stat = await s.stat(storagePath); const size = stat?.size ?? 0;
await db .insert(repo_backup) .values({ account_did: target.did, rev: stampedRev, storage_path: storagePath, size_bytes: size, taken_at: now }) .onConflictDoNothing();
await db .update(account) .set({ last_archived_rev: stampedRev, last_repo_backup_at: now }) .where(eq(account.did, target.did));
out.tookSnapshot = true; out.bytesWritten += size; } catch (err) { errors.push({ did: target.did, stage: 'getRepo', message: err instanceof StreamXrpcError ? `${err.status} ${err.body.slice(0, 120)}` : (err as Error).message }); } }
// --- Blob sync (incremental by CID) --- try { // Load the set of CIDs we already have for this account. Blob backups // are forever-kept, so anything in the table is durable. const existingRows = await db .select({ cid: blob_backup.cid }) .from(blob_backup) .where(eq(blob_backup.account_did, target.did)); const existing = new Set(existingRows.map((r) => r.cid));
for await (const cid of iterateBlobs(target.pds_endpoint, target.did)) { if (existing.has(cid)) continue; try { const res = await streamBlob(target.pds_endpoint, target.did, cid); if (!res.body) throw new Error('getBlob returned no body'); const storagePath = `blobs/${target.did}/${cid}`; await s.put(storagePath, res.body); const stat = await s.stat(storagePath); const size = stat?.size ?? 0; await db .insert(blob_backup) .values({ account_did: target.did, cid, storage_path: storagePath, size_bytes: size }) .onConflictDoNothing(); existing.add(cid); out.blobsFetched += 1; out.bytesWritten += size; } catch (err) { errors.push({ did: target.did, stage: 'getBlob', message: `${cid}: ${(err as Error).message}` }); } } await db .update(account) .set({ last_blob_sync_at: now }) .where(eq(account.did, target.did)); } catch (err) { errors.push({ did: target.did, stage: 'listBlobs', message: (err as Error).message }); }
// --- Retention pruning (only for healthy accounts). --- try { await pruneAccountBackups(target.did, now); } catch (err) { errors.push({ did: target.did, stage: 'prune', message: (err as Error).message }); }
return out;}
export async function runBackupCycle(): Promise<RunSummary> { const startedAt = new Date(); const [inserted] = await db .insert(backup_run) .values({ started_at: startedAt }) .returning({ id: backup_run.id }); const runId = inserted!.id;
const { targets, pdsErrors } = await collectTargets(); const errors: RunError[] = pdsErrors.map((e) => ({ did: '(pds)', stage: 'listRepos', message: `${e.hostname}: ${e.message}` }));
let accountsBackedUp = 0; let repoSnapshotsTaken = 0; let blobsFetched = 0; let bytesWritten = 0;
for (const t of targets) { try { const res = await backupOneAccount(t, errors); if (res.tookSnapshot || res.blobsFetched > 0) accountsBackedUp += 1; if (res.tookSnapshot) repoSnapshotsTaken += 1; blobsFetched += res.blobsFetched; bytesWritten += res.bytesWritten; } catch (err) { errors.push({ did: t.did, stage: 'account', message: (err as Error).message }); } }
const summary: RunSummary = { accountsConsidered: targets.length, accountsBackedUp, repoSnapshotsTaken, blobsFetched, bytesWritten, errors };
await db .update(backup_run) .set({ finished_at: new Date(), accounts_considered: summary.accountsConsidered, accounts_backed_up: summary.accountsBackedUp, repo_snapshots_taken: summary.repoSnapshotsTaken, blobs_fetched: summary.blobsFetched, bytes_written: summary.bytesWritten, errors: errors.length ? JSON.stringify(errors) : null }) .where(eq(backup_run.id, runId));
return summary;}