record node ID and API user for every write

3 files changed, 77 insertions, 14 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index f121ff6..82cac30 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -202,7 +202,9 @@ pub(crate) async fn put_store_entry(
 
     let written = value.clone();
     state
-        .mutate_doc(move |doc| store::write_path(doc, &path, &written))
+        .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            store::write_path(doc, &path, &written)
+        })
         .await
         .map_err(|e| store_status(e, "Failed to write store value"))?;
 
@@ -330,7 +332,9 @@ pub(crate) async fn post_store_entry(
     let index = path.last().and_then(|leaf| store::list_index(leaf));
     let value = body_value(&headers, &body)?;
     let inserted = state
-        .mutate_doc(move |doc| store::insert_path(doc, &path, &value))
+        .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            store::insert_path(doc, &path, &value)
+        })
         .await;
     match inserted {
         Ok(()) => Ok((
@@ -369,7 +373,9 @@ pub(crate) async fn delete_store_entry(
 
     let prune = query.prune;
     state
-        .mutate_doc(move |doc| store::delete_path(doc, &path, prune))
+        .mutate_doc_for(Some(user.0.clone()), move |doc| {
+            store::delete_path(doc, &path, prune)
+        })
         .await
         .map_err(|e| store_status(e, "Failed to delete store value"))?;
 
diff --git a/src/document/replica.rs b/src/document/replica.rs
index df2fafd..3717865 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -175,14 +175,16 @@ impl Replica {
     }
 
     /// The writes `f` makes, as one change stamped `at` if it succeeds (receivers tell
-    /// by the stamp how long the change took to reach them, `swarm::stats`), dropped if
-    /// not. Doesn't refresh the heads: the caller does, under the same lock.
+    /// by the stamp how long the change took to reach them, `swarm::stats`), by
+    /// `writer`, dropped if not. Doesn't refresh the heads: the caller does, under the
+    /// same lock.
     pub(crate) fn transact<T, E>(
         &mut self,
         at: crate::clock::UnixMillis,
+        writer: &Writer,
         f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
     ) -> Result<T, E> {
-        transact(&mut self.doc, at, f)
+        transact(&mut self.doc, at, writer, f)
     }
 }
 
@@ -231,7 +233,11 @@ mod tests {
         let path = vec!["a".to_owned()];
         let mut replica = Replica::new(Epoch::GENESIS, Document::new());
 
-        let refused: Result<(), &str> = replica.transact(UnixMillis(1), |doc| {
+        let writer = Writer {
+            node: "node",
+            user: None,
+        };
+        let refused: Result<(), &str> = replica.transact(UnixMillis(1), &writer, |doc| {
             store::write_path(doc, &path, &json!(1)).unwrap();
             Err("refused")
         });
@@ -242,7 +248,7 @@ mod tests {
 
         // The same writer commits afterwards, and that saves too.
         replica
-            .transact(UnixMillis(2), |doc| {
+            .transact(UnixMillis(2), &writer, |doc| {
                 store::write_path(doc, &path, &json!(2))
             })
             .unwrap();
@@ -261,20 +267,33 @@ mod tests {
     }
 }
 
-/// The writes `f` makes to `doc`, as one change stamped `at` if it succeeds, dropped if
-/// not ([`Replica::transact`], and writes into the document before a rejoin).
+/// Who makes a change, as it records: the node, as the change's author (its node ID),
+/// and the user it wrote for through the API, if any, as its message. Every node holds
+/// every change, so who wrote what is known swarm-wide.
+pub(crate) struct Writer<'a> {
+    pub(crate) node: &'a str,
+    pub(crate) user: Option<&'a str>,
+}
+
+/// The writes `f` makes to `doc`, as one change stamped `at` by `writer` if it succeeds,
+/// dropped if not ([`Replica::transact`], and writes into the document before a rejoin).
 pub(crate) fn transact<T, E>(
     doc: &mut Document,
     at: crate::clock::UnixMillis,
+    writer: &Writer,
     f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
 ) -> Result<T, E> {
+    authored(doc, writer.node);
     let mut transaction = doc.transaction();
     let out = f(&mut transaction);
     match &out {
         Ok(_) => {
-            transaction.commit_with(
-                automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64),
-            );
+            let mut options =
+                automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64);
+            if let Some(user) = writer.user {
+                options.set_message(user);
+            }
+            transaction.commit_with(options);
         }
         Err(_) => {
             transaction.rollback();
@@ -282,3 +301,18 @@ pub(crate) fn transact<T, E>(
     }
     out
 }
+
+/// Makes `node` the author of `doc`'s changes from now on. Setting an author gives the
+/// document a new actor; this node keeps its own (`with_actor`), which it tracks
+/// (`RemakeState::own_actors`, the document file), as only it writes with it.
+fn authored(doc: &mut Document, node: &str) {
+    if doc
+        .get_author()
+        .is_some_and(|author| author.as_bytes() == node.as_bytes())
+    {
+        return;
+    }
+    let actor = doc.get_actor().clone();
+    doc.set_author(Some(automerge::Author::from(node.as_bytes().to_vec())));
+    doc.set_actor(actor);
+}
diff --git a/src/state/mod.rs b/src/state/mod.rs
index cd96aaf..85955fe 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -418,6 +418,20 @@ impl AppState {
         T: Send + 'static,
         E: From<StoreError> + Send + 'static,
     {
+        self.mutate_doc_for(None, f).await
+    }
+
+    /// [`mutate_doc`](Self::mutate_doc) for `user`, who wrote through the API: the
+    /// change names them (`replica::Writer`).
+    pub(crate) async fn mutate_doc_for<T, E>(
+        &self,
+        user: Option<String>,
+        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
+    ) -> Result<T, E>
+    where
+        T: Send + 'static,
+        E: From<StoreError> + Send + 'static,
+    {
         let state = self.clone();
 
         tokio::task::spawn_blocking(move || {
@@ -433,15 +447,24 @@ impl AppState {
                         return Err(E::from(StoreError::Rejoining));
                     };
                     let mut doc = stale.write_or_recover();
+                    let writer = crate::document::replica::Writer {
+                        node: state.local_id(),
+                        user: user.as_deref(),
+                    };
                     return crate::document::replica::transact(
                         &mut doc,
                         state.clock.now_millis(),
+                        &writer,
                         f,
                     );
                 }
                 // One API write = one change; on error, its partial writes (e.g. the old
                 // value deleted, the new one not written) are dropped.
-                let result = replica.transact(state.clock.now_millis(), f);
+                let writer = crate::document::replica::Writer {
+                    node: state.local_id(),
+                    user: user.as_deref(),
+                };
+                let result = replica.transact(state.clock.now_millis(), &writer, f);
                 replica.refresh_heads();
                 result
             }; // lock released before notifying