Add UserId newtype and WebSocket ping keep-alive
All checks were successful
CI / ci (push) Successful in 5m57s

Replace raw String user IDs with a UserId newtype in domain for
type safety, and add 30s ping/pong to prevent idle WebSocket
disconnects.
This commit is contained in:
2026-08-18 17:02:19 +02:00
parent 4db04329b4
commit 99f614b7b7
11 changed files with 100 additions and 61 deletions

View File

@@ -1,11 +1,12 @@
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use axum::extract::State;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::response::IntoResponse;
use domain::{BroadcastEvent, BroadcastSubscription, Color, Position};
use domain::{BroadcastEvent, BroadcastSubscription, Color, Position, UserId};
use flate2::Compression;
use flate2::write::GzEncoder;
use futures::{SinkExt, StreamExt, stream::SplitSink};
@@ -29,10 +30,12 @@ pub async fn ws_upgrade(
async fn handle_connection(socket: WebSocket, state: Arc<WsState>) {
let (mut sender, mut receiver) = socket.split();
let connection_id = state
.connection_counter
.fetch_add(1, Ordering::Relaxed)
.to_string();
let connection_id = UserId::new(
state
.connection_counter
.fetch_add(1, Ordering::Relaxed)
.to_string(),
);
info!("WebSocket connected: {connection_id}");
@@ -105,11 +108,16 @@ fn gzip_compress(data: &[u8]) -> Option<Vec<u8>> {
encoder.finish().ok()
}
const PING_INTERVAL: Duration = Duration::from_secs(30);
async fn run_send_loop(
mut sender: WsSender,
mut subscription: BroadcastSubscription,
mut error_receiver: mpsc::UnboundedReceiver<String>,
) {
let mut ping_interval = tokio::time::interval(PING_INTERVAL);
ping_interval.tick().await;
loop {
tokio::select! {
Some(event) = subscription.recv() => {
@@ -123,6 +131,11 @@ async fn run_send_loop(
break;
}
}
_ = ping_interval.tick() => {
if sender.send(Message::Ping(Vec::new().into())).await.is_err() {
break;
}
}
else => break,
}
}
@@ -143,7 +156,7 @@ fn serialize_broadcast_event(event: &BroadcastEvent) -> Option<String> {
fn handle_client_message(
state: &AppState,
error_sender: &mpsc::UnboundedSender<String>,
connection_id: &str,
connection_id: &UserId,
text: &str,
) {
let Ok(message) = serde_json::from_str::<ClientMessage>(text) else {
@@ -160,7 +173,7 @@ fn handle_client_message(
fn handle_place_pixel(
state: &AppState,
error_sender: &mpsc::UnboundedSender<String>,
connection_id: &str,
connection_id: &UserId,
x: u32,
y: u32,
color: u32,