Merge pull request #13 from CurtisTechSolutions/feat/live-match-stream

feat: stream committed match states over WebSockets

Author: nosam1998

Cargo.lock

@@ -42,11 +42,14 @@ sqlx = { version = "0.8.6", default-features = false, features = ["runtime-tokio
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "net", "signal", "time"] }
 tempfile = "3"
 
-axum = { version = "0.8.9", default-features = false, features = ["http1", "json", "query", "tokio"] }
+axum = { version = "0.8.9", default-features = false, features = ["http1", "json", "query", "tokio", "ws"] }
 tower = { version = "0.5", features = ["util"] }
 
 uuid = { version = "1", features = ["v4"] }
 
+tokio-tungstenite = { version = "0.28", default-features = false, features = ["connect", "handshake"] }
+futures-util = { version = "0.3", features = ["sink"] }
+
 [workspace.lints.rust]
 unsafe_code = "forbid"
 missing_docs = "warn"

Cargo.toml

@@ -11,9 +11,9 @@ gfa-service.workspace = true
 axum.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+tokio.workspace = true
 
 [dev-dependencies]
-tokio.workspace = true
 tower.workspace = true
 
 [lints]

crates/adapters/gfa-http/Cargo.toml

@@ -5,6 +5,9 @@
 //! It does not issue seat credentials and must not be mounted behind a public proxy.
 mod access;
 mod error;
+mod stream;
+
+pub use stream::LiveUpdates;
 
 use axum::{
     extract::{
@@ -33,14 +36,34 @@ pub const MAX_BODY_BYTES: usize = 64 * 1024;
 /// Serve using `into_make_service_with_connect_info::<SocketAddr>()`; requests
 /// without peer information are refused. The address must already be loopback.
 pub fn local_router(service: Arc<GameService>, address: SocketAddr) -> Result<Router, ApiError> {
+    build_router(service, address, None)
+}
+
+/// Construct local REST and WebSocket routes using the service's commit observer.
+///
+/// The host must install this same LiveUpdates instance with GameService::with_observer.
+/// Periodic reconciliation also catches commits from other hosts sharing the store.
+pub fn local_router_with_updates(
+    service: Arc<GameService>,
+    address: SocketAddr,
+    updates: Arc<LiveUpdates>,
+) -> Result<Router, ApiError> {
+    build_router(service, address, Some(updates))
+}
+
+fn build_router(
+    service: Arc<GameService>,
+    address: SocketAddr,
+    updates: Option<Arc<LiveUpdates>>,
+) -> Result<Router, ApiError> {
     if !address.ip().is_loopback() || address.port() == 0 {
         return Err(ApiError::new(
             "INVALID_CONFIG",
             "A bound loopback listener is required",
             "Bind 127.0.0.1 or ::1, then pass listener.local_addr().",
         ));
     }
-    Ok(Router::new()
+    let mut router = Router::new()
         .route("/healthz", get(health))
         .route("/v1/games", get(games))
         .route("/v1/games/{game_id}", get(game))
@@ -50,7 +73,14 @@ pub fn local_router(service: Arc<GameService>, address: SocketAddr) -> Result<Ro
         .route("/v1/matches/{id}/actions", post(make_move))
         .route("/v1/matches/{id}/replay", get(replay))
         .fallback(not_found)
-        .method_not_allowed_fallback(method_not_allowed)
+        .method_not_allowed_fallback(method_not_allowed);
+    if let Some(updates) = updates {
+        router = router.route(
+            "/v1/matches/{id}/stream",
+            get(stream::upgrade).layer(axum::Extension(updates)),
+        );
+    }
+    Ok(router
         .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
         .layer(middleware::from_fn_with_state(
             access::LocalAccess::new(address),