hyperhive/swarm-controller/src/forge_events.rs

//! Keeping verified `issues` and `pull_request` deliveries for the
//! notification router.
//!
//! [`crate::webhook`] checks the HMAC, then [`publish`] puts the body,
//! unchanged, on `swarm_wire::forge_events::subject(<event type>)` in the
//! `SWARM_EVENTS` stream. The router reads that stream, so it needs no
//! webhook secret and registers no hook of its own.
//!
//! The event type is Forgejo's `X-Forgejo-Event-Type` header (`issue_assign`,
//! `pull_request_sync`, …), not `X-Forgejo-Event`, which folds those into
//! `issues` and `pull_request`. It reaches the subject only once it is one
//! plain token.
//!
//! [`publish_event`] keeps an event this daemon makes, such as the
//! `pull_request_mergeability` event of `crate::pr_conflicts`, the same way.

use axum::http::HeaderMap;
use std::time::Duration;

/// The header naming the delivery's event type.
const EVENT_TYPE_HEADER: &str = "x-forgejo-event-type";

/// The header naming the delivery. A redelivery carries the same one, and the
/// stream drops a second message with the same id.
const DELIVERY_HEADER: &str = "x-forgejo-delivery";

/// How long the stream may take to ack a publish. Inside Forgejo's own
/// delivery timeout.
const PUBLISH_DEADLINE: Duration = Duration::from_secs(5);

/// Why a delivery was not kept.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum NotKept {
    /// No `X-Forgejo-Event-Type`, or one that is not a single subject token.
    BadEventType,
    /// The controller has no queue, or the stream refused or timed out.
    Queue(String),
}

/// The subject a delivery with `headers` goes to.
fn subject(headers: &HeaderMap) -> Result<String, NotKept> {
    let event_type = headers
        .get(EVENT_TYPE_HEADER)
        .and_then(|v| v.to_str().ok())
        .filter(|t| swarm_wire::forge_events::is_subject_token(t))
        .ok_or(NotKept::BadEventType)?;
    Ok(swarm_wire::forge_events::subject(event_type))
}

/// Keep one verified delivery in the stream, and wait for the stream's ack.
pub(crate) async fn publish(
    state: &crate::routes::AppState,
    headers: &HeaderMap,
    body: axum::body::Bytes,
) -> Result<(), NotKept> {
    let subject = subject(headers)?;
    let mut message = async_nats::jetstream::message::PublishMessage::build().payload(body);
    if let Some(id) = headers.get(DELIVERY_HEADER).and_then(|v| v.to_str().ok()) {
        message = message.message_id(id);
    }
    keep(state, subject, message).await
}

/// Keep an event this daemon makes itself, of `event_type`, in the stream, and
/// wait for the stream's ack. The router reads it like a delivery.
pub(crate) async fn publish_event(
    state: &crate::routes::AppState,
    event_type: &str,
    body: Vec<u8>,
) -> Result<(), NotKept> {
    if !swarm_wire::forge_events::is_subject_token(event_type) {
        return Err(NotKept::BadEventType);
    }
    let message = async_nats::jetstream::message::PublishMessage::build().payload(body.into());
    keep(
        state,
        swarm_wire::forge_events::subject(event_type),
        message,
    )
    .await
}

async fn keep(
    state: &crate::routes::AppState,
    subject: String,
    message: async_nats::jetstream::message::PublishMessage,
) -> Result<(), NotKept> {
    let Some(status) = state.status.as_ref() else {
        return Err(NotKept::Queue("no swarm queue is configured".to_owned()));
    };
    let client = status.queue_client();
    swarm_queue_client::ensure_connected(&client).map_err(|e| NotKept::Queue(e.to_string()))?;
    let kept = async {
        swarm_queue_client::forge_events::open_or_create(&client)
            .await
            .map_err(|e| swarm_queue_client::chain(&e))?;
        async_nats::jetstream::new(client.clone())
            .send_publish(subject.clone(), message)
            .await
            .map_err(|e| swarm_queue_client::chain(&e))?
            .await
            .map_err(|e| swarm_queue_client::chain(&e))
    };
    match tokio::time::timeout(PUBLISH_DEADLINE, kept).await {
        Ok(Ok(ack)) => {
            tracing::info!(%subject, seq = ack.sequence, duplicate = ack.duplicate, "webhook: forge event kept");
            Ok(())
        }
        Ok(Err(e)) => Err(NotKept::Queue(e)),
        Err(_) => Err(NotKept::Queue(format!(
            "the stream did not ack within {}s",
            PUBLISH_DEADLINE.as_secs()
        ))),
    }
}

#[cfg(test)]
mod tests {
    use super::{NotKept, subject};
    use axum::http::{HeaderMap, HeaderValue};

    fn with_type(value: &str) -> HeaderMap {
        let mut headers = HeaderMap::new();
        headers.insert(
            "X-Forgejo-Event-Type",
            HeaderValue::from_str(value).expect("a header value"),
        );
        headers
    }

    #[test]
    fn the_event_type_header_names_the_subject() {
        assert_eq!(
            subject(&with_type("issue_assign")),
            Ok("$SWARM.events.forge.issue_assign".to_owned())
        );
    }

    #[test]
    fn a_missing_or_multi_token_event_type_is_not_kept() {
        assert_eq!(subject(&HeaderMap::new()), Err(NotKept::BadEventType));
        for bad in ["issues.x", ">", "*", ""] {
            assert_eq!(
                subject(&with_type(bad)),
                Err(NotKept::BadEventType),
                "{bad:?}"
            );
        }
    }
}