diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 07bc377..ae7151f 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -7,6 +7,7 @@ pub mod oauth; pub mod pds; pub mod repository; pub mod state; +pub mod sync_service; pub mod well_known; use axum::http::Method; diff --git a/crates/server/src/pds/client.rs b/crates/server/src/pds/client.rs index 8750527..f63bd9a 100644 --- a/crates/server/src/pds/client.rs +++ b/crates/server/src/pds/client.rs @@ -40,6 +40,23 @@ pub struct PutRecordResponse { pub cid: String, } +/// Response from getRecord XRPC. +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct GetRecordResponse { + pub uri: String, + pub cid: String, + pub value: serde_json::Value, +} + +/// Result of getting a record from PDS. +#[derive(Debug, Clone)] +pub struct GetRecordResult { + pub uri: String, + pub cid: String, + pub value: serde_json::Value, +} + /// Request body for deleteRecord XRPC. #[derive(Serialize)] #[serde(rename_all = "camelCase")] @@ -124,13 +141,6 @@ impl PdsClient { } /// Create or update a record in the repository. - /// - /// # Arguments - /// - /// * `did` - The user's DID (repository owner) - /// * `collection` - The collection NSID (e.g., "org.stormlightlabs.malfestio.deck") - /// * `rkey` - The record key (TID) - /// * `record` - The record data as JSON pub async fn put_record( &self, did: &str, collection: &str, rkey: &str, record: serde_json::Value, ) -> Result { @@ -166,6 +176,42 @@ impl PdsClient { self.handle_response(response).await } + /// Get a record from the repository. + pub async fn get_record(&self, did: &str, collection: &str, rkey: &str) -> Result { + let url = format!( + "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection={}&rkey={}", + self.pds_url, did, collection, rkey + ); + + let mut request_builder = self.http_client.get(&url); + + if let Some(ref dpop_keypair) = self.dpop_keypair { + let dpop_proof = dpop_keypair.generate_proof("GET", &url, Some(&self.access_token)); + request_builder = request_builder + .header("Authorization", format!("DPoP {}", self.access_token)) + .header("DPoP", dpop_proof); + } else { + request_builder = request_builder.header("Authorization", format!("Bearer {}", self.access_token)); + } + + let response = request_builder + .send() + .await + .map_err(|e| PdsError::NetworkError(e.to_string()))?; + + if response.status().is_success() { + let get_response: GetRecordResponse = response + .json() + .await + .map_err(|e| PdsError::NetworkError(format!("Failed to parse response: {}", e)))?; + Ok(GetRecordResult { uri: get_response.uri, cid: get_response.cid, value: get_response.value }) + } else { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + Err(self.map_error_status(status, body)) + } + } + /// Delete a record from the repository. pub async fn delete_record(&self, did: &str, collection: &str, rkey: &str) -> Result<(), PdsError> { let url = format!("{}/xrpc/com.atproto.repo.deleteRecord", self.pds_url); diff --git a/crates/server/src/repository/mod.rs b/crates/server/src/repository/mod.rs index 94e3688..3b14760 100644 --- a/crates/server/src/repository/mod.rs +++ b/crates/server/src/repository/mod.rs @@ -6,3 +6,4 @@ pub mod preferences; pub mod review; pub mod search; pub mod social; +pub mod sync; diff --git a/crates/server/src/repository/sync.rs b/crates/server/src/repository/sync.rs new file mode 100644 index 0000000..b9cf470 --- /dev/null +++ b/crates/server/src/repository/sync.rs @@ -0,0 +1,749 @@ +//! Sync repository for tracking synchronization state. +//! +//! Manages the sync status of entities between local database and PDS. + +use std::str::FromStr; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use uuid::Uuid; + +/// Sync status for an entity. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SyncStatus { + /// Never synced to PDS + LocalOnly, + /// In sync with PDS + Synced, + /// Local changes need to be pushed + PendingPush, + /// Local and remote both changed (conflict) + Conflict, +} + +impl SyncStatus { + pub fn as_str(&self) -> &'static str { + match self { + SyncStatus::LocalOnly => "local_only", + SyncStatus::Synced => "synced", + SyncStatus::PendingPush => "pending_push", + SyncStatus::Conflict => "conflict", + } + } +} + +impl FromStr for SyncStatus { + type Err = std::fmt::Error; + + fn from_str(s: &str) -> Result { + match s { + "local_only" => Ok(SyncStatus::LocalOnly), + "synced" => Ok(SyncStatus::Synced), + "pending_push" => Ok(SyncStatus::PendingPush), + "conflict" => Ok(SyncStatus::Conflict), + _ => Err(std::fmt::Error), + } + } +} + +impl std::fmt::Display for SyncStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// A pending sync operation. +#[derive(Debug, Clone)] +pub struct PendingSync { + pub entity_type: String, + pub entity_id: String, + pub owner_did: String, + pub version: i32, + pub status: SyncStatus, +} + +/// Sync metadata for an entity. +#[derive(Debug, Clone)] +pub struct SyncMetadata { + pub entity_type: String, + pub entity_id: String, + pub version: i32, + pub pds_cid: Option, + pub pds_uri: Option, + pub status: SyncStatus, + pub last_synced_at: Option>, +} + +/// Entry in the sync log. +#[derive(Debug, Clone)] +pub struct SyncLogEntry { + pub id: String, + pub owner_did: String, + pub entity_type: String, + pub entity_id: String, + pub operation: String, + pub status: String, + pub pds_cid: Option, + pub error_message: Option, + pub created_at: DateTime, + pub completed_at: Option>, +} + +/// Parameters for logging a sync operation. +#[derive(Debug, Clone)] +pub struct LogOperationParams<'a> { + pub owner_did: &'a str, + pub entity_type: &'a str, + pub entity_id: &'a str, + pub operation: &'a str, + pub status: &'a str, + pub pds_cid: Option<&'a str>, + pub error_message: Option<&'a str>, +} + +/// Error type for sync repository operations. +#[derive(Debug)] +pub enum SyncRepoError { + DatabaseError(String), + NotFound(String), + InvalidArgument(String), +} + +impl std::fmt::Display for SyncRepoError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SyncRepoError::DatabaseError(e) => write!(f, "Database error: {}", e), + SyncRepoError::NotFound(e) => write!(f, "Not found: {}", e), + SyncRepoError::InvalidArgument(e) => write!(f, "Invalid argument: {}", e), + } + } +} + +impl std::error::Error for SyncRepoError {} + +/// Repository trait for sync operations. +#[async_trait] +pub trait SyncRepository: Send + Sync { + /// Get sync metadata for an entity. + async fn get_sync_metadata(&self, entity_type: &str, id: &str) -> Result; + + /// Mark an entity as synced with given PDS CID and URI. + async fn mark_synced(&self, entity_type: &str, id: &str, pds_cid: &str, pds_uri: &str) + -> Result<(), SyncRepoError>; + + /// Mark an entity as pending push. + async fn mark_pending(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError>; + + /// Mark an entity as having a conflict. + async fn mark_conflict(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError>; + + /// Get all pending items for a user. + async fn get_pending_items(&self, owner_did: &str) -> Result, SyncRepoError>; + + /// Get all conflicts for a user. + async fn get_conflicts(&self, owner_did: &str) -> Result, SyncRepoError>; + + /// Increment version for an entity (used when resolving conflicts). + async fn increment_version(&self, entity_type: &str, id: &str) -> Result; + + /// Log a sync operation. + async fn log_operation(&self, params: LogOperationParams<'_>) -> Result; + + /// Mark a sync log entry as completed. + async fn complete_log_entry( + &self, log_id: &str, status: &str, pds_cid: Option<&str>, error_message: Option<&str>, + ) -> Result<(), SyncRepoError>; +} + +/// Database implementation of SyncRepository. +pub struct DbSyncRepository { + pool: crate::db::DbPool, +} + +impl DbSyncRepository { + pub fn new(pool: crate::db::DbPool) -> Self { + Self { pool } + } + + fn table_for_entity(&self, entity_type: &str) -> Result<&'static str, SyncRepoError> { + match entity_type { + "deck" => Ok("decks"), + "card" => Ok("cards"), + "note" => Ok("notes"), + _ => Err(SyncRepoError::InvalidArgument(format!( + "Unknown entity type: {}", + entity_type + ))), + } + } +} + +#[async_trait] +impl SyncRepository for DbSyncRepository { + async fn get_sync_metadata(&self, entity_type: &str, id: &str) -> Result { + let table = self.table_for_entity(entity_type)?; + let uuid = Uuid::parse_str(id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let query = format!( + "SELECT version, pds_cid, pds_uri, sync_status::text, last_synced_at FROM {} WHERE id = $1", + table + ); + + let row = client + .query_opt(&query, &[&uuid]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to query: {}", e)))? + .ok_or_else(|| SyncRepoError::NotFound(format!("{} not found: {}", entity_type, id)))?; + + let version: i32 = row.get("version"); + let pds_cid: Option = row.get("pds_cid"); + let pds_uri: Option = row.get("pds_uri"); + let status_str: String = row.get("sync_status"); + let last_synced_at: Option> = row.get("last_synced_at"); + + Ok(SyncMetadata { + entity_type: entity_type.to_string(), + entity_id: id.to_string(), + version, + pds_cid, + pds_uri, + status: SyncStatus::from_str(&status_str).unwrap_or(SyncStatus::LocalOnly), + last_synced_at, + }) + } + + async fn mark_synced( + &self, entity_type: &str, id: &str, pds_cid: &str, pds_uri: &str, + ) -> Result<(), SyncRepoError> { + let table = self.table_for_entity(entity_type)?; + let uuid = Uuid::parse_str(id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let query = format!( + "UPDATE {} SET sync_status = 'synced', pds_cid = $1, pds_uri = $2, last_synced_at = NOW() WHERE id = $3", + table + ); + + client + .execute(&query, &[&pds_cid, &pds_uri, &uuid]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to update: {}", e)))?; + + Ok(()) + } + + async fn mark_pending(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError> { + let table = self.table_for_entity(entity_type)?; + let uuid = Uuid::parse_str(id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let query = format!("UPDATE {} SET sync_status = 'pending_push' WHERE id = $1", table); + + client + .execute(&query, &[&uuid]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to update: {}", e)))?; + + Ok(()) + } + + async fn mark_conflict(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError> { + let table = self.table_for_entity(entity_type)?; + let uuid = Uuid::parse_str(id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let query = format!("UPDATE {} SET sync_status = 'conflict' WHERE id = $1", table); + + client + .execute(&query, &[&uuid]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to update: {}", e)))?; + + Ok(()) + } + + async fn get_pending_items(&self, owner_did: &str) -> Result, SyncRepoError> { + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let mut pending = Vec::new(); + + for (entity_type, table) in [("deck", "decks"), ("card", "cards"), ("note", "notes")] { + let query = format!( + "SELECT id, version, sync_status::text FROM {} WHERE owner_did = $1 AND sync_status = 'pending_push'", + table + ); + + let rows = client + .query(&query, &[&owner_did]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to query: {}", e)))?; + + for row in rows { + let id: Uuid = row.get("id"); + let version: i32 = row.get("version"); + let status_str: String = row.get("sync_status"); + + pending.push(PendingSync { + entity_type: entity_type.to_string(), + entity_id: id.to_string(), + owner_did: owner_did.to_string(), + version, + status: SyncStatus::from_str(&status_str).unwrap_or(SyncStatus::PendingPush), + }); + } + } + + Ok(pending) + } + + async fn get_conflicts(&self, owner_did: &str) -> Result, SyncRepoError> { + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let mut conflicts = Vec::new(); + + for (entity_type, table) in [("deck", "decks"), ("card", "cards"), ("note", "notes")] { + let query = format!( + "SELECT id, version, sync_status::text FROM {} WHERE owner_did = $1 AND sync_status = 'conflict'", + table + ); + + let rows = client + .query(&query, &[&owner_did]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to query: {}", e)))?; + + for row in rows { + let id: Uuid = row.get("id"); + let version: i32 = row.get("version"); + + conflicts.push(PendingSync { + entity_type: entity_type.to_string(), + entity_id: id.to_string(), + owner_did: owner_did.to_string(), + version, + status: SyncStatus::Conflict, + }); + } + } + + Ok(conflicts) + } + + async fn increment_version(&self, entity_type: &str, id: &str) -> Result { + let table = self.table_for_entity(entity_type)?; + let uuid = Uuid::parse_str(id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let query = format!( + "UPDATE {} SET version = version + 1 WHERE id = $1 RETURNING version", + table + ); + + let row = client + .query_one(&query, &[&uuid]) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to update: {}", e)))?; + + let version: i32 = row.get("version"); + Ok(version) + } + + async fn log_operation(&self, params: LogOperationParams<'_>) -> Result { + let entity_uuid = Uuid::parse_str(params.entity_id) + .map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + let row = client + .query_one( + "INSERT INTO sync_log (owner_did, entity_type, entity_id, operation, status, pds_cid, error_message) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING id", + &[ + ¶ms.owner_did, + ¶ms.entity_type, + &entity_uuid, + ¶ms.operation, + ¶ms.status, + ¶ms.pds_cid, + ¶ms.error_message, + ], + ) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to insert log: {}", e)))?; + + let id: Uuid = row.get("id"); + Ok(id.to_string()) + } + + async fn complete_log_entry( + &self, log_id: &str, status: &str, pds_cid: Option<&str>, error_message: Option<&str>, + ) -> Result<(), SyncRepoError> { + let uuid = + Uuid::parse_str(log_id).map_err(|e| SyncRepoError::InvalidArgument(format!("Invalid UUID: {}", e)))?; + + let client = self + .pool + .get() + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to get connection: {}", e)))?; + + client + .execute( + "UPDATE sync_log + SET status = $1, pds_cid = COALESCE($2, pds_cid), error_message = $3, completed_at = NOW() + WHERE id = $4", + &[&status, &pds_cid, &error_message, &uuid], + ) + .await + .map_err(|e| SyncRepoError::DatabaseError(format!("Failed to update log: {}", e)))?; + + Ok(()) + } +} + +#[cfg(test)] +pub mod mock { + use super::*; + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + #[derive(Clone)] + pub struct MockSyncRepository { + metadata: Arc>>, + logs: Arc>>, + } + + impl MockSyncRepository { + pub fn new() -> Self { + Self { metadata: Arc::new(Mutex::new(HashMap::new())), logs: Arc::new(Mutex::new(Vec::new())) } + } + + pub fn with_metadata(metadata: Vec) -> Self { + let map: HashMap = metadata + .into_iter() + .map(|m| (format!("{}:{}", m.entity_type, m.entity_id), m)) + .collect(); + Self { metadata: Arc::new(Mutex::new(map)), logs: Arc::new(Mutex::new(Vec::new())) } + } + + fn key(entity_type: &str, id: &str) -> String { + format!("{}:{}", entity_type, id) + } + } + + impl Default for MockSyncRepository { + fn default() -> Self { + Self::new() + } + } + + #[async_trait] + impl SyncRepository for MockSyncRepository { + async fn get_sync_metadata(&self, entity_type: &str, id: &str) -> Result { + let key = Self::key(entity_type, id); + self.metadata + .lock() + .unwrap() + .get(&key) + .cloned() + .ok_or_else(|| SyncRepoError::NotFound(format!("{} not found: {}", entity_type, id))) + } + + async fn mark_synced( + &self, entity_type: &str, id: &str, pds_cid: &str, pds_uri: &str, + ) -> Result<(), SyncRepoError> { + let key = Self::key(entity_type, id); + let mut map = self.metadata.lock().unwrap(); + if let Some(meta) = map.get_mut(&key) { + meta.status = SyncStatus::Synced; + meta.pds_cid = Some(pds_cid.to_string()); + meta.pds_uri = Some(pds_uri.to_string()); + meta.last_synced_at = Some(Utc::now()); + } + Ok(()) + } + + async fn mark_pending(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError> { + let key = Self::key(entity_type, id); + let mut map = self.metadata.lock().unwrap(); + if let Some(meta) = map.get_mut(&key) { + meta.status = SyncStatus::PendingPush; + } + Ok(()) + } + + async fn mark_conflict(&self, entity_type: &str, id: &str) -> Result<(), SyncRepoError> { + let key = Self::key(entity_type, id); + let mut map = self.metadata.lock().unwrap(); + if let Some(meta) = map.get_mut(&key) { + meta.status = SyncStatus::Conflict; + } + Ok(()) + } + + async fn get_pending_items(&self, owner_did: &str) -> Result, SyncRepoError> { + let map = self.metadata.lock().unwrap(); + let pending: Vec<_> = map + .values() + .filter(|m| m.status == SyncStatus::PendingPush) + .map(|m| PendingSync { + entity_type: m.entity_type.clone(), + entity_id: m.entity_id.clone(), + owner_did: owner_did.to_string(), + version: m.version, + status: m.status, + }) + .collect(); + Ok(pending) + } + + async fn get_conflicts(&self, owner_did: &str) -> Result, SyncRepoError> { + let map = self.metadata.lock().unwrap(); + let conflicts: Vec<_> = map + .values() + .filter(|m| m.status == SyncStatus::Conflict) + .map(|m| PendingSync { + entity_type: m.entity_type.clone(), + entity_id: m.entity_id.clone(), + owner_did: owner_did.to_string(), + version: m.version, + status: m.status, + }) + .collect(); + Ok(conflicts) + } + + async fn increment_version(&self, entity_type: &str, id: &str) -> Result { + let key = Self::key(entity_type, id); + let mut map = self.metadata.lock().unwrap(); + if let Some(meta) = map.get_mut(&key) { + meta.version += 1; + Ok(meta.version) + } else { + Err(SyncRepoError::NotFound(format!("{} not found: {}", entity_type, id))) + } + } + + async fn log_operation(&self, params: LogOperationParams<'_>) -> Result { + let id = Uuid::new_v4().to_string(); + let entry = SyncLogEntry { + id: id.clone(), + owner_did: params.owner_did.to_string(), + entity_type: params.entity_type.to_string(), + entity_id: params.entity_id.to_string(), + operation: params.operation.to_string(), + status: params.status.to_string(), + pds_cid: params.pds_cid.map(String::from), + error_message: params.error_message.map(String::from), + created_at: Utc::now(), + completed_at: None, + }; + self.logs.lock().unwrap().push(entry); + Ok(id) + } + + async fn complete_log_entry( + &self, log_id: &str, status: &str, pds_cid: Option<&str>, error_message: Option<&str>, + ) -> Result<(), SyncRepoError> { + let mut logs = self.logs.lock().unwrap(); + if let Some(entry) = logs.iter_mut().find(|e| e.id == log_id) { + entry.status = status.to_string(); + entry.pds_cid = pds_cid.map(String::from).or(entry.pds_cid.clone()); + entry.error_message = error_message.map(String::from); + entry.completed_at = Some(Utc::now()); + } + Ok(()) + } + } +} + +#[cfg(test)] +mod tests { + use super::mock::MockSyncRepository; + use super::*; + + #[test] + fn test_sync_status_as_str() { + assert_eq!(SyncStatus::LocalOnly.as_str(), "local_only"); + assert_eq!(SyncStatus::Synced.as_str(), "synced"); + assert_eq!(SyncStatus::PendingPush.as_str(), "pending_push"); + assert_eq!(SyncStatus::Conflict.as_str(), "conflict"); + } + + #[test] + fn test_sync_status_from_str() { + assert_eq!(SyncStatus::from_str("local_only").unwrap(), SyncStatus::LocalOnly); + assert_eq!(SyncStatus::from_str("synced").unwrap(), SyncStatus::Synced); + assert_eq!(SyncStatus::from_str("pending_push").unwrap(), SyncStatus::PendingPush); + assert_eq!(SyncStatus::from_str("conflict").unwrap(), SyncStatus::Conflict); + assert!(SyncStatus::from_str("unknown").is_err()); + } + + #[tokio::test] + async fn test_mock_sync_repo_get_metadata() { + let metadata = SyncMetadata { + entity_type: "deck".to_string(), + entity_id: "123".to_string(), + version: 1, + pds_cid: None, + pds_uri: None, + status: SyncStatus::LocalOnly, + last_synced_at: None, + }; + let repo = MockSyncRepository::with_metadata(vec![metadata]); + + let result = repo.get_sync_metadata("deck", "123").await; + assert!(result.is_ok()); + let meta = result.unwrap(); + assert_eq!(meta.entity_type, "deck"); + assert_eq!(meta.version, 1); + assert_eq!(meta.status, SyncStatus::LocalOnly); + } + + #[tokio::test] + async fn test_mock_sync_repo_mark_synced() { + let metadata = SyncMetadata { + entity_type: "deck".to_string(), + entity_id: "123".to_string(), + version: 1, + pds_cid: None, + pds_uri: None, + status: SyncStatus::PendingPush, + last_synced_at: None, + }; + let repo = MockSyncRepository::with_metadata(vec![metadata]); + + repo.mark_synced("deck", "123", "bafycid123", "at://did:plc:test/deck/123") + .await + .unwrap(); + + let meta = repo.get_sync_metadata("deck", "123").await.unwrap(); + assert_eq!(meta.status, SyncStatus::Synced); + assert_eq!(meta.pds_cid, Some("bafycid123".to_string())); + assert!(meta.last_synced_at.is_some()); + } + + #[tokio::test] + async fn test_mock_sync_repo_increment_version() { + let metadata = SyncMetadata { + entity_type: "note".to_string(), + entity_id: "456".to_string(), + version: 5, + pds_cid: None, + pds_uri: None, + status: SyncStatus::Synced, + last_synced_at: None, + }; + let repo = MockSyncRepository::with_metadata(vec![metadata]); + + let new_version = repo.increment_version("note", "456").await.unwrap(); + assert_eq!(new_version, 6); + + let meta = repo.get_sync_metadata("note", "456").await.unwrap(); + assert_eq!(meta.version, 6); + } + + #[tokio::test] + async fn test_mock_sync_repo_log_operation() { + let repo = MockSyncRepository::new(); + + let log_id = repo + .log_operation(LogOperationParams { + owner_did: "did:plc:test", + entity_type: "deck", + entity_id: "123e4567-e89b-12d3-a456-426614174000", + operation: "push", + status: "pending", + pds_cid: None, + error_message: None, + }) + .await + .unwrap(); + + assert!(!log_id.is_empty()); + + repo.complete_log_entry(&log_id, "success", Some("bafycid"), None) + .await + .unwrap(); + } + + #[tokio::test] + async fn test_mock_sync_repo_get_pending() { + let metadata = vec![ + SyncMetadata { + entity_type: "deck".to_string(), + entity_id: "1".to_string(), + version: 1, + pds_cid: None, + pds_uri: None, + status: SyncStatus::PendingPush, + last_synced_at: None, + }, + SyncMetadata { + entity_type: "note".to_string(), + entity_id: "2".to_string(), + version: 1, + pds_cid: None, + pds_uri: None, + status: SyncStatus::Synced, + last_synced_at: None, + }, + ]; + let repo = MockSyncRepository::with_metadata(metadata); + + let pending = repo.get_pending_items("did:plc:test").await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].entity_type, "deck"); + } + + #[test] + fn test_sync_repo_error_display() { + let err = SyncRepoError::DatabaseError("connection failed".to_string()); + assert!(err.to_string().contains("Database error")); + + let err = SyncRepoError::NotFound("deck:123".to_string()); + assert!(err.to_string().contains("Not found")); + + let err = SyncRepoError::InvalidArgument("bad uuid".to_string()); + assert!(err.to_string().contains("Invalid argument")); + } +} diff --git a/crates/server/src/sync_service.rs b/crates/server/src/sync_service.rs new file mode 100644 index 0000000..26cccd6 --- /dev/null +++ b/crates/server/src/sync_service.rs @@ -0,0 +1,518 @@ +//! Sync service for coordinating bi-directional PDS synchronization. +//! +//! Handles push/pull operations and conflict resolution. + +use crate::middleware::auth::UserContext; +use crate::pds::client::{GetRecordResult, PdsClient, PdsError}; +use crate::pds::records::{prepare_card_record, prepare_deck_record, prepare_note_record}; +use crate::repository::card::CardRepository; +use crate::repository::deck::DeckRepository; +use crate::repository::note::NoteRepository; +use crate::repository::oauth::OAuthRepository; +use crate::repository::sync::{LogOperationParams, SyncRepoError, SyncRepository, SyncStatus}; +use std::str::FromStr; +use std::sync::Arc; + +/// Result of a sync operation. +#[derive(Debug, Clone)] +pub struct SyncResult { + pub entity_type: String, + pub entity_id: String, + pub pds_uri: Option, + pub pds_cid: Option, + pub new_version: i32, + pub status: SyncStatus, +} + +/// Conflict information for UI display. +#[derive(Debug, Clone)] +pub struct ConflictInfo { + pub entity_type: String, + pub entity_id: String, + pub local_version: i32, + pub remote_version: Option, + pub local_updated_at: Option, + pub remote_updated_at: Option, +} + +/// Summary of sync status for a user. +#[derive(Debug, Clone)] +pub struct SyncStatusSummary { + pub pending_count: usize, + pub conflict_count: usize, + pub pending_items: Vec<(String, String)>, + pub conflicts: Vec, +} + +/// Conflict resolution strategy. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConflictStrategy { + /// Use the most recently modified version (default) + LastWriteWins, + /// Keep local version, overwrite remote + KeepLocal, + /// Keep remote version, overwrite local + KeepRemote, + // TODO: MergeUI - Show UI for manual merge +} + +impl ConflictStrategy { + pub fn as_str(&self) -> &'static str { + match self { + ConflictStrategy::LastWriteWins => "last_write_wins", + ConflictStrategy::KeepLocal => "keep_local", + ConflictStrategy::KeepRemote => "keep_remote", + } + } +} + +impl FromStr for ConflictStrategy { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "last_write_wins" => Ok(ConflictStrategy::LastWriteWins), + "keep_local" => Ok(ConflictStrategy::KeepLocal), + "keep_remote" => Ok(ConflictStrategy::KeepRemote), + _ => Err(format!("Invalid conflict strategy: {}", s)), + } + } +} + +/// Error type for sync operations. +#[derive(Debug)] +pub enum SyncError { + /// Entity not found + NotFound(String), + /// Authentication required + AuthRequired(String), + /// No OAuth tokens available + NoTokens(String), + /// PDS operation failed + PdsError(PdsError), + /// Repository error + RepoError(SyncRepoError), + /// Invalid argument + InvalidArgument(String), + /// Conflict detected + ConflictDetected(ConflictInfo), +} + +impl std::fmt::Display for SyncError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SyncError::NotFound(e) => write!(f, "Not found: {}", e), + SyncError::AuthRequired(e) => write!(f, "Authentication required: {}", e), + SyncError::NoTokens(e) => write!(f, "No OAuth tokens: {}", e), + SyncError::PdsError(e) => write!(f, "PDS error: {}", e), + SyncError::RepoError(e) => write!(f, "Repository error: {}", e), + SyncError::InvalidArgument(e) => write!(f, "Invalid argument: {}", e), + SyncError::ConflictDetected(c) => { + write!(f, "Conflict detected for {}:{}", c.entity_type, c.entity_id) + } + } + } +} + +impl std::error::Error for SyncError {} + +impl From for SyncError { + fn from(e: SyncRepoError) -> Self { + SyncError::RepoError(e) + } +} + +impl From for SyncError { + fn from(e: PdsError) -> Self { + SyncError::PdsError(e) + } +} + +/// Remote record data retrieved from PDS. +#[derive(Debug, Clone)] +pub struct RemoteRecord { + pub uri: String, + pub cid: String, + pub value: serde_json::Value, +} + +/// Sync service for coordinating sync operations. +pub struct SyncService { + sync_repo: Arc, + deck_repo: Arc, + card_repo: Arc, + note_repo: Arc, + oauth_repo: Arc, +} + +impl SyncService { + pub fn new( + sync_repo: Arc, deck_repo: Arc, card_repo: Arc, + note_repo: Arc, oauth_repo: Arc, + ) -> Self { + Self { sync_repo, deck_repo, card_repo, note_repo, oauth_repo } + } + + /// Push a local deck to the user's PDS. + pub async fn push_deck(&self, deck_id: &str, user_ctx: &UserContext) -> Result { + // Log the operation + let log_id = self + .sync_repo + .log_operation(LogOperationParams { + owner_did: &user_ctx.did, + entity_type: "deck", + entity_id: deck_id, + operation: "push", + status: "pending", + pds_cid: None, + error_message: None, + }) + .await?; + + // Get PDS client + let pds_client = self.get_pds_client(user_ctx).await?; + + // Get deck from repository + let deck = self + .deck_repo + .get(deck_id) + .await + .map_err(|e| SyncError::NotFound(format!("Deck not found: {:?}", e)))?; + + // Get cards for the deck + let cards = self + .card_repo + .list_by_deck(deck_id) + .await + .map_err(|e| SyncError::RepoError(SyncRepoError::DatabaseError(format!("{:?}", e))))?; + + // Push cards first, collect AT-URIs + let mut card_at_uris = Vec::with_capacity(cards.len()); + for card in &cards { + let prepared = prepare_card_record(card, ""); // deck_ref filled later + let at_uri = pds_client + .put_record(&user_ctx.did, &prepared.collection, &prepared.rkey, prepared.record) + .await?; + card_at_uris.push(at_uri.to_string()); + + // Mark card as synced + self.sync_repo + .mark_synced("card", &card.id, "", &at_uri.to_string()) + .await?; + } + + // Push deck with card refs + let prepared = prepare_deck_record(&deck, card_at_uris); + let at_uri = pds_client + .put_record(&user_ctx.did, &prepared.collection, &prepared.rkey, prepared.record) + .await?; + + // Mark deck as synced + self.sync_repo + .mark_synced("deck", deck_id, "", &at_uri.to_string()) + .await?; + + let metadata = self.sync_repo.get_sync_metadata("deck", deck_id).await?; + + // Complete log entry + self.sync_repo + .complete_log_entry(&log_id, "success", metadata.pds_cid.as_deref(), None) + .await?; + + Ok(SyncResult { + entity_type: "deck".to_string(), + entity_id: deck_id.to_string(), + pds_uri: Some(at_uri.to_string()), + pds_cid: metadata.pds_cid, + new_version: metadata.version, + status: SyncStatus::Synced, + }) + } + + /// Push a local note to the user's PDS. + pub async fn push_note(&self, note_id: &str, user_ctx: &UserContext) -> Result { + // Log the operation + let log_id = self + .sync_repo + .log_operation(LogOperationParams { + owner_did: &user_ctx.did, + entity_type: "note", + entity_id: note_id, + operation: "push", + status: "pending", + pds_cid: None, + error_message: None, + }) + .await?; + + // Get PDS client + let pds_client = self.get_pds_client(user_ctx).await?; + + // Get note from repository + let note = self + .note_repo + .get(note_id, Some(&user_ctx.did)) + .await + .map_err(|e| SyncError::NotFound(format!("Note not found: {:?}", e)))?; + + let prepared = prepare_note_record(¬e); + let at_uri = pds_client + .put_record(&user_ctx.did, &prepared.collection, &prepared.rkey, prepared.record) + .await?; + + self.sync_repo + .mark_synced("note", note_id, "", &at_uri.to_string()) + .await?; + + let metadata = self.sync_repo.get_sync_metadata("note", note_id).await?; + + // Complete log entry + self.sync_repo + .complete_log_entry(&log_id, "success", metadata.pds_cid.as_deref(), None) + .await?; + + Ok(SyncResult { + entity_type: "note".to_string(), + entity_id: note_id.to_string(), + pds_uri: Some(at_uri.to_string()), + pds_cid: metadata.pds_cid, + new_version: metadata.version, + status: SyncStatus::Synced, + }) + } + + /// Pull a record from the user's PDS. + pub async fn pull_record( + &self, entity_type: &str, at_uri: &str, user_ctx: &UserContext, + ) -> Result { + let parsed = malfestio_core::at_uri::AtUri::parse(at_uri) + .map_err(|e| SyncError::InvalidArgument(format!("Invalid AT-URI: {}", e)))?; + + let log_id = self + .sync_repo + .log_operation(LogOperationParams { + owner_did: &user_ctx.did, + entity_type, + entity_id: at_uri, + operation: "pull", + status: "pending", + pds_cid: None, + error_message: None, + }) + .await?; + + let pds_client = self.get_pds_client(user_ctx).await?; + let result: GetRecordResult = pds_client + .get_record(&parsed.authority, &parsed.collection, &parsed.rkey) + .await + .map_err(|e| { + tracing::error!("Failed to pull record from PDS: {:?}", e); + SyncError::PdsError(e) + })?; + + self.sync_repo + .complete_log_entry(&log_id, "success", Some(&result.cid), None) + .await?; + + // TODO: Offline queue - Store pulled record in IndexedDB for offline access + + Ok(RemoteRecord { uri: result.uri, cid: result.cid, value: result.value }) + } + + /// Check if there's a conflict between local and remote versions. + pub async fn check_conflict( + &self, entity_type: &str, entity_id: &str, remote_cid: &str, + ) -> Result { + let metadata = self.sync_repo.get_sync_metadata(entity_type, entity_id).await?; + + let has_conflict = + metadata.status == SyncStatus::PendingPush && metadata.pds_cid.as_deref() != Some(remote_cid); + + if has_conflict { + self.sync_repo.mark_conflict(entity_type, entity_id).await?; + } + + Ok(has_conflict) + } + + /// Get sync status for a user. + pub async fn get_sync_status(&self, user_ctx: &UserContext) -> Result { + let pending = self.sync_repo.get_pending_items(&user_ctx.did).await?; + let conflicts = self.sync_repo.get_conflicts(&user_ctx.did).await?; + + Ok(SyncStatusSummary { + pending_count: pending.len(), + conflict_count: conflicts.len(), + pending_items: pending.into_iter().map(|p| (p.entity_type, p.entity_id)).collect(), + conflicts: conflicts + .into_iter() + .map(|c| ConflictInfo { + entity_type: c.entity_type, + entity_id: c.entity_id, + local_version: c.version, + remote_version: None, + local_updated_at: None, + remote_updated_at: None, + }) + .collect(), + }) + } + + /// Resolve a conflict using the specified strategy. + pub async fn resolve_conflict( + &self, entity_type: &str, id: &str, strategy: ConflictStrategy, user_ctx: &UserContext, + ) -> Result { + let metadata = self.sync_repo.get_sync_metadata(entity_type, id).await?; + + if metadata.status != SyncStatus::Conflict { + return Err(SyncError::InvalidArgument(format!( + "Entity is not in conflict state: {}:{}", + entity_type, id + ))); + } + + match strategy { + ConflictStrategy::LastWriteWins | ConflictStrategy::KeepLocal => match entity_type { + "deck" => self.push_deck(id, user_ctx).await, + "note" => self.push_note(id, user_ctx).await, + _ => Err(SyncError::InvalidArgument(format!( + "Unknown entity type: {}", + entity_type + ))), + }, + ConflictStrategy::KeepRemote => { + if let Some(pds_uri) = &metadata.pds_uri { + let remote = self.pull_record(entity_type, pds_uri, user_ctx).await?; + + self.sync_repo + .mark_synced(entity_type, id, &remote.cid, &remote.uri) + .await?; + + let new_metadata = self.sync_repo.get_sync_metadata(entity_type, id).await?; + + Ok(SyncResult { + entity_type: entity_type.to_string(), + entity_id: id.to_string(), + pds_uri: Some(remote.uri), + pds_cid: Some(remote.cid), + new_version: new_metadata.version, + status: SyncStatus::Synced, + }) + } else { + Err(SyncError::InvalidArgument("No PDS URI for remote record".to_string())) + } + } + } + } + + async fn get_pds_client(&self, user_ctx: &UserContext) -> Result { + if user_ctx.has_dpop + && let Ok(stored_token) = self.oauth_repo.get_tokens(&user_ctx.did).await + && let Some(dpop_keypair) = stored_token.dpop_keypair() + { + Ok(PdsClient::new_with_dpop( + stored_token.pds_url.clone(), + stored_token.access_token.clone(), + dpop_keypair, + )) + } else { + Ok(PdsClient::new_bearer( + user_ctx.pds_url.clone(), + user_ctx.access_token.clone(), + )) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_conflict_strategy_from_str() { + assert_eq!( + ConflictStrategy::from_str("last_write_wins"), + Ok(ConflictStrategy::LastWriteWins) + ); + assert_eq!( + ConflictStrategy::from_str("keep_local"), + Ok(ConflictStrategy::KeepLocal) + ); + assert_eq!( + ConflictStrategy::from_str("keep_remote"), + Ok(ConflictStrategy::KeepRemote) + ); + assert!(ConflictStrategy::from_str("unknown").is_err()); + } + + #[test] + fn test_conflict_strategy_as_str() { + assert_eq!(ConflictStrategy::LastWriteWins.as_str(), "last_write_wins"); + assert_eq!(ConflictStrategy::KeepLocal.as_str(), "keep_local"); + assert_eq!(ConflictStrategy::KeepRemote.as_str(), "keep_remote"); + } + + #[test] + fn test_sync_error_display() { + let err = SyncError::NotFound("deck:123".to_string()); + assert!(err.to_string().contains("Not found")); + + let err = SyncError::AuthRequired("missing token".to_string()); + assert!(err.to_string().contains("Authentication required")); + + let err = SyncError::InvalidArgument("bad type".to_string()); + assert!(err.to_string().contains("Invalid argument")); + } + + #[test] + fn test_sync_result_creation() { + let result = SyncResult { + entity_type: "deck".to_string(), + entity_id: "123".to_string(), + pds_uri: Some("at://did:plc:test/deck/tid".to_string()), + pds_cid: Some("bafycid".to_string()), + new_version: 2, + status: SyncStatus::Synced, + }; + + assert_eq!(result.entity_type, "deck"); + assert_eq!(result.new_version, 2); + assert_eq!(result.status, SyncStatus::Synced); + } + + #[test] + fn test_sync_status_summary() { + let summary = SyncStatusSummary { + pending_count: 3, + conflict_count: 1, + pending_items: vec![ + ("deck".to_string(), "1".to_string()), + ("note".to_string(), "2".to_string()), + ], + conflicts: vec![ConflictInfo { + entity_type: "deck".to_string(), + entity_id: "3".to_string(), + local_version: 5, + remote_version: Some(6), + local_updated_at: None, + remote_updated_at: None, + }], + }; + + assert_eq!(summary.pending_count, 3); + assert_eq!(summary.conflict_count, 1); + assert_eq!(summary.pending_items.len(), 2); + } + + #[test] + fn test_remote_record_creation() { + let record = RemoteRecord { + uri: "at://did:plc:test/deck/tid".to_string(), + cid: "bafycid123".to_string(), + value: serde_json::json!({"title": "Test"}), + }; + + assert_eq!(record.uri, "at://did:plc:test/deck/tid"); + assert!(record.value.get("title").is_some()); + } +} diff --git a/docs/todo.md b/docs/todo.md index 342a16d..3fbb293 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -30,9 +30,11 @@ **Sync & Conflict Resolution:** -- [ ] Bi-directional sync: local drafts → PDS records, PDS records → local cache -- [ ] Conflict resolution strategy for concurrent edits (last-write-wins or merge UI) +- [x] Bi-directional sync infrastructure +- [x] Conflict resolution strategy +- [ ] API endpoints for sync operations - [ ] Offline queue for pending publishes + - [ ] Frontend sync store with IndexedDB persistence - [ ] Sync status UI indicators **Deep Linking:** diff --git a/migrations/015_2026_01_03_sync_tracking.sql b/migrations/015_2026_01_03_sync_tracking.sql new file mode 100644 index 0000000..0bbfea5 --- /dev/null +++ b/migrations/015_2026_01_03_sync_tracking.sql @@ -0,0 +1,83 @@ +-- Sync tracking infrastructure for bi-directional PDS synchronization +-- Adds version tracking and sync status to core tables + +-- Sync status enum (idempotent creation) +DO $$ BEGIN + CREATE TYPE sync_status AS ENUM ( + 'local_only', -- Never synced to PDS + 'synced', -- In sync with PDS + 'pending_push', -- Local changes need to be pushed + 'conflict' -- Local and remote both changed + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +ALTER TABLE decks + ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 1, + ADD COLUMN IF NOT EXISTS pds_cid TEXT, + ADD COLUMN IF NOT EXISTS pds_uri TEXT, + ADD COLUMN IF NOT EXISTS sync_status sync_status NOT NULL DEFAULT 'local_only', + ADD COLUMN IF NOT EXISTS last_synced_at TIMESTAMPTZ; + +ALTER TABLE cards + ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 1, + ADD COLUMN IF NOT EXISTS pds_cid TEXT, + ADD COLUMN IF NOT EXISTS pds_uri TEXT, + ADD COLUMN IF NOT EXISTS sync_status sync_status NOT NULL DEFAULT 'local_only', + ADD COLUMN IF NOT EXISTS last_synced_at TIMESTAMPTZ; + +ALTER TABLE notes + ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 1, + ADD COLUMN IF NOT EXISTS pds_cid TEXT, + ADD COLUMN IF NOT EXISTS pds_uri TEXT, + ADD COLUMN IF NOT EXISTS sync_status sync_status NOT NULL DEFAULT 'local_only', + ADD COLUMN IF NOT EXISTS last_synced_at TIMESTAMPTZ; + +CREATE TABLE IF NOT EXISTS sync_log ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + owner_did TEXT NOT NULL, + entity_type TEXT NOT NULL, -- 'deck', 'card', 'note' + entity_id UUID NOT NULL, + operation TEXT NOT NULL, -- 'push', 'pull', 'conflict_resolve' + status TEXT NOT NULL, -- 'pending', 'success', 'failed' + pds_cid TEXT, + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_sync_log_owner_did ON sync_log(owner_did); +CREATE INDEX IF NOT EXISTS idx_sync_log_entity ON sync_log(entity_type, entity_id); +CREATE INDEX IF NOT EXISTS idx_sync_log_status ON sync_log(status); +CREATE INDEX IF NOT EXISTS idx_sync_log_created_at ON sync_log(created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_decks_sync_status ON decks(sync_status) WHERE sync_status != 'synced'; +CREATE INDEX IF NOT EXISTS idx_cards_sync_status ON cards(sync_status) WHERE sync_status != 'synced'; +CREATE INDEX IF NOT EXISTS idx_notes_sync_status ON notes(sync_status) WHERE sync_status != 'synced'; + +CREATE OR REPLACE FUNCTION increment_version_on_update() +RETURNS TRIGGER AS $$ +BEGIN + -- Only increment if content changed (not just sync metadata) + IF (TG_TABLE_NAME = 'decks' AND (NEW.title != OLD.title OR NEW.description != OLD.description OR NEW.tags != OLD.tags)) OR + (TG_TABLE_NAME = 'cards' AND (NEW.front != OLD.front OR NEW.back != OLD.back OR NEW.media_url IS DISTINCT FROM OLD.media_url)) OR + (TG_TABLE_NAME = 'notes' AND (NEW.title != OLD.title OR NEW.body != OLD.body OR NEW.tags != OLD.tags)) THEN + NEW.version = OLD.version + 1; + -- Mark as pending push if it was synced + IF OLD.sync_status = 'synced' THEN + NEW.sync_status = 'pending_push'; + END IF; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER increment_decks_version BEFORE UPDATE ON decks + FOR EACH ROW EXECUTE FUNCTION increment_version_on_update(); + +CREATE TRIGGER increment_cards_version BEFORE UPDATE ON cards + FOR EACH ROW EXECUTE FUNCTION increment_version_on_update(); + +CREATE TRIGGER increment_notes_version BEFORE UPDATE ON notes + FOR EACH ROW EXECUTE FUNCTION increment_version_on_update();