record node ID and API user for every write

1 files changed, 24 insertions, 1 deletions

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