gunnar-server 2.0.0

//! The event stream's OUTBOUND leg: deliver each event to an external URL (webhooks).
//!
//! ```text
//!   git push ──▶ gunnar ──▶ event bus ──▶ this subscriber ──▶ per-endpoint queue ──▶ POST ──▶ your CI
//!                   │                                                                   │
//!                   └── the client is already done here ──────────────────────────────┘
//! ```
//!
//! # It is a subscriber to the SAME bus, and that is the whole architecture
//!
//! A webhook is a consumer of the event broadcast bus, exactly as the
//! replicator ([`crate::mirror`]) and the console's SSE fan-out are (one implementation). It
//! is NOT a second event source: nothing in the push path calls into this
//! module, `receive-pack` publishes to the bus as it always did, and
//! [`WebhookDelivery::spawn`] subscribes. So the push path cannot be made
//! slower, or made to fail, by a webhook — which is the same hard requirement
//! replication has, for the same reason: a push that succeeded locally has
//! succeeded, whatever a subscriber is doing with the news.
//!
//! Three things follow from being a subscriber:
//!
//! * `EventBus::publish` is a `broadcast::send`: it never blocks and never
//!   fails, with or without a webhook attached.
//! * a delivery that cannot keep up **lags the bus**, and the bus answers a
//!   lagging subscriber with an `EventGap` rather than with memory. A gap is
//!   surfaced per endpoint as a `dropped` count — the receiver missed events
//!   and should reconcile over the API, the same drop-with-a-gap contract every
//!   event-stream subscriber gets.
//! * the POST runs on a **background worker**, off the broadcast receiver. The
//!   subscribe task only ever ENQUEUES, into a bounded queue, and returns. A
//!   dead endpoint therefore cannot wedge the bus: the worker blocks on the
//!   socket, the subscribe task does not, and past the queue bound events are
//!   dropped-with-a-gap rather than buffered without limit. A webhook that
//!   wedges the server is worse than no webhook.
//!
//! # The queue bound
//!
//! Each endpoint has its own bounded queue ([`DEFAULT_QUEUE_CAPACITY`]). Past
//! the bound the NEWEST event is refused and the endpoint's `dropped` counter
//! climbs: a receiver that has fallen a thousand events behind is not a
//! queueing problem, it is a receiver that is down, and the answer to a down
//! receiver is a bounded gap it can reconcile from, not an unbounded queue that
//! turns one slow endpoint into an OOM. That is the shape that OOM'd a machine
//! in this tree on 2026-08-03, and it is designed out here by construction.
//!
//! # Signed, and why
//!
//! Each delivery is signed with an `sshsig` under the `gunnar-webhook`
//! namespace ([`gunnar_core::webhook`]), over the exact request body. The
//! receiver verifies with the server's published public key and no shared
//! secret — see the module docs there. An optional HMAC header is set when an
//! endpoint configures a secret, for GitHub-webhook-compatible receivers.

use std::collections::VecDeque;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use anyhow::{Context, Result};
use gunnar_core::ssh_key;
use gunnar_grpc::{EventBus, EventFilter};
use gunnar_proto::{event, Event, EventKind, RepoPath};
use serde::{Deserialize, Serialize};

use crate::stores::ServerStores;

/// The on-disk configuration file, in the data root beside `replication.json`.
pub const WEBHOOK_CONFIG: &str = "webhooks.json";

/// Per-endpoint queue bound. See the module header.
pub const DEFAULT_QUEUE_CAPACITY: usize = 1024;

/// First retry delay and the ceiling it doubles up to. The backoff ladder
/// itself is [`crate::durable::backoff`], shared with the replicator (one implementation).
const RETRY_BASE: Duration = Duration::from_secs(2);
const RETRY_MAX: Duration = Duration::from_secs(300);

/// **A delivery attempt is bounded.** Unlike a replication job — where a dead
/// standby retries forever because the second copy still has to be made — a
/// missed webhook is a missed notification, and retrying a dead endpoint
/// forever with a growing queue is the wedge this module exists to avoid. After
/// this many failures the event is dropped-with-a-gap (the `dropped` counter),
/// the receiver reconciles over the API, and the queue keeps draining.
const MAX_ATTEMPTS: u32 = 5;

/// Per-attempt ceiling on the whole POST, so a black-holing endpoint costs one
/// timeout rather than a stuck worker.
const DELIVER_TIMEOUT: Duration = Duration::from_secs(15);

const TARGET_WEBHOOK: &str = "gunnar::webhook";

// ───────────────────────────────────────────────────────────────────────────
// Configuration — durable, same shape and hazards as replication.json
// ───────────────────────────────────────────────────────────────────────────

/// The set of configured endpoints. One JSON document in the data root.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WebhookConfig {
    #[serde(default)]
    pub endpoints: Vec<Endpoint>,
}

