feat(qbzd): unified event bus + SSE endpoint (Phase 3)

Playback event loop polls QbzCore player state with adaptive timing
(250ms playing, 1s paused, 5s idle) and broadcasts PlaybackSnapshot
events to the event bus.

New SSE endpoint at /api/events streams all DaemonEvents to clients.
Broadcast channel bounded to 256 events with lagged client handling.

cc

crates/qbzd/Cargo.toml

@@ -26,7 +26,9 @@ tower-http = { version = "0.6", features = ["cors"] }
 
 # Async
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
 async-trait = { workspace = true }
+futures-util = { workspace = true }
 
 # Serialization
 serde = { workspace = true }

crates/qbzd/src/api/events.rs

@@ -0,0 +1,96 @@
+//! SSE (Server-Sent Events) endpoint for real-time event streaming.
+//!
+//! Clients connect to `/api/events` and receive a continuous stream of
+//! DaemonEvents formatted as SSE. Slow clients that fall behind the
+//! broadcast buffer are disconnected (SSE auto-reconnect handles this).
+
+use std::convert::Infallible;
+use std::sync::Arc;
+
+use axum::response::sse::{Event, KeepAlive, Sse};
+use futures_util::stream::Stream;
+use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::StreamExt;
+
+use crate::adapter::{DaemonEvent, PlaybackSnapshot, RuntimeEvent};
+use crate::daemon::DaemonCore;
+
+/// SSE endpoint handler. Each connected client gets its own broadcast receiver.
+pub async fn sse_handler(
+    daemon: Arc<DaemonCore>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let rx = daemon.event_bus.subscribe();
+    let stream = BroadcastStream::new(rx).filter_map(|result| match result {
+        Ok(event) => Some(Ok(format_sse_event(&event))),
+        Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
+            log::warn!("[SSE] Client lagged, dropped {} events", n);
+            Some(Ok(Event::default().comment(format!("lagged:{}", n))))
+        }
+    });
+
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
+/// Convert a DaemonEvent to an SSE Event with typed event name.
+fn format_sse_event(event: &DaemonEvent) -> Event {
+    match event {
+        DaemonEvent::Core(core_event) => {
+            let (event_type, data) = format_core_event(core_event);
+            Event::default().event(event_type).data(data)
+        }
+        DaemonEvent::Playback(snapshot) => Event::default()
+            .event("playback")
+            .data(serde_json::to_string(snapshot).unwrap_or_default()),
+        DaemonEvent::Runtime(runtime_event) => Event::default()
+            .event("runtime")
+            .data(serde_json::to_string(runtime_event).unwrap_or_default()),
+    }
+}
+
+/// Map CoreEvent variants to SSE event type + JSON data.
+fn format_core_event(event: &qbz_models::CoreEvent) -> (&'static str, String) {
+    use qbz_models::CoreEvent;
+    match event {
+        CoreEvent::TrackStarted { track, .. } => (
+            "track-started",
+            serde_json::to_string(track).unwrap_or_default(),
+        ),
+        CoreEvent::TrackEnded { track_id } => (
+            "track-ended",
+            serde_json::json!({"track_id": track_id}).to_string(),
+        ),
+        CoreEvent::QueueUpdated { state } => (
+            "queue",
+            serde_json::to_string(state).unwrap_or_default(),
+        ),
+        CoreEvent::FavoritesUpdated { favorite_type } => (
+            "favorites-updated",
+            serde_json::json!({"type": favorite_type}).to_string(),
+        ),
+        CoreEvent::PlaylistCreated { playlist } => (
+            "playlist-created",
+            serde_json::to_string(playlist).unwrap_or_default(),
+        ),
+        CoreEvent::PlaylistUpdated { playlist_id } => (
+            "playlist-updated",
+            serde_json::json!({"playlist_id": playlist_id}).to_string(),
+        ),
+        CoreEvent::Error {
+            code,
+            message,
+            recoverable,
+        } => (
+            "error",
+            serde_json::json!({"code": code, "message": message, "recoverable": recoverable})
+                .to_string(),
+        ),
+        CoreEvent::LoggedIn { session } => (
+            "logged-in",
+            serde_json::json!({"user_id": session.user_id, "display_name": session.display_name})
+                .to_string(),
+        ),
+        CoreEvent::LoggedOut => ("logged-out", "{}".to_string()),
+        // Catch-all for any future CoreEvent variants
+        _ => ("core-event", format!("{:?}", event)),
+    }
+}

