Update README

The warp repo hasn't seen any activity in over a year; project seems
fairly dead.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bebot"
-version = "0.2.2"
+version = "0.2.3"
 description = "Gitlab webhook bot that publishes events to Matrix"
 license = "GPL-3.0"
 authors = [
@@ -24,20 +24,21 @@ exclude = [
 
 [dependencies]
 anyhow = "1"
+axum = "0.8.4"
+axum-extra = { version = "0.10.1", features = ["typed-header"] }
 chrono = { version = "0.4", features = ["serde"] }
-constant_time_eq = "0.3"
+constant_time_eq = "0.4"
 dateparser = "0.2"
-env_logger = "0.10"
+env_logger = "0.11"
 futures = "0.3"
-http = "0.2"
+http = "1.3"
 log = { version = "0.4", features = ["std"] }
-matrix-sdk = { version = "0.6", features = ["anyhow", "markdown", "rustls-tls"], default-features = false }
+matrix-sdk = { version = "0.13", features = ["anyhow", "markdown", "rustls-tls"], default-features = false }
-quick-xml = { version = "0.30", features = ["serialize"] }
+quick-xml = { version = "0.38", features = ["serialize"] }
 regex = "1"
-reqwest = { version = "0.11", default-features = false, features = ["tokio-rustls", "rustls-tls-native-roots", "gzip"] }
+reqwest = { version = "0.12", default-features = false, features = ["charset", "http2", "gzip", "rustls-tls-native-roots", "system-proxy"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_regex = "1"
 serde_yaml = "0.9"
 tokio = { version = "1", default-features = false, features = ["rt-multi-thread", "macros", "time"] }
-warp = "0.3"

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.72-slim-bullseye AS builder
+FROM rust:1.88-slim-bookworm AS builder
 
 WORKDIR /bebot-build
 

README.md

@@ -11,11 +11,17 @@ Currently-supported Gitlab event types:
 * Merge request events
 * Pipeline events (only publishes on failure for now)
 
+It can also watch for new messages in a mailing list (assuming that
+mailing list is tracked by [The Mail
+Archive](https://mail-archive.com/)), and publish notifications of new
+messages to Matrix.
+
 ## Building
 
 Bebot is written in Rust, and requires a Rust toolchain in order to
-build.  The usual `cargo build` or `cargo build --release` will do the
+build.  I'm actually not sure what Bebot's MSRV is, but as of this
-trick.
+writing, 1.72 worked.  The usual `cargo build` or `cargo build
+--release` will do the trick.
 
 You can also build and install the latest released version of Bebot by
 running `cargo install bebot`.
@@ -26,6 +32,8 @@ Bebot requires a configuration file in YAML format.  See
 `sample-config.yaml` for all existing configuration options, as well as
 documentation on what each option does.
 
+### Gitlab Hooks
+
 When setting up the webhook in Gitlab, use the same `token` from the
 configuration file in the webhook's "Secret token" field.  You should
 only select "Push events", "Tag push events", "Issues events", "Merge
@@ -42,6 +50,18 @@ output to stdout a YAML snippet that goes under the `repo_configs`
 section of the configuration file.  If you run the script with no
 arguments, it will print out usage details.
 
+### `mail-archive.com`
+
+If you want Bebot to publish a Matrix message when a new email hits one
+of your configured mailing lists, you need to provide a directory for
+Bebot to store state, so it can keep track of what emails it has already
+sent a Matrix message for.  Otherwise, it will re-publish messages every
+time you restart Bebot.
+
+Remember that if you are running Bebot in a Docker container, you'll
+need to mount a volume to store state so it persists across container
+restarts and upgrades.
+
 ## Running
 
 After you've done all that, simply run Bebot:
@@ -57,7 +77,18 @@ logging verbosity.  (Try `debug`, `info`, `warn` `error`, or `off`.)
 
 A `Dockerfile` is also provided.  When running the container it builds,
 mount the configuration file so it appears inside the container as
-`/bebot/config/bebot.yaml`.
+`/bebot/config/bebot.yaml`.  If you are publishing from
+`mail-archive.com`, also remember to mount a state storage directory or
+volume wherever you've specified in the configuration file.
 
 Release images are [published to Docker
 Hub](https://hub.docker.com/r/kelnos/bebot).
+
+## Contributing
+
+I currently host Bebot on my [private Gitea
+server](https://git.spurint.org/brian/bebot).  Since I don't want to
+deal with spam, I don't enable user registrations.  If you'd like to
+submit issues and/or merge requests, please [message me on
+Matrix](https://matrix.to/#/@brian:tarricone.org) with your email
+address and preferred username, and I'll create an account for you.

src/config.rs

@@ -105,7 +105,7 @@ fn load_blocking(path: &String) -> anyhow::Result<Config> {
 pub async fn load<S: AsRef<str>>(path: S) -> anyhow::Result<Config> {
     let p = String::from(path.as_ref());
     let config = tokio::task::spawn_blocking(move || {
-        load_blocking(&p).with_context(|| format!("Failed to load config from {}", p))
+        load_blocking(&p).with_context(|| format!("Failed to load config from {p}"))
     })
     .await??;
     Ok(config)

src/gitlab_event.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+#![allow(unused)]
+
 use core::fmt;
 
 use crate::config::PublishEvent;
@@ -281,7 +283,7 @@ impl GitlabEventExt for GitlabEvent {
                     false
                 }
             }
-            GitlabEvent::TagPush { .. } => find_publish_event!(publish_events, PublishEvent::TagPush { .. }).is_some(),
+            GitlabEvent::TagPush { .. } => find_publish_event!(publish_events, PublishEvent::TagPush).is_some(),
             GitlabEvent::Issue { object_attributes, .. } => {
                 if let Some(PublishEvent::Issues { actions }) =
                     find_publish_event!(publish_events, PublishEvent::Issues { .. })
@@ -437,7 +439,7 @@ impl GitlabEventExt for GitlabEvent {
                         .or(merge_request.as_ref().map(|mr| mr.title.clone()))
                         .iter()
                         .fold(format!("Pipeline **{}**", object_attributes.status), |accum, title| {
-                            format!("{}: {}", accum, title)
+                            format!("{accum}: {title}")
                         });
                     vec![markdown_link(&title, &object_attributes.url)]
                 } else {
@@ -451,7 +453,7 @@ impl GitlabEventExt for GitlabEvent {
 
 #[inline]
 fn markdown_link(title: &String, url: &String) -> String {
-    format!("[{}]({})", title, url)
+    format!("[{title}]({url})")
 }
 
 pub fn parse_ref(r#ref: &str) -> String {

src/gitlab_webhook.rs

@@ -16,14 +16,15 @@
 
 use std::sync::Arc;
 
+use axum::{extract::State, routing::post, Json, Router};
+use axum_extra::{headers, TypedHeader};
 use constant_time_eq::constant_time_eq;
-use http::StatusCode;
+use http::{HeaderName, HeaderValue, StatusCode};
 use matrix_sdk::{
     ruma::{events::room::message::RoomMessageEventContent, OwnedRoomOrAliasId},
     Client,
 };
-use tokio::sync::mpsc;
+use tokio::sync::mpsc::{self, Sender};
-use warp::{filters::BoxedFilter, reply::Reply, Filter};
 
 use crate::{
     config::GitlabWebhookConfig,
@@ -31,6 +32,42 @@ use crate::{
     matrix,
 };
 
+static X_GITLAB_TOKEN: HeaderName = HeaderName::from_static("x-gitlab-token");
+
+struct XGitlabToken(String);
+
+impl headers::Header for XGitlabToken {
+    fn name() -> &'static HeaderName {
+        &X_GITLAB_TOKEN
+    }
+
+    fn decode<'i, I>(values: &mut I) -> Result<Self, headers::Error>
+    where
+        Self: Sized,
+        I: Iterator<Item = &'i http::HeaderValue>,
+    {
+        let value = values.next().ok_or_else(headers::Error::invalid)?;
+
+        if value.is_empty() {
+            Err(headers::Error::invalid())
+        } else {
+            Ok(XGitlabToken(
+                value.to_str().map_err(|_| headers::Error::invalid())?.to_string(),
+            ))
+        }
+    }
+
+    fn encode<E: Extend<http::HeaderValue>>(&self, values: &mut E) {
+        values.extend(std::iter::once(HeaderValue::from_str(self.0.as_str()).unwrap()));
+    }
+}
+
+#[derive(Clone)]
+struct WebhookState {
+    config: Arc<GitlabWebhookConfig>,
+    event_tx: Sender<(GitlabEvent, OwnedRoomOrAliasId)>,
+}
+
 pub fn build_gitlab_messages(event: &GitlabEvent) -> Vec<String> {
     let project = event.project();
     let refname = event.r#ref().map(parse_ref);
@@ -41,7 +78,7 @@ pub fn build_gitlab_messages(event: &GitlabEvent) -> Vec<String> {
             format!(
                 "\\[{}\\] {}*{}* {}",
                 project.path_with_namespace,
-                refname.as_ref().map(|rn| format!("`{}` ", rn)).unwrap_or_default(),
+                refname.as_ref().map(|rn| format!("`{rn}` ")).unwrap_or_default(),
                 event.user(),
                 title,
             )
@@ -56,87 +93,73 @@ pub async fn handle_gitlab_event(
 ) -> anyhow::Result<()> {
     let room = matrix::ensure_room_joined(matrix_client, room_id).await?;
     for msg in build_gitlab_messages(&event) {
-        debug!("Sending message to {}: {}", room_id, msg);
+        debug!("Sending message to {room_id}: {msg}");
         let msg_content = RoomMessageEventContent::text_markdown(&msg);
-        room.send(msg_content, None).await?;
+        room.send(msg_content).await?;
     }
     Ok(())
 }
 
-pub fn build_route(config: GitlabWebhookConfig, matrix_client: Client) -> anyhow::Result<BoxedFilter<(impl Reply,)>> {
+async fn handle_hooks_gitlab(
+    State(state): State<WebhookState>,
+    TypedHeader(token): TypedHeader<XGitlabToken>,
+    Json(event): Json<GitlabEvent>,
+) -> (StatusCode, &'static str) {
+    match event {
+        GitlabEvent::Other => (StatusCode::BAD_REQUEST, "Unsupported Gitlab event type"),
+        _ => {
+            let project = event.project();
+            let config_key = project.web_url.replace("http://", "").replace("https://", "");
+            if let Some(repo_config) = state.config.repo_configs.get(&config_key) {
+                if !constant_time_eq(token.0.as_bytes(), repo_config.token.as_bytes()) {
+                    warn!("Invalid token for repo '{config_key}'");
+                    (StatusCode::FORBIDDEN, "Invalid token")
+                } else {
+                    debug!("payload: {event:?}");
+                    if let Some(room) = &repo_config.room.as_ref().or(state.config.default_room.as_ref()) {
+                        let publish_events = repo_config
+                            .publish_events
+                            .as_ref()
+                            .or(state.config.default_publish_events.as_ref());
+                        if publish_events.map(|ecs| event.should_publish(ecs)).unwrap_or(true) {
+                            if let Err(err) = state.event_tx.send((event, (*room).clone())).await {
+                                warn!("Failed to enqueue payload: {err}");
+                            }
+                        }
+                        (StatusCode::OK, "OK")
+                    } else {
+                        info!("Channel not configured for repo '{config_key}'");
+                        (StatusCode::NOT_FOUND, "Matrix room not configured for repo")
+                    }
+                }
+            } else {
+                info!("Repo '{config_key}' unconfigured");
+                (StatusCode::NOT_FOUND, "Repo not configured")
+            }
+        }
+    }
+}
+
+pub fn build_route(config: GitlabWebhookConfig, matrix_client: Client) -> Router {
     let (event_tx, mut event_rx) = mpsc::channel::<(GitlabEvent, OwnedRoomOrAliasId)>(100);
     tokio::spawn(async move {
         while let Some((event, room)) = event_rx.recv().await {
             if let Err(err) = handle_gitlab_event(event, &room, &matrix_client).await {
-                warn!("Failed to handle payload: {}", err);
+                warn!("Failed to handle payload: {err}");
             }
         }
     });
 
-    let gitlab_root_path = if let Some(url_prefix) = config.url_prefix.as_ref() {
+    let path = if let Some(url_prefix) = &config.url_prefix {
-        url_prefix.split('/').fold(warp::any().boxed(), |last, segment| {
+        format!("{url_prefix}/hooks/gitlab")
-            if segment.is_empty() {
-                last
     } else {
-                last.and(warp::path(segment.to_string())).boxed()
+        "/hooks/gitlab".to_owned()
-            }
-        })
-    } else {
-        warp::any().boxed()
     };
 
-    let config = Arc::new(config);
+    let state = WebhookState {
-    let gitlab = gitlab_root_path
+        config: Arc::new(config),
-        .and(warp::path!("hooks" / "gitlab"))
+        event_tx,
-        .and(warp::post())
+    };
-        .and(warp::header::<String>("x-gitlab-token"))
-        .and(warp::body::json())
-        .then(move |token: String, event: GitlabEvent| {
-            let event_tx = event_tx.clone();
-            let config = Arc::clone(&config);
 
-            async move {
+    Router::new().route(&path, post(handle_hooks_gitlab)).with_state(state)
-                match event {
-                    GitlabEvent::Other => {
-                        warp::reply::with_status("Unsupported Gitlab event type", StatusCode::BAD_REQUEST)
-                    }
-                    _ => {
-                        let project = event.project();
-                        let config_key = project.web_url.replace("http://", "").replace("https://", "");
-                        if let Some(repo_config) = config.repo_configs.get(&config_key) {
-                            if !constant_time_eq(token.as_bytes(), repo_config.token.as_bytes()) {
-                                warn!("Invalid token for repo '{}'", config_key);
-                                warp::reply::with_status("Invalid token", StatusCode::FORBIDDEN)
-                            } else {
-                                debug!("payload: {:?}", event);
-                                if let Some(room) = &repo_config.room.as_ref().or(config.default_room.as_ref()) {
-                                    let publish_events = repo_config
-                                        .publish_events
-                                        .as_ref()
-                                        .or(config.default_publish_events.as_ref());
-                                    if publish_events.map(|ecs| event.should_publish(ecs)).unwrap_or(true) {
-                                        if let Err(err) = event_tx.send((event, (*room).clone())).await {
-                                            warn!("Failed to enqueue payload: {}", err);
-                                        }
-                                    }
-                                    warp::reply::with_status("OK", StatusCode::OK)
-                                } else {
-                                    info!("Channel not configured for repo '{}'", config_key);
-                                    warp::reply::with_status(
-                                        "Matrix room not configured for repo",
-                                        StatusCode::NOT_FOUND,
-                                    )
-                                }
-                            }
-                        } else {
-                            info!("Repo '{}' unconfigured", config_key);
-                            warp::reply::with_status("Repo not configured", StatusCode::NOT_FOUND)
-                        }
-                    }
-                }
-            }
-        })
-        .boxed();
-
-    Ok(gitlab)
 }

src/mail_archive.rs

@@ -112,14 +112,14 @@ async fn handle_list(
 
     let rooms_f = room_ids.iter().map(|room_id| {
         matrix::ensure_room_joined(matrix_client, room_id)
-            .map(move |res| res.with_context(|| format!("Failed to join Matrix room '{}'", room_id)))
+            .map(move |res| res.with_context(|| format!("Failed to join Matrix room '{room_id}'")))
     });
     let rooms = join_all(rooms_f)
         .await
         .into_iter()
         .flat_map(|room_res| match room_res {
             Err(err) => {
-                warn!("{:#}", err);
+                warn!("{err:#}");
                 vec![]
             }
             Ok(room) => vec![room],
@@ -133,7 +133,7 @@ async fn handle_list(
         .get(url)
         .send()
         .await
-        .with_context(|| format!("Failed to fetch mail RSS feed from '{}'", url))
+        .with_context(|| format!("Failed to fetch mail RSS feed from '{url}'"))
         .and_then(|response| {
             if !response.status().is_success() {
                 Err(anyhow!(
@@ -148,10 +148,10 @@ async fn handle_list(
     let body = response
         .text()
         .await
-        .with_context(|| format!("Failed to decode RSS response body for '{}'", url))?;
+        .with_context(|| format!("Failed to decode RSS response body for '{url}'"))?;
     let mail_rss = tokio::task::spawn_blocking(move || quick_xml::de::from_str::<MailRss>(&body))
         .await?
-        .with_context(|| format!("Failed to parse RSS feed for '{}'", url))?;
+        .with_context(|| format!("Failed to parse RSS feed for '{url}'"))?;
     let items = mail_rss
         .channel
         .items
@@ -167,7 +167,7 @@ async fn handle_list(
                     "\\[{}\\] [{}]({})",
                     list.name, item.title, item.link
                 ));
-                room.send(msg, None)
+                room.send(msg)
                     .await
                     .with_context(|| format!("Failed to send message to room '{}'", room.room_id()))?;
             }
@@ -223,7 +223,7 @@ pub fn start_polling(config: MailArchiveConfig, matrix_client: Client) -> anyhow
                         )
                         .await
                         {
-                            warn!("{:#}", err);
+                            warn!("{err:#}");
                         }
 
                         sleep(update_interval).await;
@@ -269,7 +269,7 @@ mod test {
         let f = File::open(format!("{}/test-data/maillist.xml", env!("CARGO_MANIFEST_DIR")))?;
         let r = BufReader::new(f);
         let mail_rss = quick_xml::de::from_reader::<_, MailRss>(r)?;
-        println!("{:#?}", mail_rss);
+        println!("{mail_rss:#?}");
         Ok(())
     }
 }

src/main.rs

@@ -27,13 +27,15 @@ mod gitlab_webhook;
 mod mail_archive;
 mod matrix;
 
-use std::{env, net::IpAddr, process::exit};
+use std::{env, process::exit};
 
 use anyhow::Context;
 use futures::future::join_all;
-use warp::Filter;
+use tokio::net::TcpListener;
 
 async fn run() -> anyhow::Result<()> {
+    info!("{} v{} starting...", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
+
     let config_path = env::args()
         .nth(1)
         .ok_or_else(|| anyhow!("Config file should be passed as only parameter"))?;
@@ -48,18 +50,14 @@ async fn run() -> anyhow::Result<()> {
     };
 
     if let Some(gitlab_webhook) = config.gitlab_webhook.take() {
-        let gitlab = gitlab_webhook::build_route(gitlab_webhook, matrix_client.clone())?;
+        let gitlab = gitlab_webhook::build_route(gitlab_webhook, matrix_client.clone());
-        let routes = gitlab.with(warp::log("bebot"));
+        let bind_addr = format!(
-
+            "{}:{}",
-        let addr = config
+            config.bind_address.as_deref().unwrap_or("127.0.0.1"),
-            .bind_address
+            config.bind_port.unwrap_or(3000)
-            .as_ref()
+        );
-            .cloned()
+        let listener = TcpListener::bind(bind_addr).await?;
-            .unwrap_or_else(|| "127.0.0.1".to_string())
+        axum::serve(listener, gitlab).await?;
-            .parse::<IpAddr>()
-            .context("Failed to parse bind_address")?;
-        let port = config.bind_port.unwrap_or(3000);
-        warp::serve(routes).run((addr, port)).await;
     }
 
     join_all(handles).await;
@@ -76,7 +74,7 @@ async fn main() {
     env_logger::init_from_env(lenv);
 
     if let Err(err) = run().await {
-        error!("{:#}", err);
+        error!("{err:#}");
         exit(1);
     }
 }

src/matrix.rs

@@ -18,7 +18,7 @@ use std::{fmt, process::exit, time::Duration};
 
 use matrix_sdk::{
     config::SyncSettings,
-    room::Joined,
+    room::Room,
     ruma::{OwnedRoomOrAliasId, OwnedUserId, RoomOrAliasId, UserId},
     BaseRoom, Client,
 };
@@ -26,12 +26,8 @@ use serde::de;
 
 use crate::config::Config;
 
-async fn build_sync_settings(matrix_client: &Client) -> SyncSettings {
+async fn build_sync_settings() -> SyncSettings {
-    let mut settings = SyncSettings::default().timeout(Duration::from_secs(30));
+    SyncSettings::default().timeout(Duration::from_secs(30))
-    if let Some(token) = matrix_client.sync_token().await {
-        settings = settings.token(token);
-    }
-    settings
 }
 
 pub async fn connect(config: &Config) -> anyhow::Result<Client> {
@@ -41,21 +37,21 @@ pub async fn connect(config: &Config) -> anyhow::Result<Client> {
         .build()
         .await?;
     client
+        .matrix_auth()
         .login_username(&config.user_id, &config.password)
         .initial_device_display_name("Bebot")
-        .send()
         .await?;
     info!("Connected to matrix as {}; waiting for first sync", config.user_id);
 
-    let settings = build_sync_settings(&client).await;
+    let settings = build_sync_settings().await;
     client.sync_once(settings).await?;
     info!("First matrix sync complete");
 
     let sync_client = client.clone();
     tokio::spawn(async move {
-        let settings = build_sync_settings(&sync_client).await;
+        let settings = build_sync_settings().await;
         if let Err(err) = sync_client.sync(settings).await {
-            error!("Matrix sync failed: {}", err);
+            error!("Matrix sync failed: {err}");
             exit(1);
         }
     });
@@ -63,7 +59,7 @@ pub async fn connect(config: &Config) -> anyhow::Result<Client> {
     Ok(client)
 }
 
-pub async fn ensure_room_joined(matrix_client: &Client, room_id: &OwnedRoomOrAliasId) -> anyhow::Result<Joined> {
+pub async fn ensure_room_joined(matrix_client: &Client, room_id: &OwnedRoomOrAliasId) -> anyhow::Result<Room> {
     fn room_matches(a_room: &BaseRoom, our_room: &OwnedRoomOrAliasId) -> bool {
         let our_room_str = our_room.as_str();
         a_room.room_id().as_str() == our_room_str
@@ -85,11 +81,11 @@ pub async fn ensure_room_joined(matrix_client: &Client, room_id: &OwnedRoomOrAli
             .iter()
             .find(|a_room| room_matches(a_room, room_id))
         {
-            invited.accept_invitation().await?;
+            invited.join().await?;
         } else {
             matrix_client.join_room_by_id_or_alias(room_id, &[]).await?;
         }
-        let settings = build_sync_settings(matrix_client).await;
+        let settings = build_sync_settings().await;
         matrix_client.sync_once(settings).await?;
         room = matrix_client
             .joined_rooms()