Implement sensor status debouncing

This introduces a configurable debouncing delay before which no update
is sent to matrix, which avoids flooding the matrix room with updates
when the door is opened and closed in quick succession. It also allows
to quickly open and close the door without sending notifications.

The debouncing window can be configured in config.yml, see example.

Cargo.toml

@@ -13,6 +13,7 @@ tracing = "0.1"
 serde_yaml = "0.9"
 serde_json = "1.0"
 tracing-subscriber = "0.3.18"
+clap = { version = "4.6.7", features = ["derive"] }
 tokio = { version = "1.41", features = ["macros", "rt-multi-thread"] }
-matrix-sdk = { version = "0.8.0", default-features = false, features = [ "native-tls", "anyhow", "markdown" ] }
+matrix-sdk = { version = "0.18.0", default-features = false, features = [ "anyhow", "markdown" ] }
 rumqttc = "0.25"

README.md

@@ -0,0 +1,42 @@
+# matrix-lolopener
+
+A simple matrix bot which posts to a list of matrix rooms everytime the front door is opened or closed.
+The door state is retrieved by subscribing to an MQTT topic, and a matrix event is sent every time that state changes.
+
+## building
+
+```sh
+cargo build --release
+```
+
+## installing
+
+```sh
+useradd -r -M -s /usr/sbin/nologin matrix-lolopener
+install -o matrix-lolopener -g nogroup target/release/matrix-lolopener /opt/matrix-lolopener/bin/
+install -o matrix-lolopener -g nogroup -m 600 -T config.yml.example /opt/matrix-lolopener/etc/config.yml
+cp matrix-lolopener.service /etc/systemd/system/
+$EDITOR /opt/matrix-lolopener/etc/config.yml
+systemctl daemon-reload && systemctl enable --now matrix-lolopener.service
+```
+
+## usage
+
+```
+Matrix bot posting when the front door is opened or closed
+
+Usage: matrix-lolopener [OPTIONS]
+
+Options:
+  -c, --config <CONFIG>  Configuration file path [default: config.yml]
+  -h, --help             Print help
+  -V, --version          Print version
+```
+
+## contributing
+
+Send PRs to https://git.labolyon.fr/mirsal/matrix-lolopener
+
+## license
+
+GPL v3 or later

config.yml.example

@@ -0,0 +1,25 @@
+---
+
+mqtt:
+  id: "matrix-lolopener" # has to be unique
+  host: mosquitto.labolyon.dn42
+  port: 1883
+  topic: "zigbee/entree"
+
+rooms:
+  - "!iFCnqQcvrVIeehrjQO:chat.labolyon.fr"
+
+debounce_delay: 10 # seconds
+
+# username / password login
+# /!\ will create a new device every time the bot starts
+username: "login"
+password: "password"
+
+# alternatively, token-based login from an existing session
+# See: https://spec.matrix.org/v1.19/client-server-api/#login
+#user_id: "@lolopener:chat.labolyon.fr"
+#device_id: "STCRDGFKWZ"
+#access_token: "syt_XXXXXXXXXXXX_XXXXXXXXXXXXXXXXXXXX_XXXXXX"
+
+homeserver_uri: "https://chat.labolyon.fr/"

matrix-lolopener.service

@@ -0,0 +1,13 @@
+[Unit]
+Description=LOL-Opener Matrix Bot
+
+[Service]
+Type=simple
+User=matrix-lolopener
+WorkingDirectory=/opt/matrix-lolopener
+ExecStart=/opt/matrix-lolopener/bin/matrix-lolopener -c etc/config.yml
+Restart=on-failure
+RestartSec=5
+
+[Install]
+WantedBy=multi-user.target

src/config.rs