/// One outbound endpoint, as an operator declares it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Endpoint {
    /// The operator's name for it — appears in every log line and status row.
    pub name: String,
    /// Where deliveries are POSTed. `http://` or `https://`.
    pub url: String,
    /// Event-kind names to deliver (e.g. `ref.updated`). Empty = every kind.
    #[serde(default)]
    pub kinds: Vec<String>,
    /// Deliver only events about this namespace. Empty = every namespace.
    #[serde(default)]
    pub namespace: String,
    /// Deliver only events about this repository. `None` = every repository.
    #[serde(default)]
    pub repo: Option<RepoPath>,
    /// Optional HMAC shared secret for `X-Hub-Signature-256`. Stored, never
    /// returned over the control plane.
    #[serde(default)]
    pub secret: Option<String>,
    #[serde(default)]
    pub added_at_unix_ms: i64,
}

impl WebhookConfig {
    /// Read `<root>/webhooks.json`, or the empty configuration when it is
    /// absent. A missing file is not an error — most servers have no webhooks —
    /// but a **malformed** one is, so a typo cannot silently deliver nowhere.
    /// An empty or whitespace-only file is the power-cut torn-write artifact and is
    /// treated as "no webhooks configured", exactly as `replication.json` is.
    pub fn load(root: impl AsRef<Path>) -> Result<Self> {
        let path = root.as_ref().join(WEBHOOK_CONFIG);
        let text = match std::fs::read_to_string(&path) {
            Ok(text) => text,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                return Ok(WebhookConfig::default())
            }
            Err(err) => return Err(err).with_context(|| format!("reading {}", path.display())),
        };
        if text.trim().is_empty() {
            tracing::warn!(
                target: TARGET_WEBHOOK,
                event = "webhook.config_empty",
                path = %path.display(),
                "webhooks.json is empty — the signature of a power cut during a save. Treating \
                 it as no webhooks configured and coming up serving; it is rewritten on the next \
                 apply."
            );
            return Ok(WebhookConfig::default());
        }
        serde_json::from_str(&text).with_context(|| {
            format!(
                "parsing {}; gunnar will not deliver to a misconfigured endpoint because of a \
                 syntax error",
                path.display()
            )
        })
    }

    /// Write the configuration back, atomically and durably, through
    /// the shared writer `replication.json` uses (one implementation).
    pub fn save(&self, root: impl AsRef<Path>) -> Result<()> {
        let text = serde_json::to_string_pretty(self)?;
        crate::durable::write_atomic(root.as_ref(), WEBHOOK_CONFIG, text.as_bytes())
    }

    /// [`WebhookConfig::save`], **and a copy in the metadata mirror on the slow
    /// disk** — the same treatment `replication.json` gets, for the same
    /// reason: the endpoints an operator configured are a handful of bytes, and
    /// losing them with the stripe means nobody downstream is told anything
    /// until somebody remembers to reconfigure it by hand.
    pub fn save_mirrored(
        &self,
        root: impl AsRef<Path>,
        slow: &crate::slow::SlowRoot,
    ) -> Result<()> {
        let text = serde_json::to_string_pretty(self)?;
        crate::durable::write_atomic_mirrored(root.as_ref(), slow, WEBHOOK_CONFIG, text.as_bytes())
    }

    /// Add or replace an endpoint by name. Replace, so a retried `add` on a
    /// flaky link does not leave two.
    pub fn set_endpoint(&mut self, endpoint: Endpoint) {
        match self.endpoints.iter_mut().find(|e| e.name == endpoint.name) {
            Some(existing) => *existing = endpoint,
            None => self.endpoints.push(endpoint),
        }
    }

    /// Remove an endpoint by name. `true` when one was actually removed.
    pub fn remove_endpoint(&mut self, name: &str) -> bool {
        let before = self.endpoints.len();
        self.endpoints.retain(|e| e.name != name);
        self.endpoints.len() != before
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Event-kind names — the stable strings a `--kinds` filter and the payload use
// ───────────────────────────────────────────────────────────────────────────

/// The stable wire name of an event kind, e.g. `ref.updated`. This is what a
/// `--kinds` filter names and what the delivered body's `kind` field carries.
pub fn kind_name(kind: EventKind) -> &'static str {
    match kind {
        EventKind::Unspecified => "unspecified",
        EventKind::RefUpdated => "ref.updated",
        EventKind::RefDeleted => "ref.deleted",
        EventKind::PushRejected => "push.rejected",
        EventKind::RepoCreated => "repo.created",
        EventKind::RepoDeleted => "repo.deleted",
        EventKind::MirrorSynced => "mirror.synced",
        EventKind::KeyUsed => "key.used",
        EventKind::PushAttestation => "push.attestation",
        EventKind::Gap => "gap",
        EventKind::ReplicationFailed => "replication.failed",
        EventKind::PullRequestOpened => "pull_request.opened",
        EventKind::PullRequestMerged => "pull_request.merged",
        EventKind::PullRequestClosed => "pull_request.closed",
        EventKind::StreamOpened => "stream.opened",
        EventKind::VisibilityChanged => "repo.visibility_changed",
        EventKind::EncryptedChanged => "repo.encrypted_changed",
    }
}

