-86

src/spaces/notifications.rs

Reviewed

··· 1 1 - use serde::Deserialize; 2 2 - use uuid::Uuid; 3 3 - 4 4 - use crate::db::DatabaseBackend; 5 5 - use crate::error::AppError; 6 6 - use crate::spaces::db; 7 7 - use crate::spaces::types::*; 8 8 - 9 9 - #[derive(Debug, Deserialize)] 10 10 - #[serde(rename_all = "camelCase")] 11 11 - pub struct WriteNotification { 12 12 - pub space_uri: String, 13 13 - pub author_did: String, 14 14 - pub collection: String, 15 15 - pub rkey: String, 16 16 - pub action: WriteAction, 17 17 - } 18 18 - 19 19 - #[derive(Debug, Deserialize)] 20 20 - #[serde(rename_all = "lowercase")] 21 21 - pub enum WriteAction { 22 22 - Create, 23 23 - Update, 24 24 - Delete, 25 25 - } 26 26 - 27 27 - /// Process a write notification by queuing a sync pull for the affected member. 28 28 - /// 29 29 - /// This marks the member's sync state as pending so the next sync pass picks it up. 30 30 - pub async fn handle_write_notification( 31 31 - pool: &sqlx::AnyPool, 32 32 - backend: DatabaseBackend, 33 33 - space_id: &str, 34 34 - notification: &WriteNotification, 35 35 - ) -> Result<(), AppError> { 36 36 - let existing = db::get_sync_state(pool, backend, space_id, &notification.author_did).await?; 37 37 - 38 38 - let state = SpaceSyncState { 39 39 - id: existing 40 40 - .map(|s| s.id) 41 41 - .unwrap_or_else(|| Uuid::new_v4().to_string()), 42 42 - space_id: space_id.to_string(), 43 43 - member_did: notification.author_did.clone(), 44 44 - cursor: None, 45 45 - last_synced_at: None, 46 46 - status: SyncStatus::Pending, 47 47 - error: None, 48 48 - }; 49 49 - 50 50 - db::upsert_sync_state(pool, backend, &state).await?; 51 51 - 52 52 - Ok(()) 53 53 - } 54 54 - 55 55 - #[cfg(test)] 56 56 - mod tests { 57 57 - use super::*; 58 58 - 59 59 - #[test] 60 60 - fn write_action_deserializes() { 61 61 - let action: WriteAction = serde_json::from_str("\"create\"").unwrap(); 62 62 - assert!(matches!(action, WriteAction::Create)); 63 63 - 64 64 - let action: WriteAction = serde_json::from_str("\"update\"").unwrap(); 65 65 - assert!(matches!(action, WriteAction::Update)); 66 66 - 67 67 - let action: WriteAction = serde_json::from_str("\"delete\"").unwrap(); 68 68 - assert!(matches!(action, WriteAction::Delete)); 69 69 - } 70 70 - 71 71 - #[test] 72 72 - fn write_notification_deserializes() { 73 73 - let json = r#"{ 74 74 - "spaceUri": "ats://did:plc:owner/com.example.forum/main", 75 75 - "authorDid": "did:plc:alice", 76 76 - "collection": "com.example.forum.post", 77 77 - "rkey": "3k2abc", 78 78 - "action": "create" 79 79 - }"#; 80 80 - 81 81 - let notif: WriteNotification = serde_json::from_str(json).unwrap(); 82 82 - assert_eq!(notif.author_did, "did:plc:alice"); 83 83 - assert_eq!(notif.collection, "com.example.forum.post"); 84 84 - assert!(matches!(notif.action, WriteAction::Create)); 85 85 - } 86 86 - }

-293

src/spaces/sync.rs

Reviewed

