Resolve a {"repo": …} events notify under a store prefix

crates/walgit-server/src/bridge.rs

@@ -14,7 +14,8 @@
 //!   `…/manifest.pb` — the commit point itself as a notification. Accepted
 //!   shapes: a GCS Pub/Sub push envelope (`message.attributes.objectId`), an S3
 //!   event notification (`Records[].s3.object.key`), or a plain `{"key": "…"}`
-//!   / `{"repo": "owner/name"}`. A non-2xx here is meant to be redelivered;
+//!   (all bucket keys, `store.prefix` included) / `{"repo": "owner/name"}`
+//!   (whatever the prefix). A non-2xx here is meant to be redelivered;
 //! * the sweep (`events.sweep_interval`): every repo — the backstop. A sweep
 //!   that finds unpublished entries means the notifications are not flowing
 //!   and says so (`events_bridge_sweep_found_total`, warn).
@@ -258,8 +259,9 @@ impl Bridge {
     }
 }
 
-/// The object keys (or repository ids) a notification body names. Store-agnostic.
-fn notified_keys(v: &serde_json::Value) -> Vec<String> {
+/// The bucket object keys a notification body names, `store.prefix` included as the
+/// notifications carry it. Store-agnostic.
+fn notified_keys(v: &serde_json::Value, store_prefix: &str) -> Vec<String> {
     let mut keys = Vec::new();
     // GCS → Pub/Sub push envelope.
     let attrs = &v["message"]["attributes"];
@@ -299,8 +301,9 @@ fn notified_keys(v: &serde_json::Value) -> Vec<String> {
     if let Some(k) = v["key"].as_str() {
         keys.push(k.to_string());
     }
+    // A repository id names no object: use the key its manifest has in this bucket.
     if let Some(r) = v["repo"].as_str() {
-        keys.push(format!("repos/{r}/manifest.pb"));
+        keys.push(format!("{store_prefix}repos/{r}/manifest.pb"));
     }
     keys
 }
@@ -324,7 +327,7 @@ pub async fn http_notify(
     let v: serde_json::Value = serde_json::from_slice(&bytes)
         .map_err(|e| ApiError::BadRequest(format!("notify body: {e}")))?;
     let mut reports = Vec::new();
-    for key in notified_keys(&v) {
+    for key in notified_keys(&v, &bridge.store_prefix) {
         match bridge.object_finalized(&key).await {
             Ok(Some(report)) => reports.push(report),
             Ok(None) => {}
@@ -390,7 +393,25 @@ mod tests {
             .manifest_repo("prefix/repos/acme/monorepo/manifest.pb")
             .unwrap();
         assert_eq!(id.to_string(), "acme/monorepo");
+        // Every body shape names the same repository under a non-empty prefix.
+        for body in [
+            serde_json::json!({"repo": "acme/monorepo"}),
+            serde_json::json!({"key": "prefix/repos/acme/monorepo/manifest.pb"}),
+            serde_json::json!({"message": {"attributes": {
+                "eventType": "OBJECT_FINALIZE",
+                "objectId": "prefix/repos/acme/monorepo/manifest.pb"}}}),
+            serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put",
+                "s3": {"object": {"key": "prefix/repos/acme/monorepo/manifest.pb"}}}]}),
+        ] {
+            let ids: Vec<_> = notified_keys(&body, &b.store_prefix)
+                .iter()
+                .filter_map(|k| b.manifest_repo(k))
+                .map(|id| id.to_string())
+                .collect();
+            assert_eq!(ids, ["acme/monorepo"], "{body}");
+        }
         for other in [
+            "repos/t/r/manifest.pb",
             "prefix/repos/t/r/wal/abc.pack",
             "prefix/repos/t/r/events/cursor.json",
             "walgit-go/repos/t/r/manifest.pb",

crates/walgit-server/tests/events.rs

@@ -206,6 +206,73 @@ async fn bridge_publishes_from_cursor_exactly_once() -> TestResult {
     Ok(())
 }
 
+/// With a bucket prefix (`store.prefix`), every notification shape wakes the bridge: bucket
+/// notifications carry the prefixed object key, and a plain `{"repo": …}` names the repository
+/// whatever the prefix. A key outside this walgit's prefix is acked and ignored.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn notify_wakes_the_bridge_under_a_store_prefix() -> TestResult {
+    use walgit_store::ObjectStoreExt;
+    let (url, captured) = webhook().await;
+    let server = Server::start_with_tweak(|c| {
+        bridge_cfg(&url, Duration::ZERO)(c);
+        c.store.prefix = "tenant-a".into();
+    })
+    .await?;
+    server.put_repo("t", "r").await?;
+    assert!(
+        server
+            .store
+            .exists("tenant-a/repos/t/r/manifest.pb")
+            .await?,
+        "the repository lives under the prefix"
+    );
+    let src = TestRepo::synthetic(1, 1)?;
+    git_in(&src, &["commit", "--allow-empty", "-m", "a"])?;
+    git_in(&src, &["branch", "-M", "main"])?;
+    git_in(
+        &src,
+        &["remote", "add", "origin", &server.repo_url("t", "r")],
+    )?;
+    git_in(&src, &["push", "-u", "origin", "main"])?;
+
+    let client = reqwest::Client::new();
+    let notify = |body: serde_json::Value| {
+        let client = client.clone();
+        let url = format!("{}/_events/notify", server.base_url);
+        async move {
+            let resp = client.post(url).json(&body).send().await?;
+            anyhow::ensure!(resp.status() == 200, "{body}: {}", resp.status());
+            anyhow::Ok(resp.json::<serde_json::Value>().await?)
+        }
+    };
+    // Outside the prefix: not this walgit's object.
+    let report = notify(serde_json::json!({"key": "repos/t/r/manifest.pb"})).await?;
+    assert_eq!(report, serde_json::json!([]));
+    assert!(captured.lock().unwrap().is_empty());
+
+    let mut events = 0;
+    for body in [
+        serde_json::json!({"repo": "t/r"}),
+        gcs_notification("tenant-a/repos/t/r/manifest.pb", "OBJECT_FINALIZE"),
+        serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put", "s3": {"object": {"key": "tenant-a/repos/t/r/manifest.pb"}}}]}),
+        serde_json::json!({"key": "tenant-a/repos/t/r/manifest.pb"}),
+    ] {
+        git_in(&src, &["commit", "--allow-empty", "-m", "next"])?;
+        git_in(&src, &["push"])?;
+        events += 1;
+        let report = notify(body.clone()).await?;
+        assert_eq!(report[0]["repo"], "t/r", "{body}: {report}");
+        assert_eq!(
+            report[0]["emitted"],
+            if events == 1 { 2 } else { 1 },
+            "{body}: {report}"
+        );
+    }
+    let got = wait_for(&captured, 5).await;
+    assert_eq!(got.len(), 5);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn bridge_sweep_timer_publishes_without_notifications() -> TestResult {
     let (url, captured) = webhook().await;

crates/walgit-server/tests/harness.rs

@@ -109,7 +109,13 @@ impl Server {
         cfg.server.public_url = Some(base_url.clone());
         cfg.validate().context("config validate")?;
 
-        let dyn_store: DynStore = store.clone();
+        // `store.prefix` scopes every key, as `walgit_store::from_config` does for real backends.
+        let prefix = cfg.store_prefix();
+        let dyn_store: DynStore = if prefix.is_empty() {
+            store.clone()
+        } else {
+            Arc::new(walgit_store::Prefixed::new(store.clone(), prefix))
+        };
         let state = AppState::new(Arc::new(cfg), dyn_store).await?;
 
         let registry = state.registry.clone();

docs/EVENTS.md

@@ -97,7 +97,9 @@ Wake-ups (both idempotent; they only ever call `catch_up`):
   itself as the notification. Accepted bodies: a GCS Pub/Sub push envelope (`message.attributes.eventType =
   OBJECT_FINALIZE`, `objectId`), an S3 event notification (`Records[].eventName = ObjectCreated:*`,
   `s3.object.key`; MinIO, rustfs and Ceph emit the same shape), or your own glue's `{"key": "repos/o/r/manifest.pb"}`
-  / `{"repo": "o/r"}`. Everything else is acked and ignored; a webhook failure answers 503 so the notifier
+  / `{"repo": "o/r"}`. Keys are bucket object keys as notifications carry them, `store.prefix` included
+  (`{"key": "<prefix>/repos/o/r/manifest.pb"}`); `{"repo": …}` names the repository whatever the prefix.
+  Everything else is acked and ignored; a webhook failure answers 503 so the notifier
   redelivers. Authenticated like every route (`require_read`): give the notifier a token.
 - The sweep (`events.sweep_interval`, default 5 min): `list` + one conditional manifest GET per repo. Not needed
   for correctness; it is the backstop *and the health check* — a sweep that publishes anything means