/// The kind a `--kinds` name selects, or `None` for a name that matches no
/// kind — which the control plane rejects at add time, because a typo here is a
/// webhook that silently delivers nothing.
pub fn kind_from_name(name: &str) -> Option<EventKind> {
    let name = name.trim();
    [
        EventKind::RefUpdated,
        EventKind::RefDeleted,
        EventKind::PushRejected,
        EventKind::RepoCreated,
        EventKind::RepoDeleted,
        EventKind::MirrorSynced,
        EventKind::KeyUsed,
        EventKind::PushAttestation,
        EventKind::ReplicationFailed,
        EventKind::PullRequestOpened,
        EventKind::PullRequestMerged,
        EventKind::PullRequestClosed,
        EventKind::VisibilityChanged,
        EventKind::EncryptedChanged,
    ]
    .into_iter()
    .find(|k| kind_name(*k) == name)
}

/// The delivered request body: a stable JSON object over the event.
///
/// Built by hand rather than from a prost serialisation, for the same reason
/// provenance signs canonical bytes: the receiver reads THESE bytes and the
/// signature is over them, so the shape has to be one gunnar controls and not
/// whatever a serialiser happens to emit. Every delivery carries `kind`, `seq`
/// and `at_unix_ms`; repo-scoped kinds carry `namespace`/`repo`; and the common
/// kinds carry the fields a CI trigger actually reads.
pub fn event_body(event: &Event) -> Vec<u8> {
    let kind = EventKind::try_from(event.kind).unwrap_or(EventKind::Unspecified);
    let mut obj = serde_json::Map::new();
    obj.insert("kind".into(), kind_name(kind).into());
    obj.insert("seq".into(), event.seq.into());
    obj.insert("at_unix_ms".into(), event.at_unix_ms.into());
    if let Some(path) = event.repo_path() {
        obj.insert("namespace".into(), path.namespace.clone().into());
        obj.insert("repo".into(), path.repo.clone().into());
    }
    let mut payload = serde_json::Map::new();
    match event.payload.as_ref() {
        Some(event::Payload::RefUpdated(e)) => {
            payload.insert("ref".into(), e.ref_name.clone().into());
            payload.insert("old_oid".into(), e.old_oid.clone().into());
            payload.insert("new_oid".into(), e.new_oid.clone().into());
            payload.insert("principal".into(), e.principal_id.clone().into());
            payload.insert("forced".into(), e.forced.into());
        }
        Some(event::Payload::RefDeleted(e)) => {
            payload.insert("ref".into(), e.ref_name.clone().into());
            payload.insert("old_oid".into(), e.old_oid.clone().into());
            payload.insert("principal".into(), e.principal_id.clone().into());
        }
        Some(event::Payload::RepoDeleted(e)) => {
            payload.insert("principal".into(), e.principal_id.clone().into());
        }
        Some(event::Payload::RepoCreated(e)) => {
            payload.insert("principal".into(), e.principal_id.clone().into());
        }
        Some(event::Payload::MirrorSynced(e)) => {
            payload.insert("remote_url".into(), e.remote_url.clone().into());
            payload.insert("refs_changed".into(), e.refs_changed.into());
        }
        Some(event::Payload::PushAttestation(e)) => {
            payload.insert("attestation_id".into(), e.attestation_id.clone().into());
            payload.insert("new_object_count".into(), e.new_object_count.into());
            payload.insert("new_object_bytes".into(), e.new_object_bytes.into());
            payload.insert("pack_hash".into(), e.pack_hash.clone().into());
        }
        Some(event::Payload::VisibilityChanged(e)) => {
            let name = |v: i32| match gunnar_proto::Visibility::try_from(v) {
                Ok(gunnar_proto::Visibility::Public) => "public",
                Ok(gunnar_proto::Visibility::Private) => "private",
                _ => "unspecified",
            };
            payload.insert("from".into(), name(e.from).into());
            payload.insert("to".into(), name(e.to).into());
            payload.insert("principal".into(), e.principal_id.clone().into());
        }
        Some(event::Payload::EncryptedChanged(e)) => {
            payload.insert("from".into(), e.from.into());
            payload.insert("to".into(), e.to.into());
            payload.insert("principal".into(), e.principal_id.clone().into());
        }
        Some(event::Payload::PullRequestOpened(_))
        | Some(event::Payload::PullRequestMerged(_))
        | Some(event::Payload::PullRequestClosed(_)) => {}
        _ => {}
    }
    obj.insert("payload".into(), serde_json::Value::Object(payload));
    // to_vec, not to_vec_pretty: the bytes are what the signature covers, and a
    // compact stable rendering is what the receiver hashes.
    serde_json::to_vec(&serde_json::Value::Object(obj)).unwrap_or_default()
}