crates/qbzd/src/api/mod.rs

@@ -0,0 +1 @@
+pub mod events;

crates/qbzd/src/daemon.rs

@@ -99,6 +99,9 @@ pub async fn run(mut config: DaemonConfig) -> Result<(), String> {
         }
     }
 
+    // Start playback state polling loop (broadcasts to event bus)
+    spawn_playback_loop(daemon.core.clone(), daemon.event_bus.clone());
+
     log::info!(
         "[qbzd] Daemon ready on {}:{}",
         config.server.bind,
@@ -192,6 +195,72 @@ fn load_oauth_token() -> Option<String> {
     }
 }
 
+/// Spawn the playback state polling loop.
+/// Reads player state and broadcasts PlaybackSnapshot events.
+/// Adaptive polling: 250ms playing, 1s paused, 5s idle.
+fn spawn_playback_loop(
+    core: Arc<QbzCore<DaemonAdapter>>,
+    event_tx: broadcast::Sender<DaemonEvent>,
+) {
+    tokio::spawn(async move {
+        let mut last_position: u64 = 0;
+        let mut last_is_playing = false;
+        let mut last_track_id: u64 = 0;
+
+        loop {
+            let player = core.player();
+            let state = &player.state;
+
+            let is_playing = state.is_playing();
+            let track_id = state.current_track_id();
+            let position = state.current_position();
+            let duration = state.duration();
+            let volume = state.volume();
+            let sample_rate = state.get_sample_rate();
+            let bit_depth = state.get_bit_depth();
+
+            let track_cleared = track_id == 0 && last_track_id != 0;
+            let should_emit = (track_id != 0
+                && (is_playing != last_is_playing
+                    || track_id != last_track_id
+                    || (is_playing && position != last_position)))
+                || track_cleared;
+
+            if should_emit {
+                let snapshot = crate::adapter::PlaybackSnapshot {
+                    state: if is_playing {
+                        "Playing".to_string()
+                    } else if track_id != 0 {
+                        "Paused".to_string()
+                    } else {
+                        "Stopped".to_string()
+                    },
+                    track_id,
+                    position_secs: position,
+                    duration_secs: duration,
+                    volume,
+                    sample_rate,
+                    bit_depth,
+                };
+                let _ = event_tx.send(DaemonEvent::Playback(snapshot));
+                last_position = position;
+                last_is_playing = is_playing;
+                last_track_id = track_id;
+            }
+
+            // Adaptive polling
+            let sleep_ms = if is_playing {
+                250
+            } else if track_id == 0 {
+                5000
+            } else {
+                1000
+            };
+            tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await;
+        }
+    });
+}
+
 /// Build the Axum HTTP router.
 fn build_router(daemon: Arc<DaemonCore>) -> axum::Router {
     use axum::routing::get;
@@ -212,6 +281,13 @@ fn build_router(daemon: Arc<DaemonCore>) -> axum::Router {
                 move || status_handler(d.clone())
             }),
         )
+        .route(
+            "/api/events",
+            get({
+                let d = daemon.clone();
+                move || crate::api::events::sse_handler(d.clone())
+            }),
+        )
 }
 
 async fn ping_handler() -> &'static str {

crates/qbzd/src/main.rs

@@ -1,3 +1,4 @@
+mod api;
 mod config;
 mod daemon;
 mod adapter;