Something went wrong. Try again.
Chris’s PDS implementation
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544//! Ordered schema migrations and compatibility checks.
use std::collections::BTreeSet;
use rusqlite::Connection;use serde_json::Value;
use crate::{ domain::blob::referenced_blob_cids, error::{Error, Result},};
const CURRENT_SCHEMA_VERSION: i64 = 9;const BUSY_TIMEOUT_MS: i64 = 5_000;const WAL_AUTOCHECKPOINT_PAGES: i64 = 1_000;const JOURNAL_SIZE_LIMIT_BYTES: i64 = 64 * 1024 * 1024;
const CONNECTION_PRAGMAS: &str = "PRAGMA foreign_keys = ON;PRAGMA journal_mode = WAL;PRAGMA synchronous = FULL;PRAGMA busy_timeout = 5000;PRAGMA trusted_schema = OFF;PRAGMA wal_autocheckpoint = 1000;PRAGMA journal_size_limit = 67108864;";
const INITIAL_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS accounts ( did TEXT PRIMARY KEY, handle TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, created_at INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS records ( did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT NOT NULL, record_json TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (did, collection, rkey));
CREATE INDEX IF NOT EXISTS records_by_collection ON records(did, collection, rkey);
CREATE TABLE IF NOT EXISTS spaces ( uri TEXT PRIMARY KEY, authority TEXT NOT NULL, space_type TEXT NOT NULL, skey TEXT NOT NULL, policy_json TEXT NOT NULL, app_access_json TEXT NOT NULL, deleted_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, UNIQUE (authority, space_type, skey));
CREATE TABLE IF NOT EXISTS space_members ( space_uri TEXT NOT NULL REFERENCES spaces(uri) ON DELETE CASCADE, did TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (space_uri, did));
CREATE TABLE IF NOT EXISTS space_records ( space_uri TEXT NOT NULL, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT NOT NULL, record_json TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (space_uri, did, collection, rkey));
CREATE INDEX IF NOT EXISTS space_records_by_repo ON space_records(space_uri, did, collection, rkey);
CREATE TABLE IF NOT EXISTS space_repo_ops ( id INTEGER PRIMARY KEY AUTOINCREMENT, space_uri TEXT NOT NULL, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, rev TEXT NOT NULL, collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT, prev TEXT, record_json TEXT, created_at INTEGER NOT NULL);
CREATE INDEX IF NOT EXISTS space_ops_by_repo ON space_repo_ops(space_uri, did, rev, id);
CREATE TABLE IF NOT EXISTS blobs ( did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, cid TEXT NOT NULL, media_type TEXT NOT NULL, size INTEGER NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (did, cid));
CREATE TABLE IF NOT EXISTS oauth_codes ( code_hash TEXT PRIMARY KEY, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, client_id TEXT NOT NULL, redirect_uri TEXT NOT NULL, code_challenge TEXT NOT NULL, scope TEXT NOT NULL, expires_at INTEGER NOT NULL, used_at INTEGER);
CREATE TABLE IF NOT EXISTS access_tokens ( token_hash TEXT PRIMARY KEY, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, client_id TEXT NOT NULL, scope TEXT NOT NULL, expires_at INTEGER NOT NULL, created_at INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS space_auth_replays ( kind TEXT NOT NULL, issuer TEXT NOT NULL, jti TEXT NOT NULL, expires_at INTEGER NOT NULL, PRIMARY KEY (kind, issuer, jti));
CREATE TABLE IF NOT EXISTS space_writers ( space_uri TEXT NOT NULL REFERENCES spaces(uri) ON DELETE CASCADE, did TEXT NOT NULL, rev TEXT NOT NULL, hash BLOB NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (space_uri, did));
CREATE TABLE IF NOT EXISTS space_notify_registrations ( space_uri TEXT NOT NULL REFERENCES spaces(uri) ON DELETE CASCADE, service TEXT NOT NULL, endpoint TEXT NOT NULL, expires_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (space_uri, service));";
const VALIDATE_INITIAL_SCHEMA: &str = "SELECT did, handle, password_hash, created_at FROM accounts LIMIT 0;SELECT did, collection, rkey, cid, record_json, created_at, updated_at FROM records LIMIT 0;SELECT uri, authority, space_type, skey, policy_json, app_access_json, deleted_at, created_at, updated_at FROM spaces LIMIT 0;SELECT space_uri, did, created_at FROM space_members LIMIT 0;SELECT space_uri, did, collection, rkey, cid, record_json, created_at, updated_at FROM space_records LIMIT 0;SELECT id, space_uri, did, rev, collection, rkey, cid, prev, record_json, created_at FROM space_repo_ops LIMIT 0;SELECT did, cid, media_type, size, created_at FROM blobs LIMIT 0;SELECT code_hash, did, client_id, redirect_uri, code_challenge, scope, expires_at, used_at FROM oauth_codes LIMIT 0;SELECT token_hash, did, client_id, scope, expires_at, created_at FROM access_tokens LIMIT 0;SELECT kind, issuer, jti, expires_at FROM space_auth_replays LIMIT 0;SELECT space_uri, did, rev, hash, updated_at FROM space_writers LIMIT 0;SELECT space_uri, service, endpoint, expires_at, updated_at FROM space_notify_registrations LIMIT 0;";
const REPOSITORY_SCHEMA: &str = "CREATE TABLE repo_blocks ( cid TEXT PRIMARY KEY, bytes BLOB NOT NULL, created_at INTEGER NOT NULL);
CREATE TABLE repo_heads ( did TEXT PRIMARY KEY REFERENCES accounts(did) ON DELETE CASCADE, root_cid TEXT NOT NULL REFERENCES repo_blocks(cid), rev TEXT NOT NULL, updated_at INTEGER NOT NULL);
CREATE TABLE repo_ops ( id INTEGER PRIMARY KEY AUTOINCREMENT, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, rev TEXT NOT NULL, action TEXT NOT NULL CHECK (action IN ('create', 'update', 'delete')), collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT, prev TEXT, created_at INTEGER NOT NULL);
CREATE INDEX repo_ops_by_repo ON repo_ops(did, id);
CREATE TABLE repo_outbox ( seq INTEGER PRIMARY KEY AUTOINCREMENT, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, rev TEXT, root_cid TEXT, event_type TEXT NOT NULL CHECK (event_type IN ('repo.commit', 'identity', 'account')), payload BLOB NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at INTEGER NOT NULL, delivered_at INTEGER, created_at INTEGER NOT NULL, CHECK ( (event_type = 'repo.commit' AND rev IS NOT NULL AND root_cid IS NOT NULL) OR (event_type IN ('identity', 'account') AND rev IS NULL AND root_cid IS NULL) ));
CREATE INDEX repo_outbox_pending ON repo_outbox(delivered_at, next_attempt_at, seq);";
const VALIDATE_REPOSITORY_SCHEMA: &str = "SELECT cid, bytes, created_at FROM repo_blocks LIMIT 0;SELECT did, root_cid, rev, updated_at FROM repo_heads LIMIT 0;SELECT id, did, rev, action, collection, rkey, cid, prev, created_at FROM repo_ops LIMIT 0;SELECT seq, did, rev, root_cid, event_type, payload, attempts, next_attempt_at, delivered_at, created_at FROM repo_outbox LIMIT 0;";
const COMMIT_BLOCK_SCHEMA: &str = "CREATE TABLE repo_commit_blocks ( seq INTEGER NOT NULL REFERENCES repo_outbox(seq) ON DELETE CASCADE, cid TEXT NOT NULL REFERENCES repo_blocks(cid), position INTEGER NOT NULL, PRIMARY KEY (seq, cid), UNIQUE (seq, position));";
const VALIDATE_COMMIT_BLOCK_SCHEMA: &str = "SELECT seq, cid, position FROM repo_commit_blocks LIMIT 0;";
const RECORD_BLOB_SCHEMA: &str = "CREATE TABLE record_blobs ( did TEXT NOT NULL, collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT NOT NULL, PRIMARY KEY (did, collection, rkey, cid), FOREIGN KEY (did, collection, rkey) REFERENCES records(did, collection, rkey) ON DELETE CASCADE);
CREATE INDEX record_blobs_by_cid ON record_blobs(did, cid, collection, rkey);";
const VALIDATE_RECORD_BLOB_SCHEMA: &str = "SELECT did, collection, rkey, cid FROM record_blobs LIMIT 0;";
const BLOB_LIFECYCLE_SCHEMA: &str = "ALTER TABLE blobs ADD COLUMN delete_started_at INTEGER;
CREATE TABLE space_record_blobs ( space_uri TEXT NOT NULL, did TEXT NOT NULL, collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT NOT NULL, PRIMARY KEY (space_uri, did, collection, rkey, cid), FOREIGN KEY (space_uri, did, collection, rkey) REFERENCES space_records(space_uri, did, collection, rkey) ON DELETE CASCADE);
CREATE INDEX space_record_blobs_by_cid ON space_record_blobs(did, cid, space_uri, collection, rkey);";
const VALIDATE_BLOB_LIFECYCLE_SCHEMA: &str = "SELECT did, cid, media_type, size, created_at, delete_started_at FROM blobs LIMIT 0;SELECT space_uri, did, collection, rkey, cid FROM space_record_blobs LIMIT 0;";
const BLOB_PUBLICATION_SCHEMA: &str = "ALTER TABLE blobs ADD COLUMN available_at INTEGER;UPDATE blobs SET available_at = created_at;";
const VALIDATE_BLOB_PUBLICATION_SCHEMA: &str = "SELECT did, cid, media_type, size, created_at, delete_started_at, available_atFROM blobs LIMIT 0;";
const STREAM_OUTBOX_SCHEMA: &str = "CREATE TABLE repo_outbox_v7 ( seq INTEGER PRIMARY KEY AUTOINCREMENT, did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, rev TEXT, root_cid TEXT, event_type TEXT NOT NULL CHECK (event_type IN ('repo.commit', 'identity', 'account')), payload BLOB NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at INTEGER NOT NULL, delivered_at INTEGER, created_at INTEGER NOT NULL, CHECK ( (event_type = 'repo.commit' AND rev IS NOT NULL AND root_cid IS NOT NULL) OR (event_type IN ('identity', 'account') AND rev IS NULL AND root_cid IS NULL) ));
INSERT INTO repo_outbox_v7 ( seq, did, rev, root_cid, event_type, payload, attempts, next_attempt_at, delivered_at, created_at)SELECT seq, did, rev, root_cid, event_type, payload, attempts, next_attempt_at, delivered_at, created_atFROM repo_outbox;
DELETE FROM sqlite_sequence WHERE name = 'repo_outbox_v7';INSERT INTO sqlite_sequence (name, seq)SELECT 'repo_outbox_v7', seq FROM sqlite_sequence WHERE name = 'repo_outbox';
CREATE TABLE repo_commit_blocks_v7 ( seq INTEGER NOT NULL REFERENCES repo_outbox_v7(seq) ON DELETE CASCADE, cid TEXT NOT NULL REFERENCES repo_blocks(cid), position INTEGER NOT NULL, PRIMARY KEY (seq, cid), UNIQUE (seq, position));
INSERT INTO repo_commit_blocks_v7 (seq, cid, position)SELECT seq, cid, position FROM repo_commit_blocks;
DROP TABLE repo_commit_blocks;DROP TABLE repo_outbox;ALTER TABLE repo_outbox_v7 RENAME TO repo_outbox;ALTER TABLE repo_commit_blocks_v7 RENAME TO repo_commit_blocks;
CREATE INDEX repo_outbox_pending ON repo_outbox(delivered_at, next_attempt_at, seq);";
const ACCOUNT_STATUS_SCHEMA: &str = "ALTER TABLE accounts ADD COLUMN status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'deactivated', 'suspended', 'takendown', 'deleted'));";
const VALIDATE_ACCOUNT_STATUS_SCHEMA: &str = "SELECT status FROM accounts LIMIT 0;";
const OAUTH_EXPIRY_SCHEMA: &str = "CREATE INDEX oauth_codes_by_expiry ON oauth_codes(expires_at);CREATE INDEX access_tokens_by_expiry ON access_tokens(expires_at);";
pub(super) fn apply(connection: &mut Connection, initialize: bool) -> Result<()> { connection.execute_batch(CONNECTION_PRAGMAS)?; verify_connection_policy(connection)?; let version: i64 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?; if !(0..=CURRENT_SCHEMA_VERSION).contains(&version) { return Err(Error::Configuration(format!( "database schema version {version} is not supported; this binary supports {CURRENT_SCHEMA_VERSION}" ))); } if version == 0 && !initialize { return Err(Error::Configuration( "existing database is not initialized as CPDS storage".to_owned(), )); } if version < CURRENT_SCHEMA_VERSION { let transaction = connection.transaction()?; migrate(&transaction, version)?; validate_schema(&transaction)?; transaction.pragma_update(None, "user_version", CURRENT_SCHEMA_VERSION)?; transaction.commit()?; return Ok(()); } validate_schema(connection)}
pub(super) fn check(connection: &Connection) -> Result<()> { verify_connection_policy(connection)?; let version: i64 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?; if version != CURRENT_SCHEMA_VERSION { return Err(Error::Configuration(format!( "database schema version {version} is not supported; this binary supports {CURRENT_SCHEMA_VERSION}" ))); } connection.query_row_and_then("SELECT 1", [], |_| Ok(()))}
fn verify_connection_policy(connection: &Connection) -> Result<()> { verify_pragma(connection, "foreign_keys", 1)?; verify_pragma(connection, "synchronous", 2)?; verify_pragma(connection, "busy_timeout", BUSY_TIMEOUT_MS)?; verify_pragma(connection, "trusted_schema", 0)?; verify_pragma(connection, "wal_autocheckpoint", WAL_AUTOCHECKPOINT_PAGES)?; verify_pragma(connection, "journal_size_limit", JOURNAL_SIZE_LIMIT_BYTES)?;
let journal_mode: String = connection.pragma_query_value(None, "journal_mode", |row| row.get(0))?; if !matches!(journal_mode.as_str(), "wal" | "memory") { return Err(Error::Configuration(format!( "SQLite journal_mode is {journal_mode}, expected WAL" ))); } Ok(())}
fn verify_pragma(connection: &Connection, name: &str, expected: i64) -> Result<()> { let actual: i64 = connection.pragma_query_value(None, name, |row| row.get(0))?; if actual != expected { return Err(Error::Configuration(format!( "SQLite {name} is {actual}, expected {expected}" ))); } Ok(())}
fn verify_integrity(connection: &Connection) -> Result<()> { let quick_check: String = connection.query_row("PRAGMA quick_check(1)", [], |row| row.get(0))?; if quick_check != "ok" { return Err(Error::Configuration(format!( "SQLite quick_check failed: {quick_check}" ))); }
let mut statement = connection.prepare("PRAGMA foreign_key_check")?; if statement.exists([])? { return Err(Error::Configuration( "SQLite foreign_key_check found an invalid reference".to_owned(), )); } Ok(())}
fn migrate(transaction: &rusqlite::Transaction<'_>, version: i64) -> Result<()> { for (threshold, schema, backfill) in [ (1, INITIAL_SCHEMA, None), (2, REPOSITORY_SCHEMA, None), (3, COMMIT_BLOCK_SCHEMA, None), (4, RECORD_BLOB_SCHEMA, Some(&PUBLIC_BLOBS)), (5, BLOB_LIFECYCLE_SCHEMA, Some(&SPACE_BLOBS)), (6, BLOB_PUBLICATION_SCHEMA, None), (8, ACCOUNT_STATUS_SCHEMA, None), // Earlier direct upgrades could skip the v7 outbox rebuild. Repair them // as well, retaining rows, block memberships, and the sequence counter. (9, STREAM_OUTBOX_SCHEMA, None), (9, OAUTH_EXPIRY_SCHEMA, None), ] { if version < threshold { transaction.execute_batch(schema)?; if let Some(index) = backfill { backfill_blob_index(transaction, index)?; } } } Ok(())}
fn validate_schema(connection: &Connection) -> Result<()> { for schema in [ VALIDATE_INITIAL_SCHEMA, VALIDATE_REPOSITORY_SCHEMA, VALIDATE_COMMIT_BLOCK_SCHEMA, VALIDATE_RECORD_BLOB_SCHEMA, VALIDATE_BLOB_LIFECYCLE_SCHEMA, VALIDATE_BLOB_PUBLICATION_SCHEMA, VALIDATE_ACCOUNT_STATUS_SCHEMA, ] { connection.execute_batch(schema)?; } verify_integrity(connection)?; verify_record_blob_indexes(connection)}
struct BlobIndex { records: &'static str, entries: &'static str, insert: &'static str, mismatch: &'static str,}
const PUBLIC_BLOBS: BlobIndex = BlobIndex { records: "SELECT did, collection, rkey, record_json FROM records ORDER BY did, collection, rkey", entries: "SELECT did, collection, rkey, cid FROM record_blobs ORDER BY did, collection, rkey, cid", insert: "INSERT INTO record_blobs (did, collection, rkey, cid) VALUES (?1, ?2, ?3, ?4)", mismatch: "record blob reference index does not match stored records",};const SPACE_BLOBS: BlobIndex = BlobIndex { records: "SELECT space_uri, did, collection, rkey, record_json FROM space_records ORDER BY space_uri, did, collection, rkey", entries: "SELECT space_uri, did, collection, rkey, cid FROM space_record_blobs ORDER BY space_uri, did, collection, rkey, cid", insert: "INSERT INTO space_record_blobs (space_uri, did, collection, rkey, cid) VALUES (?1, ?2, ?3, ?4, ?5)", mismatch: "space record blob reference index does not match stored records",};
fn backfill_blob_index(connection: &Connection, index: &BlobIndex) -> Result<()> { for reference in expected_blob_index(connection, index)? { connection.execute(index.insert, rusqlite::params_from_iter(reference))?; } Ok(())}
fn verify_record_blob_indexes(connection: &Connection) -> Result<()> { for index in [&PUBLIC_BLOBS, &SPACE_BLOBS] { let expected = expected_blob_index(connection, index)?; let actual = index_rows(connection, index.entries)?.into_iter().collect(); if expected != actual { return Err(Error::Configuration(index.mismatch.to_owned())); } } Ok(())}
fn expected_blob_index( connection: &Connection, index: &BlobIndex,) -> Result<BTreeSet<Vec<String>>> { let mut references = BTreeSet::new(); for mut key in index_rows(connection, index.records)? { let json = key .pop() .expect("record queries select JSON after the composite key"); let value: Value = serde_json::from_str(&json)?; for cid in referenced_blob_cids(&value) { let mut reference = key.clone(); reference.push(cid); references.insert(reference); } } Ok(references)}
fn index_rows(connection: &Connection, sql: &str) -> Result<Vec<Vec<String>>> { super::query(connection, sql, [], |row| { (0..row.as_ref().column_count()) .map(|column| Ok(row.get(column)?)) .collect() })}
#[cfg(test)]mod tests;