chore: log starting/stopping ntfy worker

.tci

@@ -1,9 +1,9 @@
 #!/bin/bash
 
-cargo build --release
+cargo build --release --all-features
 echo "stopping service"
 systemctl --user stop mumble-stats
 echo "replacing binary"
-cat ./target/release/mumble-stats-api > /opt/bin/mumble-stats-api
+mv ./target/release/mumble-stats-api /opt/bin/mumble-stats-api
 echo "restarting service"
 systemctl --user start mumble-stats

Cargo.toml

@@ -22,12 +22,14 @@ tokio = { version = "1.36", features = ["net", "macros", "rt-multi-thread", "io-
 tokio-native-tls = "0.3.1"
 tracing = "0.1.40"
 tracing-subscriber = "0.3.18"
+reqwest = { version = "0.12", optional = true }
 
 [build-dependencies]
 prost-build = "0.12"
 
 [features]
 default = []
+ntfy = ["dep:reqwest"]
 
 [lints.rust]
 unsafe_code = "forbid"

src/main.rs

@@ -46,6 +46,14 @@ struct CliArgs {
 	/// disable arbitrary server join and peek
 	#[arg(long, default_value_t = false)]
 	no_peek: bool,
+
+	/// ntfy.sh server to use for notifications
+	#[arg(long, default_value = "https://ntfy.sh")]
+	ntfy: String,
+
+	/// ntfy.sh topic to use for notifications, leave empty to disable notifications
+	#[arg(long)]
+	topic: Option<String>,
 }
 
 #[tokio::main]
@@ -73,6 +81,14 @@ async fn main() {
 	// 	app = app.route("/peek", get(peek_server));
 	// }
 	
+	#[cfg(feature = "ntfy")]
+	if let Some(topic) = args.topic {
+		tracing::info!("starting ntfy worker");
+		let events = session.events();
+		let target = format!("{}/{topic}", args.ntfy);
+		tokio::spawn(async move { handle_ntfy(target, events).await });
+	}
+
 	let app = app
 		.route("/info", get(server_info))
 		.route("/users", get(server_users))
@@ -100,6 +116,41 @@ async fn server_ws(ws: WebSocketUpgrade, State(session): State<Arc<Session>>) ->
 	ws.on_upgrade(|socket| handle_ws(socket, sub))
 }
 
+#[cfg(feature = "ntfy")]
+async fn handle_ntfy(target: String, mut sub: broadcast::Receiver<session::SessionEvent>) {
+	use std::collections::HashMap;
+	let client = reqwest::Client::new();
+	let mut users = HashMap::new();
+
+	while let Ok(event) = sub.recv().await {
+		if let Err(e) = match event {
+			session::SessionEvent::AddUser(user) => {
+				users.insert(user.session, user.name.clone());
+				client
+					.post(&target)
+					.body(format!("user {} connected", user.name))
+					.header("Title", "mumble")
+					.header("Priority", "default")
+					.header("Tags", "studio_microphone")
+					.send()
+					.await
+			},
+			session::SessionEvent::RemoveUser(id) =>
+				client
+					.post(&target)
+					.body(format!("user {} left", users.remove(&id).unwrap_or_default()))
+					.header("Title", "mumble")
+					.header("Priority", "default")
+					.header("Tags", "x")
+					.send()
+					.await,
+		} {
+			tracing::debug!("events channel closed: {e}");
+		}
+	}
+	tracing::info!("stopping ntfy worker");
+}
+
 async fn handle_ws(mut socket: WebSocket, mut sub: broadcast::Receiver<session::SessionEvent>) {
 	while let Ok(event) = sub.recv().await {
 		if let Err(e) = match event {

src/tcp/control.rs

@@ -41,6 +41,7 @@ impl ControlChannel {
 		let size = rx.read_u32().await?;
 		let mut buffer = vec![0u8; size as usize];
 		rx.read_exact(&mut buffer).await?;
+		drop(rx);
 		super::proto::Packet::decode(id, &buffer)
 	}
 }