@@ -1,4 +1,4 @@
-/**
+/*!
  * matrix-lolopener v0.1.0
  *
  * Copyright (C) 2026 The 1312 Media Collective
@@ -46,6 +46,7 @@ pub enum AuthConfig {
 pub struct Config {
     pub mqtt: MQTTConfig,
     pub rooms: Vec<String>,
+    pub debounce_delay: u64,
     #[serde(flatten)]
     pub auth: AuthConfig,
     pub homeserver_uri: String
@@ -56,8 +57,8 @@ impl Config {
 
         let serialized_config = fs::read_to_string(config_file).map_err(|e| {
             match e.kind() {
-                io::ErrorKind::NotFound => Error::FileNotFoundError(config_file.into()),
-                io::ErrorKind::PermissionDenied => Error::PermissionDeniedError(),
+                io::ErrorKind::NotFound => Error::FileNotFound(config_file.into()),
+                io::ErrorKind::PermissionDenied => Error::PermissionDenied(),
                 _ => e.into(),
             }
         })?;
@@ -71,21 +72,21 @@ impl Config {
 #[derive(Debug)]
 #[allow(dead_code)] // we use Debug, so inner fields are never read
 pub enum Error {
-    FileNotFoundError(String),
-    PermissionDeniedError(),
-    InvalidFormatError(serde_yaml::Error),
-    IoError(io::Error),
+    FileNotFound(String),
+    PermissionDenied(),
+    InvalidFormat(serde_yaml::Error),
+    Io(io::Error),
 }
 
 impl From<serde_yaml::Error> for Error {
     fn from(e: serde_yaml::Error) -> Self {
-        Self::InvalidFormatError(e)
+        Self::InvalidFormat(e)
     }
 }
 
 impl From<io::Error> for Error {
     fn from(e: io::Error) -> Self {
-        Self::IoError(e)
+        Self::Io(e)
     }
 }
 

src/main.rs

@@ -1,4 +1,4 @@
-/**
+/*!
  * matrix-lolopener v0.1.0
  *
  * Copyright (C) 2026 The 1312 Media Collective
@@ -17,12 +17,15 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+#![recursion_limit = "256"]
+
 mod config;
 mod matrix;
 mod mqtt;
 
 use tokio::{
-    sync::{Notify, broadcast},
+    sync::{Notify, broadcast, mpsc},
+    time::{sleep, Duration},
     task::JoinHandle,
 };
 
@@ -32,14 +35,38 @@ use crate::{
     mqtt::{MQTT, StatusMessage},
 };
 
-use std::sync::Arc;
+use std::sync::{ Arc, Mutex };
 use tracing::{ info, debug };
 
+use clap::Parser;
+
+/// Matrix bot posting when the front door is opened or closed
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct Args {
+    /// Configuration file path
+    #[arg(short, long, default_value_t = "config.yml".into())]
+    config: String,
+}
+
+#[derive(PartialEq, Clone, Copy, Debug)]
+enum DoorSensorStatus {
+    Open,
+    Closed,
+}
+#[derive(Debug)]
+struct LolOpenerState {
+    sensor_status: Option<DoorSensorStatus>,
+    pending: Option<(DoorSensorStatus, String)>,
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt::init();
 
-    let config = Config::load("config.yml").unwrap_or_else(|e| {
+    let args = Args::parse();
+
+    let config = Config::load(&args.config).unwrap_or_else(|e| {
         panic!("Failed to load config: {e:?}")
     });
     let config = Arc::new(config);
@@ -50,7 +77,7 @@ async fn main() -> anyhow::Result<()> {
     // it holds a tuple with an HTML message and a list of rooms to post to
     let (bcast_tx, bcast_rx) = broadcast::channel::<(String, Vec<String>)>(1024);
 
-    let handle: JoinHandle<_> = (|| {
+    let handle: JoinHandle<_> = {
         let matrix_ready = Arc::clone(&matrix_ready);
         let config = Arc::clone(&config);
 
@@ -59,23 +86,80 @@ async fn main() -> anyhow::Result<()> {
             matrix_ready.notified().await;
             info!("Matrix is ready");
 
-            let mut mqtt = MQTT::new(config.mqtt.clone()).await;
-            mqtt.subscribe(|status: StatusMessage| {
-                let msg = match status {
-                    StatusMessage { contact: true, .. } => "🔴 Le LOL est fermé :(",
-                    StatusMessage { contact: false, .. } => "🟢 Le LOL est ouvert :)",
-                };
-                debug!("Sending msg {}", msg);
-                //XXX: Panic if the matrix module can't get the message
-                bcast_tx.send((msg.into(), config.rooms.clone())).unwrap();
+            let current_state = Arc::new(Mutex::new(LolOpenerState {
+                sensor_status: None,
+                pending: None
+            }));
+            let current_state = Arc::clone(&current_state);
+
+            let (debounce_tx, mut debounce_rx) = mpsc::unbounded_channel::<(DoorSensorStatus, String)>();
+            let debounce_delay = config.debounce_delay;
+            let rooms = config.rooms.clone();
+            let handle = tokio::spawn(async move {
+
+                while let Some((new_status, msg)) = debounce_rx.recv().await {
+                    let should_wait = {
+                        //XXX: Panic if the lock is poisoned
+                        let mut state = current_state.lock().unwrap();
+                        let should_wait = state.pending.is_none();
+
+                        state.pending = Some((new_status, msg));
+                        should_wait
+                    };
+
+                    if should_wait {
+                        let tx = bcast_tx.clone();
+                        let rooms = rooms.clone();
+                        let state = current_state.clone();
+
+                        tokio::spawn(async move {
+                            sleep(Duration::from_secs(debounce_delay)).await;
+
+                            let msg = {
+                                //XXX: Panic if the lock is poisoned
+                                let mut state = state.lock().unwrap();
+                                let pending = state.pending.take();
+
+                                match pending {
+                                    Some((status, msg))
+                                        if state.sensor_status.as_ref() != Some(&status) => {
+                                            state.sensor_status = Some(status);
+                                            Some(msg)
+                                        }
+
+                                    _ => None,
+                                }
+                            };
+
+                            if let Some(msg) = msg {
+                                debug!("Sending msg {}", msg);
+                                //XXX: Panic if the matrix module can't get the message
+                                tx.send((msg, rooms)).unwrap();
+                            }
+                        });
+                    }
+                }
+            });
+
+            let mut mqtt = MQTT::new(config.mqtt.clone()).await;
+            mqtt.subscribe(move |status: StatusMessage| {
+                let (new_status, msg) = match status {
+                    StatusMessage { contact: true, .. } => (DoorSensorStatus::Closed, "🔴 Le LOL est fermé :("),
+                    StatusMessage { contact: false, .. } => (DoorSensorStatus::Open, "🟢 Le LOL est ouvert :)"),
+                };
+
+                //XXX: Panic if the channel is closed
+                debounce_tx.send((new_status, String::from(msg))).unwrap();
 
-                ()
             }).await.unwrap(); //XXX: Panic if the mqtt broker fails
+
+            handle.abort();
+            handle.await.unwrap();
         })
-    })();
+    };
 
     login_and_sync(
-        config.homeserver_uri.clone().into(),
+        config.homeserver_uri.clone(),
         &config.auth,
         matrix_ready,
         bcast_rx

src/matrix.rs

@@ -1,4 +1,4 @@
-/**
+/*!
  * matrix-lolopener v0.1.0
  *
  * Copyright (C) 2024 The 1312 Media Collective
@@ -18,16 +18,20 @@
  */
 
 use matrix_sdk::{
-    matrix_auth::{
-        MatrixAuth,
-        MatrixSession,
-        MatrixSessionTokens
+    authentication::{
+        matrix::{
+            MatrixAuth,
+            MatrixSession,
+        },
+        SessionTokens,
     },
     SessionMeta,
+    store::RoomLoadSettings,
     config::SyncSettings,
     ruma::events::room::{
         member::StrippedRoomMemberEvent,
-        message::RoomMessageEventContent
+        message::RoomMessageEventContent,
+        pinned_events::RoomPinnedEventsEventContent,
     },
     ruma::{RoomId, OwnedUserId, device_id},
     Client, Room
@@ -41,7 +45,7 @@ use tokio::{
 
 use std::sync::Arc;
 
-use tracing::{error, info};
+use tracing::{error, info, debug};
 use std::error::Error;
 
 use crate::config::AuthConfig;
@@ -51,6 +55,7 @@ async fn on_stripped_state_member(
     client: Client,
     room: Room,
 ) {
+    //XXX: Panic if authentication has failed
     if room_member.state_key != client.user_id().unwrap() {
         return;
     }
@@ -86,14 +91,39 @@ async fn send_to_room(
     match room.client().join_room_by_id(room.room_id()).await {
         Ok(_) => {
             info!("Posting to room {}", room.room_id());
-            room.send(RoomMessageEventContent::text_html(msg.clone(), msg)).await?;
-            return Ok(());
+            let res = room.send(RoomMessageEventContent::text_html(msg.clone(), msg)).await?;
+
+            debug!("Retrieving list of pinned events");
+            let mut old_pinned_eids = room.pinned_event_ids().unwrap_or_default();
+            let mut new_pinned_eids = Vec::with_capacity(old_pinned_eids.len());
+
+            for eid in old_pinned_eids.drain(..) {
+                match room.event(eid.as_ref(), None).await {
+                    Ok(ev) => {
+                        let client = room.client();
+
+                        if client.user_id() != ev.sender().as_deref() {
+                            new_pinned_eids.push(eid);
+                        }
+                    },
+                    Err(err) => {
+                        error!("Failed to retrieve pinned event {eid:?}: {err:?}");
+                    }
+                }
+            }
+
+            info!("Updating pinned events");
+            new_pinned_eids.push(res.response.event_id);
+            room.send_state_event(RoomPinnedEventsEventContent::new(new_pinned_eids)).await?;
+
+            Ok(())
         },
         Err(e) => {
             error!("Failed to join room {e:?}");
-            return Err(e.into());
+
+            Err(e.into())
         }
-    };
+    }
 }
 
 async fn login(
@@ -103,7 +133,9 @@ async fn login(
     match auth_config {
         AuthConfig::PasswordAuthConfig{username, password} => {
             matrix_auth.login_username(username, password)
-                .initial_device_display_name("bender v0.2.0").await?;
+                .initial_device_display_name(
+                    format!("{} v{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION")).as_str()
+                ).await?;
         },
         AuthConfig::TokenAuthConfig{user_id, device_id, access_token} => {
             matrix_auth.restore_session(
@@ -112,11 +144,12 @@ async fn login(
                         user_id: <OwnedUserId>::try_from(user_id.as_str())?,
                         device_id: device_id!(device_id.as_str()).to_owned(),
                     },
-                    tokens: MatrixSessionTokens {
+                    tokens: SessionTokens {
                         access_token: access_token.to_owned(),
                         refresh_token: None,
                     }
-                }
+                },
+                RoomLoadSettings::All
             ).await?;
         },
     }
@@ -158,7 +191,6 @@ pub async fn login_and_sync(
                 Ok(_) => return Ok(()),
                 Err(e) => {
                     error!("Sync error: {e:?}");
-                    ()
                 }
             },
             recv = rx.recv() => match recv {
@@ -179,7 +211,7 @@ pub async fn login_and_sync(
                     }
                 },
                 Err(e) => {
-                    panic!("Broadcast channel is lagging: {e:?}");
+                    panic!("Broken channel: {e:?}");
                 }
             }
         }

src/mqtt.rs

@@ -1,4 +1,4 @@
-/**
+/*!
  * matrix-lolopener v0.1.0
  *
  * Copyright (C) 2026 The 1312 Media Collective
@@ -54,13 +54,12 @@ impl MQTT {
         mqttoptions.set_keep_alive(Duration::from_secs(5));
 
         let (client, eventloop) = AsyncClient::new(mqttoptions, 1);
-        let mqtt = MQTT {
-            config: config,
-            client: client,
-            eventloop: eventloop,
-        };
 
-        mqtt
+        MQTT {
+            config,
+            client,
+            eventloop,
+        }
     }
 
     pub async fn subscribe<F: Fn(StatusMessage)>(&mut self, callback: F) -> Result<()> {
@@ -71,13 +70,10 @@ impl MQTT {
             let notification = self.eventloop.poll().await?;
             debug!("Notification = {:?}", notification);
 
-            match notification {
-                Event::Incoming(Incoming::Publish(p)) => {
-                    let msg: StatusMessage = serde_json::from_slice(p.payload.as_ref())?;
-                    info!("Message: {:?}", msg);
-                    callback(msg);
-                },
-                _ => (),
+            if let Event::Incoming(Incoming::Publish(p)) = notification {
+                let msg: StatusMessage = serde_json::from_slice(p.payload.as_ref())?;
+                info!("Message: {:?}", msg);
+                callback(msg);
             }
         }
 
@@ -89,26 +85,26 @@ impl MQTT {
 #[derive(Debug)]
 #[allow(dead_code)] // we use Debug, so inner fields are never read
 pub enum Error {
-    MQTTClientError(rumqttc::ClientError),
-    MQTTConnectionError(rumqttc::ConnectionError),
-    InvalidFormatError(serde_json::Error),
+    MQTTClient(rumqttc::ClientError),
+    MQTTConnection(rumqttc::ConnectionError),
+    InvalidFormat(serde_json::Error),
 }
 
 impl From<rumqttc::ClientError> for Error {
     fn from(e: rumqttc::ClientError) -> Self {
-        Self::MQTTClientError(e)
+        Self::MQTTClient(e)
     }
 }
 
 impl From<rumqttc::ConnectionError> for Error {
     fn from(e: rumqttc::ConnectionError) -> Self {
-        Self::MQTTConnectionError(e)
+        Self::MQTTConnection(e)
     }
 }
 
 impl From<serde_json::Error> for Error {
     fn from(e: serde_json::Error) -> Self {
-        Self::InvalidFormatError(e)
+        Self::InvalidFormat(e)
     }
 }