//! 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()` 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. 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 { 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 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 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); } 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:?}" ); } } }