··· 1 1 - use uuid::Uuid; 2 2 - 3 3 - use crate::db::DatabaseBackend; 4 4 - use crate::db::now_rfc3339; 5 5 - use crate::error::AppError; 6 6 - use crate::profile::resolve_pds_endpoint; 7 7 - use crate::spaces::types::*; 8 8 - use crate::spaces::{db, members}; 9 9 - 10 10 - /// Sync all members of a space by pulling records from their PDSes. 11 11 - pub async fn sync_space( 12 12 - http: &reqwest::Client, 13 13 - pool: &sqlx::AnyPool, 14 14 - backend: DatabaseBackend, 15 15 - plc_url: &str, 16 16 - space_id: &str, 17 17 - collections: &[String], 18 18 - ) -> Result<SyncSpaceResult, AppError> { 19 19 - let resolved = members::resolve_members(pool, backend, space_id).await?; 20 20 - let mut results = Vec::new(); 21 21 - 22 22 - for member in &resolved { 23 23 - let result = sync_member( 24 24 - http, 25 25 - pool, 26 26 - backend, 27 27 - plc_url, 28 28 - space_id, 29 29 - &member.did, 30 30 - collections, 31 31 - ) 32 32 - .await; 33 33 - 34 34 - results.push(MemberSyncResult { 35 35 - did: member.did.clone(), 36 36 - records_synced: result.as_ref().map(|r| r.records_synced).unwrap_or(0), 37 37 - error: result.err().map(|e| e.to_string()), 38 38 - }); 39 39 - } 40 40 - 41 41 - let total = results.iter().map(|r| r.records_synced).sum(); 42 42 - 43 43 - Ok(SyncSpaceResult { 44 44 - members_processed: results.len(), 45 45 - total_records_synced: total, 46 46 - member_results: results, 47 47 - }) 48 48 - } 49 49 - 50 50 - /// Sync records from a single member's PDS for a given space. 51 51 - pub async fn sync_member( 52 52 - http: &reqwest::Client, 53 53 - pool: &sqlx::AnyPool, 54 54 - backend: DatabaseBackend, 55 55 - plc_url: &str, 56 56 - space_id: &str, 57 57 - member_did: &str, 58 58 - collections: &[String], 59 59 - ) -> Result<MemberSyncSummary, AppError> { 60 60 - let state_id = match db::get_sync_state(pool, backend, space_id, member_did).await? { 61 61 - Some(s) => s.id, 62 62 - None => { 63 63 - let id = Uuid::new_v4().to_string(); 64 64 - let initial = SpaceSyncState { 65 65 - id: id.clone(), 66 66 - space_id: space_id.to_string(), 67 67 - member_did: member_did.to_string(), 68 68 - cursor: None, 69 69 - last_synced_at: None, 70 70 - status: SyncStatus::Pending, 71 71 - error: None, 72 72 - }; 73 73 - db::upsert_sync_state(pool, backend, &initial).await?; 74 74 - id 75 75 - } 76 76 - }; 77 77 - 78 78 - // Mark as syncing 79 79 - let syncing_state = SpaceSyncState { 80 80 - id: state_id.clone(), 81 81 - space_id: space_id.to_string(), 82 82 - member_did: member_did.to_string(), 83 83 - cursor: None, 84 84 - last_synced_at: None, 85 85 - status: SyncStatus::Syncing, 86 86 - error: None, 87 87 - }; 88 88 - db::upsert_sync_state(pool, backend, &syncing_state).await?; 89 89 - 90 90 - let result = pull_member_records( 91 91 - http, 92 92 - pool, 93 93 - backend, 94 94 - plc_url, 95 95 - space_id, 96 96 - member_did, 97 97 - collections, 98 98 - ) 99 99 - .await; 100 100 - 101 101 - match result { 102 102 - Ok(summary) => { 103 103 - let done = SpaceSyncState { 104 104 - id: state_id, 105 105 - space_id: space_id.to_string(), 106 106 - member_did: member_did.to_string(), 107 107 - cursor: summary.cursor.clone(), 108 108 - last_synced_at: Some(now_rfc3339()), 109 109 - status: SyncStatus::Synced, 110 110 - error: None, 111 111 - }; 112 112 - db::upsert_sync_state(pool, backend, &done).await?; 113 113 - Ok(summary) 114 114 - } 115 115 - Err(e) => { 116 116 - let err_state = SpaceSyncState { 117 117 - id: state_id, 118 118 - space_id: space_id.to_string(), 119 119 - member_did: member_did.to_string(), 120 120 - cursor: None, 121 121 - last_synced_at: Some(now_rfc3339()), 122 122 - status: SyncStatus::Error, 123 123 - error: Some(e.to_string()), 124 124 - }; 125 125 - db::upsert_sync_state(pool, backend, &err_state).await?; 126 126 - Err(e) 127 127 - } 128 128 - } 129 129 - } 130 130 - 131 131 - async fn pull_member_records( 132 132 - http: &reqwest::Client, 133 133 - pool: &sqlx::AnyPool, 134 134 - backend: DatabaseBackend, 135 135 - plc_url: &str, 136 136 - space_id: &str, 137 137 - member_did: &str, 138 138 - collections: &[String], 139 139 - ) -> Result<MemberSyncSummary, AppError> { 140 140 - let pds_url = resolve_pds_endpoint(http, plc_url, member_did).await?; 141 141 - let mut total_records = 0usize; 142 142 - let mut last_cursor = None; 143 143 - 144 144 - for collection in collections { 145 145 - let mut cursor: Option<String> = None; 146 146 - loop { 147 147 - let (records, next_cursor) = fetch_records_page( 148 148 - http, 149 149 - &pds_url, 150 150 - member_did, 151 151 - collection, 152 152 - cursor.as_deref(), 153 153 - 100, 154 154 - ) 155 155 - .await?; 156 156 - 157 157 - if records.is_empty() { 158 158 - break; 159 159 - } 160 160 - 161 161 - for record in &records { 162 162 - let uri = record["uri"].as_str().unwrap_or(""); 163 163 - let rkey = extract_rkey(uri); 164 164 - let cid = record["cid"].as_str().unwrap_or("").to_string(); 165 165 - let value = record 166 166 - .get("value") 167 167 - .cloned() 168 168 - .unwrap_or(serde_json::Value::Null); 169 169 - 170 170 - let space_record_uri = format!("ats://{space_id}/{member_did}/{collection}/{rkey}"); 171 171 - 172 172 - let space_record = SpaceRecord { 173 173 - uri: space_record_uri, 174 174 - space_id: space_id.to_string(), 175 175 - author_did: member_did.to_string(), 176 176 - collection: collection.clone(), 177 177 - rkey: rkey.to_string(), 178 178 - record: value, 179 179 - cid, 180 180 - indexed_at: now_rfc3339(), 181 181 - }; 182 182 - 183 183 - db::upsert_space_record(pool, backend, &space_record).await?; 184 184 - total_records += 1; 185 185 - } 186 186 - 187 187 - last_cursor = next_cursor.clone(); 188 188 - cursor = next_cursor; 189 189 - 190 190 - if cursor.is_none() { 191 191 - break; 192 192 - } 193 193 - } 194 194 - } 195 195 - 196 196 - Ok(MemberSyncSummary { 197 197 - records_synced: total_records, 198 198 - cursor: last_cursor, 199 199 - }) 200 200 - } 201 201 - 202 202 - async fn fetch_records_page( 203 203 - http: &reqwest::Client, 204 204 - pds_url: &str, 205 205 - repo: &str, 206 206 - collection: &str, 207 207 - cursor: Option<&str>, 208 208 - limit: u32, 209 209 - ) -> Result<(Vec<serde_json::Value>, Option<String>), AppError> { 210 210 - let mut url = format!( 211 211 - "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection={}&limit={}", 212 212 - pds_url.trim_end_matches('/'), 213 213 - repo, 214 214 - collection, 215 215 - limit, 216 216 - ); 217 217 - 218 218 - if let Some(c) = cursor { 219 219 - url.push_str(&format!("&cursor={c}")); 220 220 - } 221 221 - 222 222 - let resp = http 223 223 - .get(&url) 224 224 - .send() 225 225 - .await 226 226 - .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?; 227 227 - 228 228 - if !resp.status().is_success() { 229 229 - let status = resp.status(); 230 230 - return Err(AppError::Internal(format!( 231 231 - "PDS listRecords failed with {status} for {repo}/{collection}" 232 232 - ))); 233 233 - } 234 234 - 235 235 - let body: serde_json::Value = resp 236 236 - .json() 237 237 - .await 238 238 - .map_err(|e| AppError::Internal(format!("invalid PDS response: {e}")))?; 239 239 - 240 240 - let records = body["records"].as_array().cloned().unwrap_or_default(); 241 241 - 242 242 - let next_cursor = body["cursor"].as_str().map(|s| s.to_string()); 243 243 - 244 244 - Ok((records, next_cursor)) 245 245 - } 246 246 - 247 247 - fn extract_rkey(uri: &str) -> &str { 248 248 - uri.rsplit('/').next().unwrap_or("") 249 249 - } 250 250 - 251 251 - // --------------------------------------------------------------------------- 252 252 - // Result types 253 253 - // --------------------------------------------------------------------------- 254 254 - 255 255 - pub struct SyncSpaceResult { 256 256 - pub members_processed: usize, 257 257 - pub total_records_synced: usize, 258 258 - pub member_results: Vec<MemberSyncResult>, 259 259 - } 260 260 - 261 261 - pub struct MemberSyncResult { 262 262 - pub did: String, 263 263 - pub records_synced: usize, 264 264 - pub error: Option<String>, 265 265 - } 266 266 - 267 267 - pub struct MemberSyncSummary { 268 268 - pub records_synced: usize, 269 269 - pub cursor: Option<String>, 270 270 - } 271 271 - 272 272 - #[cfg(test)] 273 273 - mod tests { 274 274 - use super::*; 275 275 - 276 276 - #[test] 277 277 - fn extract_rkey_from_at_uri() { 278 278 - assert_eq!( 279 279 - extract_rkey("at://did:plc:abc/app.bsky.feed.post/3k2abc"), 280 280 - "3k2abc" 281 281 - ); 282 282 - } 283 283 - 284 284 - #[test] 285 285 - fn extract_rkey_from_empty() { 286 286 - assert_eq!(extract_rkey(""), ""); 287 287 - } 288 288 - 289 289 - #[test] 290 290 - fn extract_rkey_no_slash() { 291 291 - assert_eq!(extract_rkey("singlevalue"), "singlevalue"); 292 292 - } 293 293 - }