swarm-controller: wire the config-PR webhook, not just the poll

swarm-controller/src/config_pr.rs

@@ -1,4 +1,5 @@
-//! Swarm-level config-PR status, polled independently of any one hive.
+//! Swarm-level config-PR status, kept current by both a webhook nudge and a
+//! periodic poll.
 //!
 //! `hive-c0re::forge::config_pr_poll` already scans `agent-configs/*` for
 //! open PRs — but it does that once per hive, to queue that hive's own
@@ -7,20 +8,56 @@
 //! have an open config PR" that does not depend on which hive currently
 //! hosts X being reachable.
 //!
-//! This mirrors `hive-c0re`'s poll shape (period, idempotent full rescan)
-//! rather than consuming the swarm-wide `pull_request` webhook
-//! (`crate::webhook::DeliveryKind::ConfigPr`) that already lands here: that
-//! delivery is unparsed today, and building a reliable event path means
-//! building this poll as a missed-delivery backstop anyway (the same reason
-//! `hive-c0re`'s own poller exists). Shipping the backstop alone first
-//! avoids paying for both at once; wiring the webhook as a low-latency nudge
-//! on top is a cheap follow-up once this cache is the source of truth.
+//! Both paths write the same [`ConfigPrCache`]:
+//!
+//! - [`spawn`] — a periodic full rescan, mirroring `hive-c0re`'s own poll
+//!   shape. The backstop: catches anything a missed delivery loses, and is
+//!   what populates the cache before the first delivery ever arrives.
+//! - [`ConfigPrCache::apply_webhook_delivery`] — called from
+//!   `crate::webhook::post_webhook_forge` on a verified `ConfigPr` delivery.
+//!   The low-latency path: a PR opening or closing shows up immediately
+//!   instead of waiting up to [`POLL_INTERVAL`].
+//!
+//! Per mara's review call: ship both from the start rather than the poll
+//! alone — the eventual swarm-level replacement for `hive-c0re`'s own
+//! poll+webhook pair needs both anyway, so building only half here would be
+//! work redone rather than work reused.
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 
+use serde::Deserialize;
+
 use crate::forge::{Client, ConfigPrStatus};
 
+/// The handful of fields this cache needs out of a Forgejo `pull_request`
+/// webhook payload — not a full typed mirror of the event (Forgejo's own
+/// schema has dozens more), just enough to know which agent, which PR, and
+/// whether it's still open.
+#[derive(Deserialize)]
+pub struct ConfigPrWebhookPayload {
+    pull_request: WebhookPullRequest,
+    repository: WebhookRepository,
+}
+
+#[derive(Deserialize)]
+struct WebhookPullRequest {
+    number: u64,
+    /// Forgejo sends `"open"` or `"closed"` here — merged and
+    /// closed-without-merging are indistinguishable at this field, but this
+    /// cache only ever answers "is there an open PR," so the distinction
+    /// doesn't matter to it.
+    state: String,
+    html_url: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct WebhookRepository {
+    /// The config repo's name IS the agent's name — same convention
+    /// `Client::list_open_config_prs` relies on.
+    name: String,
+}
+
 /// How often to rescan `agent-configs/*`. Matches the interval named in
 /// `hive-c0re::forge::config_pr_poll`'s own doc comment — same org, same
 /// staleness tolerance, no reason for the two to disagree.
@@ -62,6 +99,53 @@ impl ConfigPrCache {
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner) = scan;
     }
+
+    /// Apply one verified `ConfigPr` webhook delivery's raw body to the
+    /// cache — the low-latency counterpart to [`spawn`]'s periodic rescan.
+    ///
+    /// A parse failure is logged and dropped, not propagated: the caller
+    /// (`crate::webhook::post_webhook_forge`) already returned 200 to
+    /// Forgejo (HMAC verification, not payload parsing, is what a retry
+    /// could fix), and the next poll tick reconciles whatever this delivery
+    /// would have changed — same "poll as backstop" property [`spawn`]'s
+    /// doc comment describes, just exercised on the failure path instead of
+    /// the steady-state one.
+    ///
+    /// An `open` PR is upserted unconditionally. A `closed` one is removed
+    /// only if the cached entry's PR number still matches — guards against
+    /// an out-of-order delivery (a stale `closed` for PR #1 arriving after a
+    /// newer `opened` for PR #2 on the same repo) wiping out a genuinely
+    /// current entry.
+    pub fn apply_webhook_delivery(&self, body: &[u8]) {
+        let payload: ConfigPrWebhookPayload = match serde_json::from_slice(body) {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    "config-pr webhook: payload did not parse, cache unchanged (next poll reconciles)"
+                );
+                return;
+            }
+        };
+        let mut map = self
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if payload.pull_request.state == "open" {
+            map.insert(
+                payload.repository.name,
+                ConfigPrStatus {
+                    pr_number: payload.pull_request.number,
+                    html_url: payload.pull_request.html_url,
+                },
+            );
+        } else if map
+            .get(&payload.repository.name)
+            .is_some_and(|cached| cached.pr_number == payload.pull_request.number)
+        {
+            map.remove(&payload.repository.name);
+        }
+    }
 }
 
 /// Build an empty cache and spawn the periodic scan that keeps it current.
@@ -89,3 +173,56 @@ pub fn spawn(client: Arc<Client>) -> Arc<ConfigPrCache> {
     });
     cache
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ConfigPrCache;
+
+    fn payload(agent: &str, number: u64, state: &str) -> Vec<u8> {
+        serde_json::json!({
+            "pull_request": { "number": number, "state": state, "html_url": "https://forge.example/pr" },
+            "repository": { "name": agent },
+        })
+        .to_string()
+        .into_bytes()
+    }
+
+    #[test]
+    fn an_open_delivery_upserts_the_entry() {
+        let cache = ConfigPrCache::new();
+        cache.apply_webhook_delivery(&payload("damocles", 5, "open"));
+        let status = cache.get("damocles").expect("entry inserted");
+        assert_eq!(status.pr_number, 5);
+    }
+
+    #[test]
+    fn a_closed_delivery_removes_a_matching_entry() {
+        let cache = ConfigPrCache::new();
+        cache.apply_webhook_delivery(&payload("damocles", 5, "open"));
+        cache.apply_webhook_delivery(&payload("damocles", 5, "closed"));
+        assert!(cache.get("damocles").is_none());
+    }
+
+    #[test]
+    fn a_stale_closed_delivery_does_not_clobber_a_newer_open_pr() {
+        let cache = ConfigPrCache::new();
+        // PR #5 opened, then closed, then a genuinely new PR #6 opens.
+        cache.apply_webhook_delivery(&payload("damocles", 5, "open"));
+        cache.apply_webhook_delivery(&payload("damocles", 6, "open"));
+        // The #5 close event arrives late, after #6 already replaced it.
+        cache.apply_webhook_delivery(&payload("damocles", 5, "closed"));
+        let status = cache.get("damocles").expect("PR #6 must survive");
+        assert_eq!(status.pr_number, 6);
+    }
+
+    #[test]
+    fn a_malformed_payload_leaves_the_cache_unchanged() {
+        let cache = ConfigPrCache::new();
+        cache.apply_webhook_delivery(&payload("damocles", 5, "open"));
+        cache.apply_webhook_delivery(b"not json");
+        let status = cache
+            .get("damocles")
+            .expect("prior entry must survive a bad delivery");
+        assert_eq!(status.pr_number, 5);
+    }
+}