20260513 mindmap优化01

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

@@ -57,8 +57,9 @@ pub async fn events(
                 state.polls += 1;
                 sleep(Duration::from_millis(poll_ms)).await;
 
+                let poll_query = live_poll_query(&state.query);
                 let Ok((workspace_id, overview)) =
-                    load_stream_overview(state.app_state.config(), &state.context, &state.query)
+                    load_stream_overview(state.app_state.config(), &state.context, &poll_query)
                         .await
                 else {
                     return None;
@@ -76,7 +77,7 @@ pub async fn events(
                         let Ok(payload) = build_stream_delta_payload(
                             state.app_state.config(),
                             &state.context,
-                            &state.query,
+                            &poll_query,
                             &workspace_id,
                             &overview,
                             change.cursor,
@@ -91,18 +92,15 @@ pub async fn events(
                         return Some((Ok(stream_event("delta", &payload)), Some(state)));
                     }
                     StreamChangeKind::Resync => {
-                        let mut next_query = state.query.clone();
-                        next_query.cursor = change.cursor;
                         let Ok(snapshot_payload) = load_stream_snapshot(
                             state.app_state.config(),
                             &state.context,
-                            &next_query,
+                            &poll_query,
                         )
                         .await
                         else {
                             return None;
                         };
-                        state.query = next_query;
                         state.current_cursor = read_stream_cursor_from_payload(&snapshot_payload);
                         return Some((
                             Ok(stream_event(
@@ -157,6 +155,14 @@ struct StreamPollState {
     initial_emitted: bool,
 }
 
+fn live_poll_query(query: &StreamSnapshotQuery) -> StreamSnapshotQuery {
+    let mut next = query.clone();
+    // Convex bridgeLogs 的 cursor 是“向更旧记录翻页”，不是 live tail 的起点；
+    // 实时轮询必须始终查最新窗口，再用 current_cursor 在 Rust 侧比较增量。
+    next.cursor = None;
+    next
+}
+
 fn stream_event(event_name: &str, payload: &Value) -> Event {
     let event_id = payload
         .get("revision")
@@ -183,6 +189,7 @@ fn stream_event(event_name: &str, payload: &Value) -> Event {
 #[cfg(test)]
 mod tests {
     use crate::app::{build_app, AppConfig, AppState};
+    use crate::routes::stream_support::StreamSnapshotQuery;
     use axum::body::{to_bytes, Body};
     use axum::http::{Request, StatusCode};
     use tower::util::ServiceExt;
@@ -289,4 +296,20 @@ mod tests {
         assert!(text.contains("id: "));
         assert!(text.contains("\"revision\""));
     }
+
+    #[test]
+    fn live_poll_query_drops_bridge_pagination_cursor() {
+        let query = StreamSnapshotQuery {
+            workspace_id: Some("ws_demo".into()),
+            cursor: Some(r#"{"createdAt":"2026-05-12T00:00:00Z","id":"clog_1"}"#.into()),
+            poll_ms: Some(250),
+            ..StreamSnapshotQuery::default()
+        };
+
+        let live_query = super::live_poll_query(&query);
+
+        assert_eq!(live_query.workspace_id, Some("ws_demo".into()));
+        assert_eq!(live_query.poll_ms, Some(250));
+        assert_eq!(live_query.cursor, None);
+    }
 }