fix(node): carry full owner DID on ref-update wire event (#144)

crates/gitlawb-node/src/api/events.rs

@@ -138,9 +138,33 @@ pub async fn list_ref_updates(
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
     let updates = collect_visible_ref_updates(&state.db, None, limit, caller).await?;
 
+    // Resolve the trusted display owner_did per row. The stored wire value is
+    // untrusted (arrives over gossip / unsigned peer notify), so it is echoed
+    // only when it matches the canonical owner of the local repo the slug
+    // names (#P1); legacy None rows are attributed via an exact unique local
+    // match (#P3). Both surfaces share this resolver so they cannot drift.
+    // The batch resolver issues at most one query per distinct local repo
+    // rather than one per event row (#P2).
+    let raw_dids: Vec<Option<String>> = state
+        .db
+        .resolve_ref_update_owner_dids(
+            &updates
+                .iter()
+                .map(|u| (u.repo.as_str(), u.owner_did.as_deref()))
+                .collect::<Vec<_>>(),
+        )
+        .await?;
+
+    let owner_dids: Vec<serde_json::Value> = raw_dids
+        .into_iter()
+        .map(|o| o.map_or(serde_json::Value::Null, |s| serde_json::json!(s)))
+        .collect();
+
     let events: Vec<serde_json::Value> = updates
         .iter()
-        .map(|u| {
+        .enumerate()
+        .map(|(i, u)| {
+            let owner_did = owner_dids[i].clone();
             serde_json::json!({
                 "id":          u.id,
                 "node_did":    u.node_did,
@@ -153,6 +177,7 @@ pub async fn list_ref_updates(
                 "cert_id":     u.cert_id,
                 "received_at": u.received_at,
                 "from_peer":   u.from_peer,
+                "owner_did":   owner_did,
             })
         })
         .collect();
@@ -225,6 +250,7 @@ pub async fn list_repo_events(
                 "pusher_did": c.pusher_did,
                 "node_did":   c.node_did,
                 "timestamp":  c.issued_at,
+                "owner_did":  record.owner_did,
                 "source":     "local",
             })
         })
@@ -240,28 +266,46 @@ pub async fn list_repo_events(
     // collect_visible_ref_updates drops any row whose slug matches a local repo the
     // caller cannot read (fail-closed), and propagates DB errors instead of swallowing
     // them.
-    let gossip_events: Vec<serde_json::Value> =
-        collect_visible_ref_updates(&state.db, Some(&repo_id_str), limit, caller)
-            .await?
-            .iter()
-            .map(|u| {
-                serde_json::json!({
-                    "type":        "gossipsub",
-                    "id":          u.id,
-                    "repo":        u.repo,
-                    "ref_name":    u.ref_name,
-                    "old_sha":     u.old_sha,
-                    "new_sha":     u.new_sha,
-                    "pusher_did":  u.pusher_did,
-                    "node_did":    u.node_did,
-                    "timestamp":   u.timestamp,
-                    "cert_id":     u.cert_id,
-                    "received_at": u.received_at,
-                    "from_peer":   u.from_peer,
-                    "source":      "gossipsub",
-                })
+    let gossip_updates =
+        collect_visible_ref_updates(&state.db, Some(&repo_id_str), limit, caller).await?;
+    let gossip_raw: Vec<Option<String>> = state
+        .db
+        .resolve_ref_update_owner_dids(
+            &gossip_updates
+                .iter()
+                .map(|u| (u.repo.as_str(), u.owner_did.as_deref()))
+                .collect::<Vec<_>>(),
+        )
+        .await?;
+
+    let gossip_owner_dids: Vec<serde_json::Value> = gossip_raw
+        .into_iter()
+        .map(|o| o.map_or(serde_json::Value::Null, |s| serde_json::json!(s)))
+        .collect();
+
+    let gossip_events: Vec<serde_json::Value> = gossip_updates
+        .iter()
+        .enumerate()
+        .map(|(i, u)| {
+            let owner_did = gossip_owner_dids[i].clone();
+            serde_json::json!({
+                "type":        "gossipsub",
+                "id":          u.id,
+                "repo":        u.repo,
+                "ref_name":    u.ref_name,
+                "old_sha":     u.old_sha,
+                "new_sha":     u.new_sha,
+                "pusher_did":  u.pusher_did,
+                "node_did":    u.node_did,
+                "timestamp":   u.timestamp,
+                "cert_id":     u.cert_id,
+                "received_at": u.received_at,
+                "from_peer":   u.from_peer,
+                "owner_did":   owner_did,
+                "source":      "gossipsub",
             })
-            .collect();
+        })
+        .collect();
 
     // Merge both lists
     let mut all_events: Vec<serde_json::Value> = cert_events;
@@ -326,6 +370,7 @@ mod ref_updates_feed_tests {
             cert_id: None,
             received_at: Utc::now().to_rfc3339(),
             from_peer: "peer1".into(),
+            owner_did: None,
         }
     }
 
@@ -1105,7 +1150,14 @@ mod ref_updates_feed_tests {
             .await
             .unwrap();
         assert_eq!(resp.status(), StatusCode::OK);
-        assert_eq!(count(&body_json(resp).await), 2);
+        let body = body_json(resp).await;
+        assert_eq!(count(&body), 2);
+        for event in body["events"].as_array().unwrap() {
+            assert_eq!(
+                event["owner_did"], OWNER,
+                "each event must carry the local owner_did"
+            );
+        }
     }
 
     // Anon reads a quarantined mirror → 404 (withheld without disclosing existence

crates/gitlawb-node/src/api/peers.rs

@@ -347,6 +347,10 @@ pub struct NotifyRequest {
     pub timestamp: Option<String>,
     #[serde(default)]
     pub cert_id: Option<String>,
+    /// Full owner DID — added in #144 for DID-aware feed gating.
+    /// Optional for backward compat with older senders.
+    #[serde(default)]
+    pub owner_did: Option<String>,
 }
 
 pub async fn notify_sync(
@@ -391,6 +395,7 @@ pub async fn notify_sync(
         node_did: req.node_did.clone(),
         pusher_did: req.pusher_did.clone().unwrap_or_default(),
         repo: req.repo.clone(),
+        owner_did: req.owner_did.clone(),
         ref_name: req.ref_name.clone(),
         old_sha: req.old_sha.clone().unwrap_or_default(),
         new_sha: req.new_sha.clone(),

crates/gitlawb-node/src/api/repos.rs

@@ -772,6 +772,7 @@ async fn notify_peer_of_ref(
     new_sha: &str,
     node_did: &str,
     pusher_did: &str,
+    owner_did: &str,
 ) {
     let body = serde_json::json!({
         "repo": repo_slug,
@@ -781,6 +782,7 @@ async fn notify_peer_of_ref(
         "pusher_did": pusher_did,
         "old_sha": old_sha,
         "timestamp": chrono::Utc::now().to_rfc3339(),
+        "owner_did": owner_did,
     });
     let body_bytes = match serde_json::to_vec(&body) {
         Ok(bytes) => bytes,
@@ -828,6 +830,7 @@ async fn notify_peer_of_refs(
     ref_updates: &[(String, String, String)],
     node_did: &str,
     pusher_did: &str,
+    owner_did: &str,
 ) {
     for (ref_name, old_sha, new_sha) in ref_updates {
         notify_peer_of_ref(
@@ -841,6 +844,7 @@ async fn notify_peer_of_refs(
             new_sha,
             node_did,
             pusher_did,
+            owner_did,
         )
         .await;
     }
@@ -1261,6 +1265,7 @@ pub async fn git_receive_pack(
                             node_did: node_did_str.clone(),
                             pusher_did: pusher_did_clone.clone(),
                             repo: repo_slug.clone(),
+                            owner_did: Some(record.owner_did.clone()),
                             ref_name: ref_name.clone(),
                             old_sha: old_sha.clone(),
                             new_sha: new_sha.clone(),
@@ -1292,6 +1297,7 @@ pub async fn git_receive_pack(
                         pusher_did: pusher_did_clone.clone(),
                         node_did: node_did_str.clone(),
                         timestamp: now_ts.clone(),
+                        owner_did: record.owner_did.clone(),
                     });
                 }
             }
@@ -1362,6 +1368,7 @@ pub async fn git_receive_pack(
                             &ref_updates_clone,
                             &node_did_str,
                             &pusher_did_clone,
+                            &record.owner_did,
                         )
                         .await;
                     }
@@ -2402,6 +2409,9 @@ mod tests {
                 mockito::Matcher::PartialJsonString(format!(r#"{{"ref_name":"{ref_a}"}}"#)),
                 mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{old_a}"}}"#)),
                 mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_a}"}}"#)),
+                mockito::Matcher::PartialJsonString(
+                    r#"{"owner_did":"did:key:zOwner"}"#.to_string(),
+                ),
             ]))
             .with_status(200)
             .expect(1)
@@ -2413,6 +2423,9 @@ mod tests {
                 mockito::Matcher::PartialJsonString(format!(r#"{{"ref_name":"{ref_b}"}}"#)),
                 mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{old_b}"}}"#)),
                 mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_b}"}}"#)),
+                mockito::Matcher::PartialJsonString(
+                    r#"{"owner_did":"did:key:zOwner"}"#.to_string(),
+                ),
             ]))
             .with_status(200)
             .expect(1)
@@ -2434,6 +2447,7 @@ mod tests {
             &ref_updates,
             "did:key:zNode",
             "did:key:zPusher",
+            "did:key:zOwner",
         )
         .await;
 
@@ -2456,6 +2470,9 @@ mod tests {
             .match_body(mockito::Matcher::AllOf(vec![
                 mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{zero}"}}"#)),
                 mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_sha}"}}"#)),
+                mockito::Matcher::PartialJsonString(
+                    r#"{"owner_did":"did:key:zOwner"}"#.to_string(),
+                ),
             ]))
             .with_status(200)
             .expect(1)
@@ -2478,6 +2495,7 @@ mod tests {
             &ref_updates,
             "did:key:zNode",
             "did:key:zPusher",
+            "did:key:zOwner",
         )
         .await;