Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313use std::collections::{BTreeSet, HashMap};
use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339};use crate::error::AppError;use crate::lua::tid::generate_tid;use crate::spaces::lthash::{LtHashState, record_element};use crate::spaces::{commit, db};
type RecordVersion = (String, String, String, String, String);
#[derive(Debug, Default, Clone, PartialEq, Eq)]pub struct BackfillReport { pub records_scanned: usize, pub records_updated: usize, pub records_unencodable: usize, pub oplog_rows_remapped: usize, pub oplog_rows_unresolved: usize, pub repos_rebuilt: usize,}
impl BackfillReport { pub fn is_noop(&self) -> bool { self.records_updated == 0 && self.oplog_rows_remapped == 0 }}
type RecordRow = (String, String, String, String, String, String, String);
const BACKFILL_MARKER_KEY: &str = "space_cid_backfill_completed_at";
async fn read_marker( pool: &sqlx::AnyPool, backend: DatabaseBackend,) -> Result<Option<String>, AppError> { let sql = adapt_sql( "SELECT value FROM happyview_instance_settings WHERE key = ?", backend, ); let row: Option<(String,)> = crate::db::query_as(&sql) .bind(BACKFILL_MARKER_KEY) .fetch_optional(pool) .await .map_err(|e| AppError::Internal(format!("failed to read backfill marker: {e}")))?; Ok(row.map(|r| r.0))}
// Run the repair only if it has not already completed against this database.pub async fn run_if_needed( pool: &sqlx::AnyPool, backend: DatabaseBackend,) -> Result<Option<BackfillReport>, AppError> { if read_marker(pool, backend).await?.is_some() { return Ok(None); } Ok(Some(run(pool, backend, false).await?))}
pub async fn run( pool: &sqlx::AnyPool, backend: DatabaseBackend, dry_run: bool,) -> Result<BackfillReport, AppError> { let mut report = BackfillReport::default();
let mut tx = pool .begin() .await .map_err(|e| AppError::Internal(format!("failed to begin transaction: {e}")))?;
// --------------------------------------------------------------------- // 1. Recompute record CIDs // --------------------------------------------------------------------- let sql = adapt_sql( "SELECT uri, space_id, author_did, collection, rkey, record, cid FROM happyview_space_records ORDER BY uri", backend, ); let rows: Vec<RecordRow> = crate::db::query_as(&sql) .fetch_all(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to load space records: {e}")))?;
// (space, author, collection, rkey, old_cid) -> new_cid let mut remap: HashMap<RecordVersion, String> = HashMap::new(); // (space_id, author_did) pairs whose repo state must be rebuilt. let mut repos: BTreeSet<(String, String)> = BTreeSet::new();
for (uri, space_id, author_did, collection, rkey, record_json, old_cid) in rows { report.records_scanned += 1; repos.insert((space_id.clone(), author_did.clone()));
let value: serde_json::Value = serde_json::from_str(&record_json) .map_err(|e| AppError::Internal(format!("record {uri} has unparseable JSON: {e}")))?;
let Some(new_cid) = crate::cid_verify::compute_record_cid(&value).map(|c| c.to_string()) else { report.records_unencodable += 1; tracing::warn!( uri, "record cannot be encoded as DAG-CBOR; leaving CID unchanged" ); continue; };
if new_cid == old_cid { continue; }
remap.insert( (space_id, author_did, collection, rkey, old_cid), new_cid.clone(), );
let update = adapt_sql( "UPDATE happyview_space_records SET cid = ? WHERE uri = ?", backend, ); crate::db::query(&update) .bind(&new_cid) .bind(&uri) .execute(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to update CID for {uri}: {e}")))?; report.records_updated += 1; }
// --------------------------------------------------------------------- // 2. Remap oplog CIDs // --------------------------------------------------------------------- let sql = adapt_sql( "SELECT id, space_id, author_did, collection, rkey, cid, prev FROM happyview_space_record_oplog", backend, ); type OplogRow = ( String, String, String, String, String, Option<String>, Option<String>, ); let ops: Vec<OplogRow> = crate::db::query_as(&sql) .fetch_all(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to load oplog: {e}")))?;
for (id, space_id, author_did, collection, rkey, cid, prev) in ops { let lookup = |c: &str| -> Option<String> { remap .get(&( space_id.clone(), author_did.clone(), collection.clone(), rkey.clone(), c.to_string(), )) .cloned() };
let new_cid = cid.as_deref().and_then(&lookup); let new_prev = prev.as_deref().and_then(&lookup);
// An op referencing a CID we could not remap points at content that is // gone (superseded or deleted), so it keeps its placeholder value. let unresolved = (cid.is_some() && new_cid.is_none() && is_placeholder(cid.as_deref())) || (prev.is_some() && new_prev.is_none() && is_placeholder(prev.as_deref())); if unresolved { report.oplog_rows_unresolved += 1; }
if new_cid.is_none() && new_prev.is_none() { continue; }
let update = adapt_sql( "UPDATE happyview_space_record_oplog SET cid = ?, prev = ? WHERE id = ?", backend, ); crate::db::query(&update) .bind(new_cid.or(cid)) .bind(new_prev.or(prev)) .bind(&id) .execute(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to remap oplog entry {id}: {e}")))?; report.oplog_rows_remapped += 1; }
// --------------------------------------------------------------------- // 3. Rebuild repo state // --------------------------------------------------------------------- // Include repos that already have a commit even if they now hold no records, // so an emptied repo is rebuilt to the empty-set hash rather than left stale. let sql = adapt_sql( "SELECT space_id, author_did FROM happyview_space_repo_state WHERE hash IS NOT NULL", backend, ); let existing: Vec<(String, String)> = crate::db::query_as(&sql) .fetch_all(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to load repo states: {e}")))?; repos.extend(existing);
for (space_id, author_did) in repos { let Some(space) = db::get_space(&mut *tx, backend, &space_id).await? else { tracing::warn!(space_id, "repo state references a missing space; skipping"); continue; };
let sql = adapt_sql( "SELECT collection, rkey, cid FROM happyview_space_records WHERE space_id = ? AND author_did = ?", backend, ); let records: Vec<(String, String, String)> = crate::db::query_as(&sql) .bind(&space_id) .bind(&author_did) .fetch_all(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to load records for rebuild: {e}")))?;
let mut set_hash = LtHashState::new(); for (collection, rkey, cid) in &records { set_hash.add(&record_element(collection, rkey, cid)); }
let space_uri = format!( "at://{}/space/{}/{}", space.did, space.type_nsid, space.skey ); let rev = generate_tid(); let signed = commit::sign_commit(&set_hash.hash(), &space_uri, &author_did, &rev)?;
let mut repo_state = db::get_or_create_repo_state(&mut tx, backend, &space_id, &author_did).await?; repo_state.lthash_state = set_hash.as_bytes().to_vec(); repo_state.rev = Some(signed.rev); repo_state.hash = Some(signed.hash.to_vec()); repo_state.ikm = Some(signed.ikm.to_vec()); repo_state.mac = Some(signed.mac.to_vec()); db::update_repo_state(&mut *tx, backend, &repo_state).await?;
db::update_space_revision(&mut *tx, backend, &space_id, &rev).await?; report.repos_rebuilt += 1; }
let now = now_rfc3339(); let marker_sql = adapt_sql( "INSERT INTO happyview_instance_settings (key, value, updated_at) VALUES (?, ?, ?) \ ON CONFLICT (key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", backend, ); crate::db::query(&marker_sql) .bind(BACKFILL_MARKER_KEY) .bind(&now) .bind(&now) .execute(&mut *tx) .await .map_err(|e| AppError::Internal(format!("failed to write backfill marker: {e}")))?;
if dry_run { tx.rollback() .await .map_err(|e| AppError::Internal(format!("failed to roll back dry run: {e}")))?; } else { tx.commit() .await .map_err(|e| AppError::Internal(format!("failed to commit repair: {e}")))?; }
Ok(report)}
fn is_placeholder(cid: Option<&str>) -> bool { let Some(cid) = cid else { return false }; let Some(rest) = cid.strip_prefix("bafyrei") else { return false; }; rest.len() == 40 && rest.bytes().all(|b| b.is_ascii_hexdigit())}
#[cfg(test)]mod tests { use super::*;
#[test] fn recognises_legacy_placeholder() { let legacy = format!("bafyrei{}", "a".repeat(40)); assert!(is_placeholder(Some(&legacy))); }
#[test] fn real_cid_is_not_a_placeholder() { let real = crate::cid_verify::compute_record_cid(&serde_json::json!({ "a": 1 })) .expect("encodable") .to_string(); assert!(real.starts_with("bafyrei"), "sanity: {real}"); assert!(!is_placeholder(Some(&real))); }
#[test] fn absent_cid_is_not_a_placeholder() { assert!(!is_placeholder(None)); }
#[test] fn report_noop_detection() { let mut report = BackfillReport::default(); assert!(report.is_noop()); report.records_updated = 1; assert!(!report.is_noop()); }}