feat: 局域网设备同步与冲突合入

crates/workspace-store/examples/workspace-runtime.rs

@@ -152,75 +152,89 @@ fn backups(
 fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     let directory = std::env::args().nth(1).ok_or("missing device directory")?;
     let mut sessions = Sessions::open(Path::new(&directory))?;
+    // A stable fixture device identity; the production store additionally scopes
+    // the journal writer to the selected workspace identity and canonical path.
+    let device = repeaty_workspace::files::digest(directory.as_bytes());
+    sessions.sync_bind_device(&device)?;
     let mut previews = HashMap::new();
     let stdin = io::stdin();
     let mut out = io::stdout().lock();
     for line in stdin.lock().lines() {
         let input: Value = serde_json::from_str(&line?)?;
         let args = &input["args"];
         let epoch = args["epoch"].as_str().unwrap_or("");
-        let result = match input["command"].as_str().unwrap_or("") {
-            "workspace_backups" => {
-                backups(&mut sessions, Path::new(&directory), &mut previews, args)
-            }
-            "workspace_status" => sessions.status(),
-            "workspace_choose" => sessions.choose(
-                epoch,
-                Path::new(args["path"].as_str().unwrap_or("")),
-                args["mode"] == "create",
-            ),
-            "workspace_close" => sessions.close(epoch),
-            "workspace_recent" => sessions.recent(
-                epoch,
-                args["path"].as_str().unwrap_or(""),
-                args["id"].as_str().unwrap_or(""),
-            ),
-            "workspace_restore" => sessions.restore(epoch),
-            "workspace_backup" => sessions
-                .backup(
+        let result = sessions.sync_bind_device(&device).and_then(|()| {
+            let result = match input["command"].as_str().unwrap_or("") {
+                "workspace_sync" => sessions.sync_request(
                     args["token"].as_str().unwrap_or(""),
+                    args["method"].as_str().unwrap_or(""),
+                    args["payload"].clone(),
+                ),
+                "workspace_backups" => {
+                    backups(&mut sessions, Path::new(&directory), &mut previews, args)
+                }
+                "workspace_status" => sessions.status(),
+                "workspace_choose" => sessions.choose(
+                    epoch,
+                    Path::new(args["path"].as_str().unwrap_or("")),
+                    args["mode"] == "create",
+                ),
+                "workspace_close" => sessions.close(epoch),
+                "workspace_recent" => sessions.recent(
+                    epoch,
+                    args["path"].as_str().unwrap_or(""),
+                    args["id"].as_str().unwrap_or(""),
+                ),
+                "workspace_restore" => sessions.restore(epoch),
+                "workspace_backup" => sessions
+                    .backup(
+                        args["token"].as_str().unwrap_or(""),
+                        Path::new(args["path"].as_str().unwrap_or("")),
+                    )
+                    .map(|v| serde_json::to_value(v).unwrap()),
+                "workspace_backup_inspect" => {
+                    Sessions::inspect_backup(Path::new(args["path"].as_str().unwrap_or("")))
+                        .map(|v| serde_json::to_value(v).unwrap())
+                }
+                "workspace_backup_search" => Sessions::search_backup(
                     Path::new(args["path"].as_str().unwrap_or("")),
+                    args["query"].as_str().unwrap_or(""),
+                    args["limit"].as_u64().unwrap_or(30) as usize,
                 )
                 .map(|v| serde_json::to_value(v).unwrap()),
-            "workspace_backup_inspect" => {
-                Sessions::inspect_backup(Path::new(args["path"].as_str().unwrap_or("")))
-                    .map(|v| serde_json::to_value(v).unwrap())
-            }
-            "workspace_backup_search" => Sessions::search_backup(
-                Path::new(args["path"].as_str().unwrap_or("")),
-                args["query"].as_str().unwrap_or(""),
-                args["limit"].as_u64().unwrap_or(30) as usize,
-            )
-            .map(|v| serde_json::to_value(v).unwrap()),
-            "workspace_backup_restore" => sessions.restore_backup(
-                epoch,
-                Path::new(args["path"].as_str().unwrap_or("")),
-                Path::new(args["target"].as_str().unwrap_or("")),
-                args["sha256"].as_str().unwrap_or(""),
-            ),
-            "workspace_backup_restore_items" => Sessions::restore_backup_items(
-                Path::new(args["path"].as_str().unwrap_or("")),
-                Path::new(args["target"].as_str().unwrap_or("")),
-                args["sha256"].as_str().unwrap_or(""),
-                &serde_json::from_value::<Vec<String>>(args["paths"].clone()).unwrap_or_default(),
-            ),
-            "workspace_maintenance" => sessions.maintenance(
-                args["token"].as_str().unwrap_or(""),
-                args["vacuum"].as_bool().unwrap_or(false),
-            ),
-            "workspace_read" => sessions.read(
-                args["token"].as_str().unwrap_or(""),
-                args["request"].clone(),
-            ),
-            "workspace_changes" => sessions
-                .changes(args["token"].as_str().unwrap_or(""))
-                .map(|value| serde_json::to_value(value).unwrap()),
-            "workspace_write" => sessions.write(
-                args["token"].as_str().unwrap_or(""),
-                args["request"].clone(),
-            ),
-            _ => Err(repeaty_workspace::error::Error("invalid-data")),
-        };
+                "workspace_backup_restore" => sessions.restore_backup(
+                    epoch,
+                    Path::new(args["path"].as_str().unwrap_or("")),
+                    Path::new(args["target"].as_str().unwrap_or("")),
+                    args["sha256"].as_str().unwrap_or(""),
+                ),
+                "workspace_backup_restore_items" => Sessions::restore_backup_items(
+                    Path::new(args["path"].as_str().unwrap_or("")),
+                    Path::new(args["target"].as_str().unwrap_or("")),
+                    args["sha256"].as_str().unwrap_or(""),
+                    &serde_json::from_value::<Vec<String>>(args["paths"].clone())
+                        .unwrap_or_default(),
+                ),
+                "workspace_maintenance" => sessions.maintenance(
+                    args["token"].as_str().unwrap_or(""),
+                    args["vacuum"].as_bool().unwrap_or(false),
+                ),
+                "workspace_read" => sessions.read(
+                    args["token"].as_str().unwrap_or(""),
+                    args["request"].clone(),
+                ),
+                "workspace_changes" => sessions
+                    .changes(args["token"].as_str().unwrap_or(""))
+                    .map(|value| serde_json::to_value(value).unwrap()),
+                "workspace_write" => sessions.write(
+                    args["token"].as_str().unwrap_or(""),
+                    args["request"].clone(),
+                ),
+                _ => Err(repeaty_workspace::error::Error("invalid-data")),
+            };
+            sessions.sync_bind_device(&device)?;
+            result
+        });
         let response = match result {
             Ok(value) => json!({"id":input["id"],"value":value}),
             Err(error) => json!({"id":input["id"],"error":error.0}),

crates/workspace-store/src/backup.rs

@@ -657,7 +657,7 @@ fn validate_database_at(root: &Path, database_path: &Path, manifest: &Manifest)
     db.pragma_update(None, "trusted_schema", "OFF")?;
     let app: i64 = db.pragma_query_value(None, "application_id", |r| r.get(0))?;
     let version: u32 = db.pragma_query_value(None, "user_version", |r| r.get(0))?;
-    if app != 0x52505459 || ![10, 11, 12, 13, 14, 15, 16, 17].contains(&version) {
+    if app != 0x52505459 || ![10, 11, 12, 13, 14, 15, 16, 17, 18].contains(&version) {
         return Err(Error("backup-incompatible"));
     }
     let identity: String = db.query_row("SELECT id FROM workspace_identity", [], |r| r.get(0))?;

crates/workspace-store/src/lib.rs

@@ -15,6 +15,7 @@ pub mod library;
 
 pub mod recovery_catalog;
 pub mod sessions;
+pub mod sync;
 pub mod watch;
 
 mod retirement;
@@ -32,3 +33,4 @@ mod p6_tests;
 #[cfg(test)]
 mod p7_tests;
 mod study;
+mod sync_objects;

crates/workspace-store/src/library.rs

@@ -53,6 +53,8 @@ pub struct Save {
     pub operation_history: Option<Value>,
     #[serde(default)]
     pub retired: Option<Vec<Value>>,
+    #[serde(default)]
+    pub sync: Option<crate::sync::Apply>,
     pub consume_card_draft: Option<String>,
     #[serde(default)]
     pub consume_source_draft: Option<String>,
@@ -194,7 +196,7 @@ impl Library {
             .query_row("SELECT id FROM workspace_identity", [], |r| r.get(0))
             .map_err(|_| fail("workspace-incompatible"))?;
         if app != APP_ID
-            || ![10, 11, 12, 13, 14, 15, 16, 17].contains(&version)
+            || ![10, 11, 12, 13, 14, 15, 16, 17, 18].contains(&version)
             || identity != marker.id
         {
             return Err(fail("workspace-incompatible"));
@@ -203,6 +205,7 @@ impl Library {
         let anchor = Arc::new(Dir::open_ambient_dir(&root, cap_std::ambient_authority())?);
         let identity = same_file::Handle::from_file(anchor.try_clone()?.into_std_file())?;
         let db = Database::open(&db_path, Arc::new(anchor.open_dir(".repeaty/media")?))?;
+        crate::sync::schema(&db.connection)?;
         let members = [
             ".repeaty",
             ".repeaty/state.sqlite3",
@@ -456,6 +459,16 @@ impl Library {
             return Err(fail("working-file-changed"));
         }
         working_media::verify(&self.anchor, &request.asset_observations)?;
+        if let Some(sync) = &request.sync {
+            let mut ids = request
+                .patches
+                .iter()
+                .filter(|patch| patch.store == "media" && patch.value.is_some())
+                .map(|patch| patch.id.clone())
+                .collect::<Vec<_>>();
+            ids.extend(sync.retired_media.keys().cloned());
+            self.publish_sync_assets(&ids)?;
+        }
         if let Some(entries) = &request.retired {
             for entry in entries {
                 if let Some(raw) = entry["document"]["raw"].as_str()
@@ -593,9 +606,12 @@ impl Library {
         let anchor = &self.anchor;
         let observations = &request.asset_observations;
         let reactivate = &metadata;
+        let sync = request.sync;
+        let sync_device = self.db.sync_device.clone();
         let revision=self.db.commit_callbacks(Commit {expected_revision:request.expected_revision,patches,reason:request.reason,operation_history:request.operation_history,retired:request.retired},|tx| { working_media::verify(anchor, observations)?; retirement::capture(tx,capture) },|tx| {
             working_media::verify(anchor, observations)?;
             retirement::reactivate(tx,reactivate)?;
+            if let Some(sync) = &sync { crate::sync::apply(tx,sync,sync_device.as_deref())?; }
             if let Some(id)=consume && tx.execute("DELETE FROM card_drafts WHERE id=?1",[id])?!=1 { return Err(Error::conflict()); }
             if let Some(id)=consume_source && tx.execute("DELETE FROM source_drafts WHERE json_extract(data,'$.id')=?1",[id])?!=1 { return Err(Error::conflict()); }
             for (doc,hash) in &documents {
@@ -1157,12 +1173,18 @@ impl Library {
                     return Err(Error::invalid());
                 }
                 if request["value"].is_null() {
-                    self.db.connection.execute(
+                    let tx = self.db.connection.transaction()?;
+                    tx.execute(
                         "DELETE FROM values_store WHERE namespace=?1 AND id=?2",
                         params![ns, id],
                     )?;
+                    crate::sync::flush(&tx, self.db.sync_device.as_deref())?;
+                    tx.commit()?;
                 } else {
-                    self.db.connection.execute("INSERT INTO values_store VALUES(?1,?2,?3) ON CONFLICT(namespace,id) DO UPDATE SET data=excluded.data",params![ns,id,request["value"].to_string()])?;
+                    let tx = self.db.connection.transaction()?;
+                    tx.execute("INSERT INTO values_store VALUES(?1,?2,?3) ON CONFLICT(namespace,id) DO UPDATE SET data=excluded.data",params![ns,id,request["value"].to_string()])?;
+                    crate::sync::flush(&tx, self.db.sync_device.as_deref())?;
+                    tx.commit()?;
                 }
                 Ok(Value::Null)
             }

crates/workspace-store/src/materials.rs

@@ -424,7 +424,7 @@ mod tests {
         );
         assert_eq!(db.load().unwrap()["revision"], 2);
         db.connection
-            .pragma_update(None, "user_version", 18)
+            .pragma_update(None, "user_version", 19)
             .unwrap();
         drop(db);
         assert!(Database::open(&path, assets).is_err());

crates/workspace-store/src/p6_tests.rs

@@ -281,7 +281,7 @@ fn future_reader_gate_is_rejected_by_library_open_and_complete_backup() {
     library
         .db
         .connection
-        .pragma_update(None, "user_version", 18)
+        .pragma_update(None, "user_version", 19)
         .unwrap();
     let archive = directory.path().join("future.repeaty-workspace");
     assert!(backup::create(&library, &archive, &AtomicBool::new(false)).is_err());

crates/workspace-store/src/records.rs

@@ -79,6 +79,7 @@ pub enum ReadAttempts {
 }
 pub struct Database {
     pub(crate) connection: Connection,
+    pub(crate) sync_device: Option<String>,
     assets: Arc<Dir>,
     validators: HashMap<String, jsonschema::Validator>,
 }
@@ -103,7 +104,7 @@ impl Database {
         connection.pragma_update(None, "foreign_keys", "ON")?;
         connection.pragma_update(None, "journal_mode", "WAL")?;
         let version: u32 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        if version > 17 {
+        if version > 18 {
             return Err(Error::invalid());
         }
         let tx = connection.transaction()?;
@@ -170,8 +171,10 @@ impl Database {
             .collect::<Result<_>>()?;
         crate::attachments::schema(&connection)?;
         crate::retirement::history_schema(&connection)?;
+        crate::sync::schema(&connection)?;
         Ok(Self {
             connection,
+            sync_device: None,
             validators,
             assets,
         })
@@ -606,6 +609,7 @@ impl Database {
         }
         tx.execute("UPDATE workspace SET revision=?1 WHERE id=1", [current + 1])?;
         Self::collect_assets(&tx)?;
+        crate::sync::flush(&tx, self.sync_device.as_deref())?;
         tx.commit()?;
         Ok(current + 1)
     }