/// The `EventFilter` that decides which events reach one endpoint.
fn filter_for(endpoint: &Endpoint) -> EventFilter {
    EventFilter {
        namespaces: if endpoint.namespace.is_empty() {
            Vec::new()
        } else {
            vec![endpoint.namespace.clone()]
        },
        repos: endpoint.repo.clone().into_iter().collect(),
        kinds: endpoint
            .kinds
            .iter()
            .filter_map(|k| kind_from_name(k))
            .collect(),
        include_attestation_object_ids: false,
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The running delivery: one subscriber, one bounded queue + worker per endpoint
// ───────────────────────────────────────────────────────────────────────────

/// Per-endpoint counters, surfaced in `webhook list`.
#[derive(Debug, Default)]
struct Counters {
    delivered: AtomicU64,
    failed: AtomicU64,
    dropped: AtomicU64,
    last_error: Mutex<String>,
}

/// One endpoint's queue, worker wake-up and counters.
struct Runner {
    endpoint: Endpoint,
    filter: EventFilter,
    queue: Mutex<VecDeque<Arc<Event>>>,
    notify: tokio::sync::Notify,
    counters: Counters,
}

impl Runner {
    fn new(endpoint: Endpoint) -> Arc<Self> {
        Arc::new(Runner {
            filter: filter_for(&endpoint),
            endpoint,
            queue: Mutex::new(VecDeque::new()),
            notify: tokio::sync::Notify::new(),
            counters: Counters::default(),
        })
    }

    /// Enqueue an event for delivery. **Never blocks** — that is what keeps the
    /// bus off the socket. Past the bound the newest is dropped-with-a-gap.
    fn enqueue(&self, event: Arc<Event>) {
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        if queue.len() >= DEFAULT_QUEUE_CAPACITY {
            drop(queue);
            self.counters.dropped.fetch_add(1, Ordering::Relaxed);
            tracing::warn!(
                target: TARGET_WEBHOOK,
                event = "webhook.queue_full",
                endpoint = %self.endpoint.name,
                "the per-endpoint queue is full; dropping the newest event. The receiver is \
                 behind and should reconcile over the API."
            );
            return;
        }
        queue.push_back(event);
        drop(queue);
        self.notify.notify_one();
    }

    fn pop(&self) -> Option<Arc<Event>> {
        self.queue
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .pop_front()
    }

    #[cfg_attr(not(test), allow(dead_code))]
    fn queue_len(&self) -> usize {
        self.queue.lock().unwrap_or_else(|e| e.into_inner()).len()
    }
}

/// Handles for the spawned tasks; aborting them stops delivery.
pub struct WebhookTasks {
    tasks: Vec<tokio::task::JoinHandle<()>>,
}

impl WebhookTasks {
    pub fn abort(&self) {
        for task in &self.tasks {
            task.abort();
        }
    }
    pub fn is_empty(&self) -> bool {
        self.tasks.is_empty()
    }
}

impl Drop for WebhookTasks {
    fn drop(&mut self) {
        self.abort();
    }
}

/// The running webhook delivery: the signing key, the bus, and one runner per
/// endpoint.
pub struct WebhookDelivery {
    bus: Arc<EventBus>,
    signing_key: Arc<ssh_key::PrivateKey>,
    config: WebhookConfig,
    runners: Vec<Arc<Runner>>,
}

impl WebhookDelivery {
    pub fn new(
        bus: Arc<EventBus>,
        signing_key: Arc<ssh_key::PrivateKey>,
        config: WebhookConfig,
    ) -> Arc<Self> {
        let runners = config
            .endpoints
            .iter()
            .cloned()
            .map(Runner::new)
            .collect();
        Arc::new(WebhookDelivery {
            bus,
            signing_key,
            config,
            runners,
        })
    }

    pub fn config(&self) -> &WebhookConfig {
        &self.config
    }

    /// Start the event-stream subscriber and one worker per endpoint.
    pub fn spawn(self: &Arc<Self>) -> WebhookTasks {
        let mut tasks = Vec::new();
        for runner in &self.runners {
            let me = Arc::clone(self);
            let runner = Arc::clone(runner);
            // A32-14, stays a task: the webhook workers; their handles are kept and aborted at the dispatcher's stop.
            tasks.push(tokio::spawn(async move { worker(me, runner).await }));
        }
        let me = Arc::clone(self);
        tasks.push(tokio::spawn(async move { subscribe(me).await }));
        WebhookTasks { tasks }
    }

    /// The endpoints, with live counters, as the control plane reports them.
    fn endpoint_rows(&self) -> Vec<gunnar_proto::WebhookEndpoint> {
        self.runners
            .iter()
            .map(|r| gunnar_proto::WebhookEndpoint {
                name: r.endpoint.name.clone(),
                url: r.endpoint.url.clone(),
                kinds: r.endpoint.kinds.clone(),
                namespace: r.endpoint.namespace.clone(),
                repo: r.endpoint.repo.clone(),
                has_secret: r.endpoint.secret.is_some(),
                added_at_unix_ms: r.endpoint.added_at_unix_ms,
                delivered: r.counters.delivered.load(Ordering::Relaxed),
                failed: r.counters.failed.load(Ordering::Relaxed),
                dropped: r.counters.dropped.load(Ordering::Relaxed),
                last_error: r
                    .counters
                    .last_error
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .clone(),
            })
            .collect()
    }
}

/// The event-stream subscriber. **The only coupling to the push path**, and it only
/// ever enqueues — never delivers — so the bus is never behind a socket.
async fn subscribe(delivery: Arc<WebhookDelivery>) {
    use tokio_stream::StreamExt;
    // Watch everything and fan out per endpoint's filter. One subscription for
    // all endpoints keeps the bus's subscriber count — and its lag accounting —
    // bounded by the server, not by how many webhooks an operator configured.
    let mut stream = Box::pin(delivery.bus.subscribe(EventFilter::default()));
    while let Some(next) = stream.next().await {
        let event = match next {
            Ok(event) => event,
            Err(status) => {
                tracing::warn!(target: TARGET_WEBHOOK, %status, "webhook subscription error; continuing");
                continue;
            }
        };
        if let Some(event::Payload::Gap(gap)) = event.payload.as_ref() {
            // A lagged subscriber lost events. Every endpoint missed them; count
            // it so `webhook list` shows a non-zero `dropped` and the receiver
            // knows to reconcile. This is the bus's drop-with-a-gap contract,
            // surfaced per endpoint.
            for runner in &delivery.runners {
                runner
                    .counters
                    .dropped
                    .fetch_add(gap.dropped, Ordering::Relaxed);
            }
            tracing::error!(
                target: TARGET_WEBHOOK,
                event = "webhook.bus_gap",
                dropped = gap.dropped,
                "the webhook subscriber fell behind the event bus and lost events; every endpoint \
                 is credited a gap"
            );
            continue;
        }
        // One `Arc` shared across every matching endpoint's queue, so a push
        // watched by ten webhooks is one allocation and not ten.
        let event = Arc::new(event);
        for runner in &delivery.runners {
            if runner.filter.matches(&event) {
                runner.enqueue(Arc::clone(&event));
            }
        }
    }
    tracing::warn!(target: TARGET_WEBHOOK, "the event stream ended; no more deliveries");
}

/// One endpoint's delivery loop.
async fn worker(delivery: Arc<WebhookDelivery>, runner: Arc<Runner>) {
    loop {
        let Some(event) = runner.pop() else {
            runner.notify.notified().await;
            continue;
        };
        deliver_with_retry(&delivery, &runner, &event).await;
    }
}

/// Deliver one event, retrying with bounded backoff. A dead endpoint costs at
/// most [`MAX_ATTEMPTS`] tries and then drops-with-a-gap — it never blocks the
/// worker forever, and it never grows the queue (the queue keeps draining while
/// this awaits its own sleeps).
async fn deliver_with_retry(delivery: &WebhookDelivery, runner: &Runner, event: &Event) {
    let body = event_body(event);
    let kind = EventKind::try_from(event.kind).unwrap_or(EventKind::Unspecified);
    let signed = match gunnar_core::webhook::sign_delivery(&delivery.signing_key, &body) {
        Ok(signed) => signed,
        Err(err) => {
            // Signing is the point of a gunnar webhook; an UNSIGNED delivery is
            // not a fallback we ship. Count it failed and drop it.
            runner.counters.failed.fetch_add(1, Ordering::Relaxed);
            *runner.counters.last_error.lock().unwrap_or_else(|e| e.into_inner()) =
                format!("signing the delivery failed: {err}");
            tracing::error!(target: TARGET_WEBHOOK, endpoint = %runner.endpoint.name, %err, "could not sign a delivery; dropping it");
            return;
        }
    };
    let delivery_id = format!("{}:{}", delivery.bus.epoch(), event.seq);

    for attempt in 1..=MAX_ATTEMPTS {
        let result = tokio::time::timeout(
            DELIVER_TIMEOUT,
            post(&runner.endpoint, &body, &signed, kind, &delivery_id),
        )
        .await;
        match result {
            Ok(Ok(status)) if (200..300).contains(&status) => {
                runner.counters.delivered.fetch_add(1, Ordering::Relaxed);
                runner
                    .counters
                    .last_error
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .clear();
                return;
            }
            Ok(Ok(status)) => {
                set_error(runner, format!("endpoint answered HTTP {status}"));
            }
            Ok(Err(err)) => set_error(runner, format!("{err:#}")),
            Err(_) => set_error(runner, format!("timed out after {DELIVER_TIMEOUT:?}")),
        }
        runner.counters.failed.fetch_add(1, Ordering::Relaxed);
        if attempt < MAX_ATTEMPTS {
            let wait = crate::durable::backoff(attempt, RETRY_BASE, RETRY_MAX);
            tokio::time::sleep(wait).await;
        }
    }
    // Exhausted: drop-with-a-gap so the receiver reconciles rather than the
    // worker retrying a dead endpoint forever behind a growing queue.
    runner.counters.dropped.fetch_add(1, Ordering::Relaxed);
    tracing::warn!(
        target: TARGET_WEBHOOK,
        event = "webhook.delivery_dropped",
        endpoint = %runner.endpoint.name,
        seq = event.seq,
        "a delivery failed {MAX_ATTEMPTS} times and is dropped; the receiver should reconcile"
    );
}

fn set_error(runner: &Runner, msg: String) {
    *runner
        .counters
        .last_error
        .lock()
        .unwrap_or_else(|e| e.into_inner()) = msg;
}

// ───────────────────────────────────────────────────────────────────────────
// The HTTP POST — hyper over a raw connection, http:// and https://
// ───────────────────────────────────────────────────────────────────────────

/// POST `body` to the endpoint with the signature and event headers, returning
/// the HTTP status. One connection per delivery: webhooks are low-rate, and a
/// fresh connection is one fewer piece of state to keep bounded.
async fn post(
    endpoint: &Endpoint,
    body: &[u8],
    signed: &gunnar_core::webhook::SignedDelivery,
    kind: EventKind,
    delivery_id: &str,
) -> Result<u16> {
    use http_body_util::Full;
    use hyper::body::Bytes;

    let target = Url::parse(&endpoint.url)?;
    let mut builder = hyper::Request::builder()
        .method("POST")
        .uri(&target.path_and_query)
        .header("host", target.host_header())
        .header("content-type", "application/json")
        .header("user-agent", format!("gunnar/{}", env!("CARGO_PKG_VERSION")))
        .header(gunnar_core::webhook::EVENT_HEADER, kind_name(kind))
        .header(gunnar_core::webhook::DELIVERY_HEADER, delivery_id)
        .header(
            gunnar_core::webhook::NAMESPACE_HEADER,
            gunnar_core::webhook::WEBHOOK_NAMESPACE,
        )
        .header(
            gunnar_core::webhook::KEY_FINGERPRINT_HEADER,
            &signed.key_fingerprint,
        )
        .header(
            gunnar_core::webhook::SIGNATURE_HEADER,
            signed.signature_header(),
        );
    if let Some(secret) = endpoint.secret.as_deref() {
        builder = builder.header(
            gunnar_core::webhook::HMAC_HEADER,
            gunnar_core::webhook::hmac_sha256_header(secret.as_bytes(), body),
        );
    }
    let request = builder
        .body(Full::new(Bytes::copy_from_slice(body)))
        .context("building the delivery request")?;

    let stream = target.connect().await?;
    let status = send_over(stream, request).await?;
    Ok(status)
}

/// Drive one HTTP/1.1 request/response over an established byte stream.
async fn send_over<S>(stream: S, request: hyper::Request<http_body_util::Full<hyper::body::Bytes>>) -> Result<u16>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
    let io = hyper_util::rt::TokioIo::new(stream);
    let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
        .await
        .context("HTTP handshake with the endpoint")?;
    // The connection future must be driven; drop it when the response is in.
    // A32-14, stays a task: drives one HTTP connection; aborted when the response is in.
    let conn = tokio::spawn(async move {
        let _ = conn.await;
    });
    let response = sender
        .send_request(request)
        .await
        .context("sending the delivery")?;
    let status = response.status().as_u16();
    conn.abort();
    Ok(status)
}

/// A parsed endpoint URL: enough of one to open a socket and write a request.
struct Url {
    https: bool,
    host: String,
    port: u16,
    path_and_query: String,
}

impl Url {
    fn parse(raw: &str) -> Result<Self> {
        let (scheme, rest) = raw
            .split_once("://")
            .ok_or_else(|| anyhow::anyhow!("webhook url {raw:?} has no scheme"))?;
        let https = match scheme {
            "http" => false,
            "https" => true,
            other => anyhow::bail!("webhook url scheme {other:?} is not http or https"),
        };
        let (authority, path) = match rest.find('/') {
            Some(i) => (&rest[..i], &rest[i..]),
            None => (rest, "/"),
        };
        let (host, port) = match authority.rsplit_once(':') {
            Some((h, p)) => (
                h.to_owned(),
                p.parse::<u16>()
                    .with_context(|| format!("webhook url port {p:?}"))?,
            ),
            None => (authority.to_owned(), if https { 443 } else { 80 }),
        };
        anyhow::ensure!(!host.is_empty(), "webhook url {raw:?} has no host");
        Ok(Url {
            https,
            host,
            port,
            path_and_query: if path.is_empty() { "/".to_owned() } else { path.to_owned() },
        })
    }

    fn host_header(&self) -> String {
        let default = if self.https { 443 } else { 80 };
        if self.port == default {
            self.host.clone()
        } else {
            format!("{}:{}", self.host, self.port)
        }
    }

    /// Open the transport: a plain TCP stream, or a TLS one for https.
    async fn connect(&self) -> Result<Box<dyn Transport>> {
        let tcp = match nordisk_pairing::egress::proxy() {
            // **A box behind its front** (the lean estate, A19 row 14): the endpoint is
            // reached through the front's egress door, end to end, the front naming the host.
            Some(door) => {
                let (door, host, port) = (door.to_string(), self.host.clone(), self.port);
                let tunnel = tokio::task::spawn_blocking(move || nordisk_pairing::egress::dial(&door, &host, port))
                    .await
                    .context("the egress door's dial")?
                    .with_context(|| format!("connecting to {}:{} through the front's egress door", self.host, self.port))?;
                tunnel.set_nonblocking(true).context("the egress tunnel")?;
                tokio::net::TcpStream::from_std(tunnel).context("the egress tunnel")?
            }
            None => tokio::net::TcpStream::connect((self.host.as_str(), self.port))
                .await
                .with_context(|| format!("connecting to {}:{}", self.host, self.port))?,
        };
        if !self.https {
            return Ok(Box::new(tcp));
        }
        let connector = tls_connector()?;
        let server_name = tokio_rustls::rustls::pki_types::ServerName::try_from(self.host.clone())
            .with_context(|| format!("{:?} is not a valid TLS server name", self.host))?;
        let tls = connector
            .connect(server_name, tcp)
            .await
            .with_context(|| format!("TLS handshake with {}", self.host))?;
        Ok(Box::new(tls))
    }
}

/// The byte-stream shape `send_over` needs, so http and https share one path.
trait Transport: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static {}
impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static> Transport for T {}

/// A rustls client using the OS trust store, built once.
fn tls_connector() -> Result<tokio_rustls::TlsConnector> {
    use std::sync::OnceLock;
    static CONNECTOR: OnceLock<tokio_rustls::TlsConnector> = OnceLock::new();
    if let Some(c) = CONNECTOR.get() {
        return Ok(c.clone());
    }
    gunnar_grpc::ensure_crypto_provider();
    let mut roots = tokio_rustls::rustls::RootCertStore::empty();
    let native = rustls_native_certs::load_native_certs();
    for cert in native.certs {
        let _ = roots.add(cert);
    }
    anyhow::ensure!(
        !roots.is_empty(),
        "no system TLS roots were loadable, so an https webhook cannot be verified; use an \
         http:// endpoint behind a TLS-terminating proxy, or install a CA bundle"
    );
    let config = tokio_rustls::rustls::ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth();
    let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
    Ok(CONNECTOR.get_or_init(|| connector).clone())
}

// ───────────────────────────────────────────────────────────────────────────
// The live slot — set over the control plane, takes effect without a reboot
// ───────────────────────────────────────────────────────────────────────────

/// Owns the delivery's lifecycle, the way [`crate::mirror::ReplicationSlot`]
/// owns the replicator's: the appliance has no shell, so an endpoint has to be
/// settable over the control plane and take effect without a reboot. `apply`
/// writes the file FIRST, then swaps the running delivery, so a box that
/// acknowledged the change and lost power comes back with it.
pub struct WebhookSlot {
    stores: Arc<ServerStores>,
    bus: Arc<EventBus>,
    runtime: tokio::runtime::Handle,
    running: Mutex<Option<(Arc<WebhookDelivery>, WebhookTasks)>>,
}

impl WebhookSlot {
    pub fn new(
        stores: Arc<ServerStores>,
        bus: Arc<EventBus>,
        runtime: tokio::runtime::Handle,
    ) -> Arc<Self> {
        Arc::new(WebhookSlot {
            stores,
            bus,
            runtime,
            running: Mutex::new(None),
        })
    }

    fn delivery(&self) -> Option<Arc<WebhookDelivery>> {
        self.running
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .as_ref()
            .map(|(d, _)| Arc::clone(d))
    }

    /// The configuration an operator is asking about: the running one, else the
    /// file (so a `list` before anything is configured reads the file rather
    /// than inventing an answer).
    pub fn config(&self) -> Result<WebhookConfig> {
        if let Some(delivery) = self.delivery() {
            return Ok(delivery.config().clone());
        }
        WebhookConfig::load(self.stores.root()).context("reading the webhook configuration")
    }

    /// The server's webhook signing public key, one OpenSSH line — what a
    /// receiver pins. Available whether or not any endpoint is configured.
    pub fn signing_public_key(&self) -> Result<String> {
        let key = self.stores.webhook_key()?;
        Ok(key
            .public_key()
            .to_openssh()
            .context("rendering the webhook public key")?
            .trim()
            .to_owned())
    }

    /// Persist `config` and make it the running one. Disk first.
    pub fn apply(&self, config: WebhookConfig) -> Result<()> {
        // A twin keeps no per-box mirror (C7, `crate::twin::forget_mirrors`): the file alone.
        let saved = if self.stores.is_twin() {
            config.save(self.stores.root())
        } else {
            config.save_mirrored(self.stores.root(), self.stores.slow())
        };
        saved.context("writing the webhook configuration")?;
        let key = Arc::new(self.stores.webhook_key()?);
        let mut running = self.running.lock().unwrap_or_else(|e| e.into_inner());
        // Take (and thereby abort) the old tasks before subscribing the new
        // delivery, so two subscribers are never on the bus at once.
        if let Some((_, tasks)) = running.take() {
            tasks.abort();
        }
        if config.endpoints.is_empty() {
            tracing::info!(target: TARGET_WEBHOOK, event = "webhook.stopped", "the last webhook endpoint was removed; delivery is stopped");
            return Ok(());
        }
        let delivery = WebhookDelivery::new(Arc::clone(&self.bus), key, config);
        let tasks = {
            let _guard = self.runtime.enter();
            // A32-14, stays a task: the delivery's workers on the git runtime; their handles are kept (`running`) and stopped with it.
            delivery.spawn()
        };
        *running = Some((delivery, tasks));
        Ok(())
    }

    /// Stop delivering without touching the file (shutdown / drop). The config
    /// stays on disk so a restart comes back delivering to the same endpoints.
    pub fn stop(&self) {
        if let Some((_, tasks)) = self.running.lock().unwrap_or_else(|e| e.into_inner()).take() {
            tasks.abort();
        }
    }

    /// Build the `WebhookListResponse` the control plane returns.
    pub fn list_response(&self) -> Result<gunnar_proto::WebhookListResponse> {
        let endpoints = if let Some(delivery) = self.delivery() {
            delivery.endpoint_rows()
        } else {
            // Nothing running: render the on-disk endpoints with zero counters.
            self.config()?
                .endpoints
                .into_iter()
                .map(|e| gunnar_proto::WebhookEndpoint {
                    name: e.name,
                    url: e.url,
                    kinds: e.kinds,
                    namespace: e.namespace,
                    repo: e.repo,
                    has_secret: e.secret.is_some(),
                    added_at_unix_ms: e.added_at_unix_ms,
                    delivered: 0,
                    failed: 0,
                    dropped: 0,
                    last_error: String::new(),
                })
                .collect()
        };
        Ok(gunnar_proto::WebhookListResponse {
            endpoints,
            signing_public_key: self.signing_public_key().unwrap_or_default(),
            signature_namespace: gunnar_core::webhook::WEBHOOK_NAMESPACE.to_owned(),
        })
    }

    /// Add or replace an endpoint by name, validate it, persist, and start.
    pub fn add(&self, endpoint: Endpoint) -> Result<gunnar_proto::WebhookListResponse> {
        // Validate the URL up front so a bad one is refused at add time rather
        // than failing silently on every delivery.
        Url::parse(&endpoint.url)?;
        // A --kinds name that matches no kind is a typo that would deliver
        // nothing; refuse it by name.
        for name in &endpoint.kinds {
            anyhow::ensure!(
                kind_from_name(name).is_some(),
                "unknown event kind {name:?}; a --kinds filter that names nothing delivers nothing"
            );
        }
        let mut config = self.config()?;
        config.set_endpoint(endpoint);
        self.apply(config)?;
        self.list_response()
    }

    /// Remove an endpoint by name.
    pub fn remove(&self, name: &str) -> Result<Option<gunnar_proto::WebhookListResponse>> {
        let mut config = self.config()?;
        if !config.remove_endpoint(name) {
            return Ok(None);
        }
        self.apply(config)?;
        Ok(Some(self.list_response()?))
    }
}

#[cfg(test)]
mod tests;