diff --git a/src/spaces/db.rs b/src/spaces/db.rs index 173a053..5577a50 100644 --- a/src/spaces/db.rs +++ b/src/spaces/db.rs @@ -662,119 +662,3 @@ fn parse_invite_row(r: InviteRow) -> Result { created_at: r.9, }) } - -// --------------------------------------------------------------------------- -// Space Sync State -// --------------------------------------------------------------------------- - -pub async fn get_sync_state( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - space_id: &str, - member_did: &str, -) -> Result, AppError> { - let sql = adapt_sql( - "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE space_id = ? AND member_did = ?", - backend, - ); - - let row: Option = sqlx::query_as(&sql) - .bind(space_id) - .bind(member_did) - .fetch_optional(pool) - .await - .map_err(|e| AppError::Internal(format!("failed to get sync state: {e}")))?; - - row.map(parse_sync_state_row).transpose() -} - -pub async fn upsert_sync_state( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - state: &SpaceSyncState, -) -> Result<(), AppError> { - let sql = match backend { - DatabaseBackend::Sqlite => { - "INSERT OR REPLACE INTO space_sync_state (id, space_id, member_did, cursor, last_synced_at, status, error) VALUES (?, ?, ?, ?, ?, ?, ?)".to_string() - } - DatabaseBackend::Postgres => adapt_sql( - "INSERT INTO space_sync_state (id, space_id, member_did, cursor, last_synced_at, status, error) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (space_id, member_did) DO UPDATE SET cursor = EXCLUDED.cursor, last_synced_at = EXCLUDED.last_synced_at, status = EXCLUDED.status, error = EXCLUDED.error", - backend, - ), - }; - - sqlx::query(&sql) - .bind(&state.id) - .bind(&state.space_id) - .bind(&state.member_did) - .bind(&state.cursor) - .bind(&state.last_synced_at) - .bind(state.status.as_str()) - .bind(&state.error) - .execute(pool) - .await - .map_err(|e| AppError::Internal(format!("failed to upsert sync state: {e}")))?; - - Ok(()) -} - -pub async fn list_sync_states_for_space( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - space_id: &str, -) -> Result, AppError> { - let sql = adapt_sql( - "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE space_id = ? ORDER BY member_did ASC", - backend, - ); - - let rows: Vec = sqlx::query_as(&sql) - .bind(space_id) - .fetch_all(pool) - .await - .map_err(|e| AppError::Internal(format!("failed to list sync states: {e}")))?; - - rows.into_iter().map(parse_sync_state_row).collect() -} - -pub async fn list_pending_syncs( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, -) -> Result, AppError> { - let sql = adapt_sql( - "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE status = 'pending' OR status = 'error' ORDER BY last_synced_at ASC NULLS FIRST LIMIT 50", - backend, - ); - - let rows: Vec = sqlx::query_as(&sql) - .fetch_all(pool) - .await - .map_err(|e| AppError::Internal(format!("failed to list pending syncs: {e}")))?; - - rows.into_iter().map(parse_sync_state_row).collect() -} - -type SyncStateRow = ( - String, - String, - String, - Option, - Option, - String, - Option, -); - -fn parse_sync_state_row(r: SyncStateRow) -> Result { - let status = SyncStatus::parse(&r.5) - .ok_or_else(|| AppError::Internal(format!("invalid sync status: {}", r.5)))?; - - Ok(SpaceSyncState { - id: r.0, - space_id: r.1, - member_did: r.2, - cursor: r.3, - last_synced_at: r.4, - status, - error: r.6, - }) -} diff --git a/src/spaces/mod.rs b/src/spaces/mod.rs index a59c5d0..385f183 100644 --- a/src/spaces/mod.rs +++ b/src/spaces/mod.rs @@ -2,9 +2,7 @@ pub mod auth; pub mod credential; pub mod db; pub mod members; -pub mod notifications; pub mod routes; -pub mod sync; pub mod types; use crate::error::AppError; diff --git a/src/spaces/notifications.rs b/src/spaces/notifications.rs deleted file mode 100644 index 3d17e12..0000000 --- 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/routes.rs b/src/spaces/routes.rs index 235e2bb..543707c 100644 --- a/src/spaces/routes.rs +++ b/src/spaces/routes.rs @@ -146,16 +146,6 @@ struct RefreshCredentialInput { credential: String, } -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct WriteNotificationInput { - space_uri: String, - author_did: String, - collection: String, - rkey: String, - action: crate::spaces::notifications::WriteAction, -} - // --------------------------------------------------------------------------- // Route registration // --------------------------------------------------------------------------- @@ -208,11 +198,6 @@ pub fn space_routes() -> Router { &format!("/xrpc/{NS}.space.refreshCredential"), post(refresh_credential), ) - // Notifications - .route( - &format!("/xrpc/{NS}.space.writeNotification"), - post(write_notification), - ) } // --------------------------------------------------------------------------- @@ -948,35 +933,3 @@ async fn refresh_credential( "expiresAt": issued.expires_at, }))) } - -// --------------------------------------------------------------------------- -// Notification handlers -// --------------------------------------------------------------------------- - -async fn write_notification( - State(state): State, - xrpc_claims: XrpcClaims, - Json(input): Json, -) -> Result, AppError> { - let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; - require_space_admin(&state, &space, claims.did()).await?; - - let notification = crate::spaces::notifications::WriteNotification { - space_uri: input.space_uri, - author_did: input.author_did, - collection: input.collection, - rkey: input.rkey, - action: input.action, - }; - - crate::spaces::notifications::handle_write_notification( - &state.db, - state.db_backend, - &space.id, - ¬ification, - ) - .await?; - - Ok(Json(serde_json::json!({ "success": true }))) -} diff --git a/src/spaces/sync.rs b/src/spaces/sync.rs deleted file mode 100644 index 40d5f72..0000000 --- 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"); - } -} diff --git a/src/spaces/types.rs b/src/spaces/types.rs index 3d55ebb..c62c116 100644 --- a/src/spaces/types.rs +++ b/src/spaces/types.rs @@ -139,47 +139,6 @@ pub struct SpaceInvite { pub created_at: String, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum SyncStatus { - Pending, - Syncing, - Synced, - Error, -} - -impl SyncStatus { - pub fn as_str(&self) -> &'static str { - match self { - SyncStatus::Pending => "pending", - SyncStatus::Syncing => "syncing", - SyncStatus::Synced => "synced", - SyncStatus::Error => "error", - } - } - - pub fn parse(s: &str) -> Option { - match s { - "pending" => Some(SyncStatus::Pending), - "syncing" => Some(SyncStatus::Syncing), - "synced" => Some(SyncStatus::Synced), - "error" => Some(SyncStatus::Error), - _ => None, - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct SpaceSyncState { - pub id: String, - pub space_id: String, - pub member_did: String, - pub cursor: Option, - pub last_synced_at: Option, - pub status: SyncStatus, - pub error: Option, -} - #[cfg(test)] mod tests { use super::*;