fix local markdown attachment regressions

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

@@ -10,16 +10,16 @@ use crate::routes::snapshot_support::ProjectionSnapshot;
 use axum::extract::{Extension, Query, State};
 use axum::http::{HeaderMap, HeaderName, HeaderValue};
 use axum::response::sse::{Event as SseEvent, Sse};
-use futures_util::StreamExt;
 use futures_util::stream;
+use futures_util::StreamExt;
 use serde::Deserialize;
-use serde_json::{Value, json};
+use serde_json::{json, Value};
 use std::convert::Infallible;
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::broadcast::error::RecvError;
-use tokio::time::{MissedTickBehavior, interval, timeout};
+use tokio::time::timeout;
 
 type BoxedEventStream =
     Pin<Box<dyn futures_util::Stream<Item = Result<SseEvent, Infallible>> + Send>>;
@@ -70,7 +70,12 @@ async fn build_document_events_stream(
         .document_id
         .as_deref()
         .and_then(local_markdown_relative_path_from_document_id)
-        .or_else(|| query.resource_path.as_deref().and_then(normalize_resource_event_path));
+        .or_else(|| {
+            query
+                .resource_path
+                .as_deref()
+                .and_then(normalize_resource_event_path)
+        });
     let subscription = state
         .local_folder_watcher_registry()
         .subscribe(&canonical_root)
@@ -147,26 +152,15 @@ async fn build_tree_live_stream(
         &file_tree_snapshot,
     );
 
-    let subscription = match state
+    let subscription = state
         .local_folder_watcher_registry()
         .subscribe(&canonical_root)
-    {
-        Ok(subscription) => subscription,
-        Err(error) => {
-            tracing::warn!(
-                error = %error,
-                root_uri = %root_uri,
-                "local_folder tree live watcher unavailable; falling back to revision polling"
-            );
-            let stream = build_tree_live_polling_stream(
-                root_uri,
-                workspace_id,
-                revision.revision,
-                initial_payload,
-            );
-            return Ok((HeaderMap::new(), stream));
-        }
-    };
+        .map_err(|error| {
+            WebError::internal(format!(
+                "local_folder tree live watcher unavailable: {error}"
+            ))
+            .with_context(&context)
+        })?;
 
     let stream = stream::unfold(
         (Some(initial_payload), subscription, root_uri, workspace_id),
@@ -224,81 +218,6 @@ async fn build_tree_live_stream(
     Ok((HeaderMap::new(), stream))
 }
 
-fn build_tree_live_polling_stream(
-    root_uri: String,
-    workspace_id: String,
-    initial_revision: String,
-    initial_payload: Value,
-) -> BoxedEventStream {
-    let mut poll_interval = interval(Duration::from_millis(1_200));
-    poll_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
-
-    stream::unfold(
-        Some(TreeLivePollingState {
-            root_uri,
-            workspace_id,
-            current_revision: initial_revision,
-            initial_payload: Some(initial_payload),
-            poll_interval,
-        }),
-        |state| async move {
-            let mut state = state?;
-            if let Some(payload) = state.initial_payload.take() {
-                return Some((Ok(stream_event("snapshot", &payload)), Some(state)));
-            }
-
-            loop {
-                state.poll_interval.tick().await;
-                let Some((next_revision, resync_payload)) = tree_live_polling_resync_payload(
-                    &state.root_uri,
-                    &state.workspace_id,
-                    &state.current_revision,
-                ) else {
-                    continue;
-                };
-                state.current_revision = next_revision;
-                return Some((Ok(stream_event("resync", &resync_payload)), Some(state)));
-            }
-        },
-    )
-    .boxed()
-}
-
-struct TreeLivePollingState {
-    root_uri: String,
-    workspace_id: String,
-    current_revision: String,
-    initial_payload: Option<Value>,
-    poll_interval: tokio::time::Interval,
-}
-
-fn tree_live_polling_resync_payload(
-    root_uri: &str,
-    workspace_id: &str,
-    current_revision: &str,
-) -> Option<(String, Value)> {
-    let next_revision = local_folder_watch_revision(root_uri).ok()?;
-    if next_revision.revision == current_revision {
-        return None;
-    }
-    let resync_payload = rebuild_tree_resync_payload(root_uri, workspace_id)?;
-    Some((next_revision.revision, resync_payload))
-}
-
-fn rebuild_tree_resync_payload(root_uri: &str, workspace_id: &str) -> Option<Value> {
-    let revision = local_folder_watch_revision(root_uri).ok()?;
-    let sidebar_snapshot = load_local_folder_page_tree_snapshot(root_uri).ok()?;
-    let file_tree_snapshot = load_local_folder_file_tree_snapshot(root_uri).ok()?;
-    Some(build_tree_snapshot_payload(
-        root_uri,
-        workspace_id,
-        &revision.revision,
-        "resync",
-        &sidebar_snapshot,
-        &file_tree_snapshot,
-    ))
-}
-
 fn parent_relative_path_for_watch_path(relative_path: &str) -> String {
     let normalized = relative_path.trim().trim_matches('/').replace('\\', "/");
     if normalized.is_empty() || normalized == "." {
@@ -472,7 +391,7 @@ fn stream_event(event_name: &str, payload: &Value) -> SseEvent {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::app::{AppConfig, AppState, build_app};
+    use crate::app::{build_app, AppConfig, AppState};
     use crate::routes::local_folder_source::initialize_local_workspace_for_actor;
     use axum::body::Body;
     use axum::http::Request;
@@ -614,7 +533,10 @@ mod tests {
         });
 
         assert!(document_event_targets_relative_path(&payload, &expected));
-        assert!(!document_event_targets_relative_path(&other_payload, &expected));
+        assert!(!document_event_targets_relative_path(
+            &other_payload,
+            &expected
+        ));
         assert!(
             normalize_resource_event_path("../escape.pdf").is_none(),
             "resource watch path 不能越过 root"
@@ -669,40 +591,6 @@ mod tests {
         assert!(payload["data"]["tree"].is_object());
     }
 
-    #[test]
-    fn tree_live_polling_resync_payload_tracks_revision_changes() {
-        let root = test_root("tree-live-polling-resync");
-        std::fs::write(root.join("README.md"), "# Initial\n").expect("write initial");
-        let root_uri = format!("file://{}", root.display());
-        let workspace_id =
-            local_workspace_id_from_root_uri(&root_uri).expect("resolve local workspace id");
-        let initial_revision = local_folder_watch_revision(&root_uri)
-            .expect("initial revision")
-            .revision;
-
-        assert!(
-            tree_live_polling_resync_payload(&root_uri, &workspace_id, &initial_revision).is_none(),
-            "revision 未变化时不应发送 resync"
-        );
-
-        std::fs::create_dir_all(root.join("docs")).expect("create docs");
-        std::fs::write(root.join("docs/new.md"), "# New\n").expect("write new markdown");
-        let (next_revision, payload) =
-            tree_live_polling_resync_payload(&root_uri, &workspace_id, &initial_revision)
-                .expect("revision change should build resync payload");
-
-        assert_ne!(next_revision, initial_revision);
-        assert_eq!(payload["kind"], "resync");
-        assert_eq!(payload["sourceKind"], "local_folder");
-        assert_eq!(payload["rootUri"], root_uri);
-        assert_eq!(payload["workspaceId"], workspace_id);
-        assert_eq!(payload["revision"], next_revision);
-        assert!(payload["data"]["dataset"]["kernel_sidebar_projection"].is_object());
-        assert!(payload["data"]["dataset"]["kernel_file_tree_projection"].is_object());
-
-        let _ = std::fs::remove_dir_all(root);
-    }
-
     #[test]
     fn local_folder_watch_batch_payload_declares_changed_paths_and_parents() {
         let root = test_root("tree-live-watch-batch");
@@ -741,13 +629,11 @@ mod tests {
                     && parent["reason"].as_str() == Some("child-watch")),
             "watch batch 应声明 docs affected parent: {payload}"
         );
-        assert!(
-            payload["eventKinds"]
-                .as_array()
-                .expect("event kinds")
-                .iter()
-                .any(|kind| kind.as_str() == Some("Modify(Data)"))
-        );
+        assert!(payload["eventKinds"]
+            .as_array()
+            .expect("event kinds")
+            .iter()
+            .any(|kind| kind.as_str() == Some("Modify(Data)")));
 
         let _ = std::fs::remove_dir_all(root);
     }