Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294use uuid::Uuid;
use crate::db::DatabaseBackend;use crate::db::now_rfc3339;use crate::error::AppError;use crate::profile::resolve_pds_endpoint;use crate::spaces::types::*;use crate::spaces::{db, members};
/// Sync all members of a space by pulling records from their PDSes.pub async fn sync_space( http: &reqwest::Client, pool: &sqlx::AnyPool, backend: DatabaseBackend, plc_url: &str, space_id: &str, collections: &[String],) -> Result<SyncSpaceResult, AppError> { let resolved = members::resolve_members(pool, backend, space_id).await?; let mut results = Vec::new();
for member in &resolved { let result = sync_member( http, pool, backend, plc_url, space_id, &member.did, collections, ) .await;
results.push(MemberSyncResult { did: member.did.clone(), records_synced: result.as_ref().map(|r| r.records_synced).unwrap_or(0), error: result.err().map(|e| e.to_string()), }); }
let total = results.iter().map(|r| r.records_synced).sum();
Ok(SyncSpaceResult { members_processed: results.len(), total_records_synced: total, member_results: results, })}
/// Sync records from a single member's PDS for a given space.pub async fn sync_member( http: &reqwest::Client, pool: &sqlx::AnyPool, backend: DatabaseBackend, plc_url: &str, space_id: &str, member_did: &str, collections: &[String],) -> Result<MemberSyncSummary, AppError> { let state_id = match db::get_sync_state(pool, backend, space_id, member_did).await? { Some(s) => s.id, None => { let id = Uuid::new_v4().to_string(); let initial = SpaceSyncState { id: id.clone(), space_id: space_id.to_string(), member_did: member_did.to_string(), cursor: None, last_synced_at: None, status: SyncStatus::Pending, error: None, }; db::upsert_sync_state(pool, backend, &initial).await?; id } };
// Mark as syncing let syncing_state = SpaceSyncState { id: state_id.clone(), space_id: space_id.to_string(), member_did: member_did.to_string(), cursor: None, last_synced_at: None, status: SyncStatus::Syncing, error: None, }; db::upsert_sync_state(pool, backend, &syncing_state).await?;
let result = pull_member_records( http, pool, backend, plc_url, space_id, member_did, collections, ) .await;
match result { Ok(summary) => { let done = SpaceSyncState { id: state_id, space_id: space_id.to_string(), member_did: member_did.to_string(), cursor: summary.cursor.clone(), last_synced_at: Some(now_rfc3339()), status: SyncStatus::Synced, error: None, }; db::upsert_sync_state(pool, backend, &done).await?; Ok(summary) } Err(e) => { let err_state = SpaceSyncState { id: state_id, space_id: space_id.to_string(), member_did: member_did.to_string(), cursor: None, last_synced_at: Some(now_rfc3339()), status: SyncStatus::Error, error: Some(e.to_string()), }; db::upsert_sync_state(pool, backend, &err_state).await?; Err(e) } }}
async fn pull_member_records( http: &reqwest::Client, pool: &sqlx::AnyPool, backend: DatabaseBackend, plc_url: &str, space_id: &str, member_did: &str, collections: &[String],) -> Result<MemberSyncSummary, AppError> { let pds_url = resolve_pds_endpoint(http, plc_url, member_did).await?; let mut total_records = 0usize; let mut last_cursor = None;
for collection in collections { let mut cursor: Option<String> = None; loop { let (records, next_cursor) = fetch_records_page( http, &pds_url, member_did, collection, cursor.as_deref(), 100, ) .await?;
if records.is_empty() { break; }
for record in &records { let uri = record["uri"].as_str().unwrap_or(""); let rkey = extract_rkey(uri); let cid = record["cid"].as_str().unwrap_or("").to_string(); let value = record .get("value") .cloned() .unwrap_or(serde_json::Value::Null);
let space_record_uri = format!("ats://{space_id}/{member_did}/{collection}/{rkey}");
let space_record = SpaceRecord { uri: space_record_uri, space_id: space_id.to_string(), author_did: member_did.to_string(), collection: collection.clone(), rkey: rkey.to_string(), record: value, cid, indexed_at: now_rfc3339(), };
db::upsert_space_record(pool, backend, &space_record).await?; total_records += 1; }
last_cursor = next_cursor.clone(); cursor = next_cursor;
if cursor.is_none() { break; } } }
Ok(MemberSyncSummary { records_synced: total_records, cursor: last_cursor, })}
async fn fetch_records_page( http: &reqwest::Client, pds_url: &str, repo: &str, collection: &str, cursor: Option<&str>, limit: u32,) -> Result<(Vec<serde_json::Value>, Option<String>), AppError> { let mut url = format!( "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection={}&limit={}", pds_url.trim_end_matches('/'), repo, collection, limit, );
if let Some(c) = cursor { url.push_str(&format!("&cursor={c}")); }
let resp = http .get(&url) .send() .await .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?;
if !resp.status().is_success() { let status = resp.status(); return Err(AppError::Internal(format!( "PDS listRecords failed with {status} for {repo}/{collection}" ))); }
let body: serde_json::Value = resp .json() .await .map_err(|e| AppError::Internal(format!("invalid PDS response: {e}")))?;
let records = body["records"].as_array().cloned().unwrap_or_default();
let next_cursor = body["cursor"].as_str().map(|s| s.to_string());
Ok((records, next_cursor))}
fn extract_rkey(uri: &str) -> &str { uri.rsplit('/').next().unwrap_or("")}
// ---------------------------------------------------------------------------// Result types// ---------------------------------------------------------------------------
pub struct SyncSpaceResult { pub members_processed: usize, pub total_records_synced: usize, pub member_results: Vec<MemberSyncResult>,}
pub struct MemberSyncResult { pub did: String, pub records_synced: usize, pub error: Option<String>,}
pub struct MemberSyncSummary { pub records_synced: usize, pub cursor: Option<String>,}
#[cfg(test)]mod tests { use super::*;
#[test] fn extract_rkey_from_at_uri() { assert_eq!( extract_rkey("at://did:plc:abc/app.bsky.feed.post/3k2abc"), "3k2abc" ); }
#[test] fn extract_rkey_from_empty() { assert_eq!(extract_rkey(""), ""); }
#[test] fn extract_rkey_no_slash() { assert_eq!(extract_rkey("singlevalue"), "singlevalue"); }}