From 91684f369eb71455783753229638bed19552ef1e Mon Sep 17 00:00:00 2001 From: Trezy Date: Fri, 01 May 2026 16:37:42 +0000 Subject: [PATCH] fix: remove public record aggregation from permissioned spaces Signed-off-by: Trezy --- src/spaces/notifications.rs | 86 -------------------------------------------------------------------------------------- src/spaces/sync.rs | 293 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- 2 file(s) changed, 0 insertion(s)(+), 379 deletion(s)(-) diff --git a/src/spaces/notifications.rs b/src/spaces/notifications.rs deleted file mode 100644 --- a/src/spaces/notifications.rs +++ /dev/null @@ -1,86 +0,0 @@ -use serde::Deserialize; -use uuid::Uuid; - -use crate::db::DatabaseBackend; -use crate::error::AppError; -use crate::spaces::db; -use crate::spaces::types::*; - -#[derive(Debug, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct WriteNotification { - pub space_uri: String, - pub author_did: String, - pub collection: String, - pub rkey: String, - pub action: WriteAction, -} - -#[derive(Debug, Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum WriteAction { - Create, - Update, - Delete, -} - -/// Process a write notification by queuing a sync pull for the affected member. -/// -/// This marks the member's sync state as pending so the next sync pass picks it up. -pub async fn handle_write_notification( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - space_id: &str, - notification: &WriteNotification, -) -> Result<(), AppError> { - let existing = db::get_sync_state(pool, backend, space_id, ¬ification.author_did).await?; - - let state = SpaceSyncState { - id: existing - .map(|s| s.id) - .unwrap_or_else(|| Uuid::new_v4().to_string()), - space_id: space_id.to_string(), - member_did: notification.author_did.clone(), - cursor: None, - last_synced_at: None, - status: SyncStatus::Pending, - error: None, - }; - - db::upsert_sync_state(pool, backend, &state).await?; - - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn write_action_deserializes() { - let action: WriteAction = serde_json::from_str("\"create\"").unwrap(); - assert!(matches!(action, WriteAction::Create)); - - let action: WriteAction = serde_json::from_str("\"update\"").unwrap(); - assert!(matches!(action, WriteAction::Update)); - - let action: WriteAction = serde_json::from_str("\"delete\"").unwrap(); - assert!(matches!(action, WriteAction::Delete)); - } - - #[test] - fn write_notification_deserializes() { - let json = r#"{ - "spaceUri": "ats://did:plc:owner/com.example.forum/main", - "authorDid": "did:plc:alice", - "collection": "com.example.forum.post", - "rkey": "3k2abc", - "action": "create" - }"#; - - let notif: WriteNotification = serde_json::from_str(json).unwrap(); - assert_eq!(notif.author_did, "did:plc:alice"); - assert_eq!(notif.collection, "com.example.forum.post"); - assert!(matches!(notif.action, WriteAction::Create)); - } -} diff --git a/src/spaces/sync.rs b/src/spaces/sync.rs deleted file mode 100644 --- a/src/spaces/sync.rs +++ /dev/null @@ -1,293 +0,0 @@ -use 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 { - 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 { - 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 { - 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 = 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, Option), 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, -} - -pub struct MemberSyncResult { - pub did: String, - pub records_synced: usize, - pub error: Option, -} - -pub struct MemberSyncSummary { - pub records_synced: usize, - pub cursor: Option, -} - -#[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"); - } -} -- tangled.sh