feat: sync tracking persistence layer

crates/server/src/lib.rs

···
 pub mod pds;
 pub mod repository;
 pub mod state;
+pub mod sync_service;
 pub mod well_known;
 
 use axum::http::Method;

crates/server/src/pds/client.rs

···
     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")]
···
     }
 
     /// 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<AtUri, PdsError> {
···
             .map_err(|e| PdsError::NetworkError(e.to_string()))?;
 
         self.handle_response(response).await
+    }
+
+    /// Get a record from the repository.
+    pub async fn get_record(&self, did: &str, collection: &str, rkey: &str) -> Result<GetRecordResult, PdsError> {
+        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.

crates/server/src/repository/mod.rs

···
 pub mod review;
 pub mod search;
 pub mod social;
+pub mod sync;

crates/server/src/repository/sync.rs

···
+//! 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<Self, Self::Err> {
+        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<String>,
+    pub pds_uri: Option<String>,
+    pub status: SyncStatus,
+    pub last_synced_at: Option<DateTime<Utc>>,
+}
+
+/// 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<String>,
+    pub error_message: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub completed_at: Option<DateTime<Utc>>,
+}
+
+/// 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<SyncMetadata, SyncRepoError>;
+
+    /// 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<Vec<PendingSync>, SyncRepoError>;
+
+    /// Get all conflicts for a user.
+    async fn get_conflicts(&self, owner_did: &str) -> Result<Vec<PendingSync>, SyncRepoError>;
+
+    /// Increment version for an entity (used when resolving conflicts).
+    async fn increment_version(&self, entity_type: &str, id: &str) -> Result<i32, SyncRepoError>;
+
+    /// Log a sync operation.
+    async fn log_operation(&self, params: LogOperationParams<'_>) -> Result<String, SyncRepoError>;
+
+    /// 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<SyncMetadata, 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!(
+            "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<String> = row.get("pds_cid");
+        let pds_uri: Option<String> = row.get("pds_uri");
+        let status_str: String = row.get("sync_status");
+        let last_synced_at: Option<DateTime<Utc>> = 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<Vec<PendingSync>, 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<Vec<PendingSync>, 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<i32, 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 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<String, SyncRepoError> {
+        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",
+                &[
+                    &params.owner_did,
+                    &params.entity_type,
+                    &entity_uuid,
+                    &params.operation,
+                    &params.status,
+                    &params.pds_cid,
+                    &params.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<Mutex<HashMap<String, SyncMetadata>>>,
+        logs: Arc<Mutex<Vec<SyncLogEntry>>>,
+    }
+
+    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<SyncMetadata>) -> Self {
+            let map: HashMap<String, SyncMetadata> = 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<SyncMetadata, SyncRepoError> {
+            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<Vec<PendingSync>, 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<Vec<PendingSync>, 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<i32, SyncRepoError> {
+            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<String, SyncRepoError> {
+            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"));
+    }
+}

crates/server/src/sync_service.rs

···
+//! 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<String>,
+    pub pds_cid: Option<String>,
+    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<i32>,
+    pub local_updated_at: Option<String>,
+    pub remote_updated_at: Option<String>,
+}
+
+/// 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<ConflictInfo>,
+}
+
+/// 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<Self, Self::Err> {
+        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<SyncRepoError> for SyncError {
+    fn from(e: SyncRepoError) -> Self {
+        SyncError::RepoError(e)
+    }
+}
+
+impl From<PdsError> 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<dyn SyncRepository>,
+    deck_repo: Arc<dyn DeckRepository>,
+    card_repo: Arc<dyn CardRepository>,
+    note_repo: Arc<dyn NoteRepository>,
+    oauth_repo: Arc<dyn OAuthRepository>,
+}
+
+impl SyncService {
+    pub fn new(
+        sync_repo: Arc<dyn SyncRepository>, deck_repo: Arc<dyn DeckRepository>, card_repo: Arc<dyn CardRepository>,
+        note_repo: Arc<dyn NoteRepository>, oauth_repo: Arc<dyn OAuthRepository>,
+    ) -> 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<SyncResult, SyncError> {
+        // 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<SyncResult, SyncError> {
+        // 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(&note);
+        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<RemoteRecord, SyncError> {
+        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<bool, SyncError> {
+        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<SyncStatusSummary, SyncError> {
+        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<SyncResult, SyncError> {
+        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<PdsClient, SyncError> {
+        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());
+    }
+}

docs/todo.md

···
 
 **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:**

migrations/015_2026_01_03_sync_tracking.sql

···
+-- 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();