mnote/rust/crates/mnote-web/src/routes/sse.rs

use crate::app::AppState;
use crate::context::RequestContext;
use crate::error::WebError;
use crate::routes::stream_support::{load_stream_snapshot, StreamSnapshotQuery};
use axum::extract::{Extension, Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::stream;
use serde_json::Value;
use std::convert::Infallible;
use std::time::Duration;

pub async fn events(
    State(state): State<AppState>,
    Extension(context): Extension<RequestContext>,
    Query(query): Query<StreamSnapshotQuery>,
) -> Result<Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>>, WebError> {
    let payload = load_stream_snapshot(state.config(), &context, &query).await?;
    let event = snapshot_event(&payload);

    Ok(Sse::new(stream::iter(vec![Ok(event)])).keep_alive(
        KeepAlive::new()
            .interval(Duration::from_secs(15))
            .text("keepalive"),
    ))
}

fn snapshot_event(payload: &Value) -> Event {
    Event::default()
        .event("snapshot")
        .json_data(payload)
        .expect("SSE snapshot 事件必须可序列化")
}

#[cfg(test)]
mod tests {
    use crate::app::{build_app, AppConfig, AppState};
    use axum::body::{to_bytes, Body};
    use axum::http::{Request, StatusCode};
    use tower::util::ServiceExt;

    fn app() -> axum::Router {
        build_app(AppState::new(AppConfig {
            service_name: "mnote-web".into(),
            service_version: "0.1.0".into(),
            bind_addr: "127.0.0.1:0".into(),
            hermes_base_path: "/api/hermes".into(),
            compat_next_base_path: "/api/compat/next".into(),
            convex_url: None,
            convex_admin_key: None,
            allow_dev_fixtures: true,
            query_fixtures_json: Some(r#"{"sidebar:datasetList":{"active_workspace_id":"ws_demo","documents":[{"id":"page_root","workspace_id":"ws_demo","title":"工作区首页","parent_id":null,"sort_order":0,"is_starred":true,"is_template":false,"created_at":"2026-04-16T00:00:00Z","updated_at":"2026-04-16T00:00:00Z"},{"id":"page_child","workspace_id":"ws_demo","title":"子页面","parent_id":"page_root","sort_order":1,"is_starred":false,"is_template":false,"created_at":"2026-04-16T00:00:00Z","updated_at":"2026-04-16T00:00:00Z"}],"trashed_documents":[],"media_assets":[],"trashed_media_assets":[],"mindmap_assets":[],"trashed_mindmap_assets":[],"table_assets":[],"trashed_table_assets":[],"mindmap_docs":[],"mindmap_asset_children":{}},"bridgeLogs:listWorkspaceOverview":{"workspace_id":"ws_demo","command_logs":[{"command_id":"cmd_1","request_id":"req_1","status":"applied","created_at":"2026-04-16T00:00:00Z"}],"domain_events":[{"command_id":"cmd_1","status":"published","created_at":"2026-04-16T00:00:00Z"}],"next_cursor":"cursor_demo","has_more":false,"filters":{"command_status":null,"event_status":null,"target_page_id":null,"target_block_id":null,"aggregate_type":null,"aggregate_id":null},"generated_at":"2026-04-16T00:00:00Z"}}"#.into()),
            mutation_fixtures_json: None,
            dev_user_id: "dev-user".into(),
            dev_user_name: "开发用户".into(),
            dev_user_email: "dev@mnote.local".into(),
        }))
    }

    #[tokio::test]
    async fn sse_route_returns_workspace_snapshot_event() {
        let response = app()
            .oneshot(
                Request::builder()
                    .uri("/api/stream/events?workspaceId=ws_demo")
                    .body(Body::empty())
                    .expect("request"),
            )
            .await
            .expect("response");

        assert_eq!(response.status(), StatusCode::OK);
        let body = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("body");
        let text = String::from_utf8(body.to_vec()).expect("utf8");
        assert!(text.contains("event: snapshot") || text.contains("event:snapshot"));
        assert!(text.contains("\"scope\":\"workspace\""));
        assert!(text.contains("\"workspaceId\":\"ws_demo\""));
    }
}