Something went wrong. Try again.
Periodic backups of ATProto repos and blobs safekeeper.dummy.cafe
atproto backup
Something went wrong. Try again.
6.1 kB · 179 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180/* Schema rules (carried over from lsky): - References to other tables are named `thing_id` / `thing_did`, and `thing` is the relation. - Foreign keys use `foreignKey()` rather than `.references()` — easier to audit when primary keys are composite.*/
import { sql } from 'drizzle-orm';import { foreignKey, integer, primaryKey, sqliteTable, text, uniqueIndex} from 'drizzle-orm/sqlite-core';
// --- Types ---
const timestamps = () => ({ updated_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull() .$onUpdate(() => new Date()), created_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull()});
const did = () => text({ length: 2048 });
// Mirrors com.atproto.sync status values; `active` is our additional in-band// value meaning the account is healthy.export const ACCOUNT_STATUSES = [ 'active', 'deactivated', 'deleted', 'suspended', 'takendown', 'desynchronized', 'throttled', 'unknown'] as const;export type AccountStatus = (typeof ACCOUNT_STATUSES)[number];
// --- Tables ---
// PDSes we back up wholesale — every account listed by `com.atproto.sync.listRepos`// on these hosts is pulled in as a backup target for each cycle.export const pds = sqliteTable('pds', { hostname: text().primaryKey(), // e.g. "bsky.social" enabled: integer({ mode: 'boolean' }).default(true).notNull(), last_listed_at: integer({ mode: 'timestamp' }), last_list_error: text(), ...timestamps()});
// Explicit per-DID targets. Kept separate from `account` so that an operator// can add a DID they want to follow regardless of which PDS it lives on// (including DIDs that have migrated away from a PDS in `pds`).export const tracked_account = sqliteTable('tracked_account', { did: did().primaryKey(), note: text(), // optional free-text label (operator's reason for tracking) ...timestamps()});
// Cached view of every account we know about, whether discovered via a PDS// or added as a tracked_account. This is where we record status and the// last successful rev so the scheduler can make decisions without re-querying.//// `pds_endpoint` is the account's *current* PDS as most recently observed,// which may differ from any entry in `pds` if the account migrated.export const account = sqliteTable('account', { did: did().primaryKey(), handle: text(), pds_endpoint: text(), // e.g. "https://bsky.social"
status: text({ enum: ACCOUNT_STATUSES }).default('unknown').notNull(), last_status_checked_at: integer({ mode: 'timestamp' }),
// Last TID rev we successfully archived a repo for. Used as the `since` // baseline for incremental backups and to detect no-op cycles. last_archived_rev: text(), last_repo_backup_at: integer({ mode: 'timestamp' }), last_blob_sync_at: integer({ mode: 'timestamp' }),
// Why we're backing this account up: discovered from a PDS, explicitly // tracked, or both. discovery_source: text({ enum: ['pds', 'tracked', 'both'] }) .default('pds') .notNull(),
...timestamps()});
// One row per repo CAR snapshot, *ever*. The retention pruner removes the// underlying file from storage but keeps the row, stamping `pruned_at`.// This preserves a durable record of every snapshot that was once taken.//// Queries for currently-available backups MUST filter `pruned_at IS NULL`.//// Retention policy: keep every non-pruned row from the last 7 days; before// that, keep 2 per calendar week. The pruner only operates on accounts// with a healthy status.export const repo_backup = sqliteTable( 'repo_backup', { id: integer().primaryKey({ autoIncrement: true }), account_did: did().notNull(), rev: text().notNull(), storage_path: text().notNull(), size_bytes: integer().notNull(), taken_at: integer({ mode: 'timestamp' }).notNull(),
// Null while the CAR is present on disk. Set when the pruner deletes // the file. Rows are never deleted. pruned_at: integer({ mode: 'timestamp' }),
created_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull() }, (t) => [ foreignKey({ columns: [t.account_did], foreignColumns: [account.did] }), uniqueIndex('repo_backup_did_rev_uq').on(t.account_did, t.rev) ]);
// Blobs are content-addressed, so one row per (did, cid). We never re-fetch// the same CID for the same account. Blobs are never pruned.export const blob_backup = sqliteTable( 'blob_backup', { account_did: did().notNull(), cid: text().notNull(), storage_path: text().notNull(), size_bytes: integer().notNull(), created_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull() }, (t) => [ foreignKey({ columns: [t.account_did], foreignColumns: [account.did] }), primaryKey({ columns: [t.account_did, t.cid] }) ]);
// DIDs that an admin has "wiped" via the panel. The wipe flow deletes every// row associated with the DID from account/tracked_account/repo_backup/// blob_backup, so we need this separate table to remember that the DID was// deliberately expunged. `collectTargets` filters it out so the scheduler// won't re-archive an account just because it still appears in a PDS's// listRepos response.//// To un-block, delete the row (no UI for this yet).export const blocked_account = sqliteTable('blocked_account', { did: did().primaryKey(), reason: text(), created_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull()});
// Scheduler run log — populated by each tick for the dashboard / debugging.export const backup_run = sqliteTable('backup_run', { id: integer().primaryKey({ autoIncrement: true }), started_at: integer({ mode: 'timestamp' }).notNull(), finished_at: integer({ mode: 'timestamp' }), accounts_considered: integer().default(0).notNull(), accounts_backed_up: integer().default(0).notNull(), repo_snapshots_taken: integer().default(0).notNull(), blobs_fetched: integer().default(0).notNull(), bytes_written: integer().default(0).notNull(), errors: text(), // JSON array of { did, stage, message } created_at: integer({ mode: 'timestamp' }) .default(sql`(unixepoch())`) .notNull()});