use crate::app::AppState; use crate::context::RequestContext; use crate::error::WebError; use crate::routes::local_folder_source::decode_local_id_segment; use axum::extract::{Extension, Query, State}; use axum::http::{HeaderMap, HeaderName, HeaderValue}; use axum::response::sse::{Event as SseEvent, KeepAlive, Sse}; use futures_util::stream; use serde::Deserialize; use serde_json::{json, Value}; use std::convert::Infallible; use std::path::PathBuf; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::broadcast::error::RecvError; #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LocalFolderEventsQuery { pub root_uri: String, pub document_id: Option, } pub async fn local_folder_events( State(state): State, Extension(context): Extension, Query(query): Query, ) -> Result< ( HeaderMap, Sse>>, ), WebError, > { let root = parse_file_root_uri(&query.root_uri)?; let canonical_root = root.canonicalize().map_err(|error| { WebError::bad_request_code( "local_folder_unavailable", format!("无法访问本地文件夹: {error}"), ) .with_context(&context) })?; let document_relative_path = query .document_id .as_deref() .and_then(local_markdown_relative_path_from_document_id); let subscription = state .local_folder_watcher_registry() .subscribe(&canonical_root) .map_err(|error| WebError::internal(error).with_context(&context))?; let initial = json!({ "sourceKind": "local_folder", "rootUri": subscription.root_uri(), "documentId": query.document_id, "revision": system_time_ms(SystemTime::now()), }); let stream = stream::unfold( (Some(initial), subscription, document_relative_path), |(initial, mut subscription, document_relative_path)| async move { if let Some(payload) = initial { return Some(( Ok(stream_event("ready", &payload)), (None, subscription, document_relative_path), )); } loop { match subscription.receiver.recv().await { Ok(payload) => { if let Some(expected) = document_relative_path.as_deref() { let relative_path = payload .get("relativePath") .and_then(Value::as_str) .unwrap_or_default(); if expected != relative_path { continue; } } return Some(( Ok(stream_event("change", &payload)), (None, subscription, document_relative_path), )); } Err(RecvError::Lagged(_)) => continue, Err(RecvError::Closed) => return None, } } }, ); let mut headers = HeaderMap::new(); if let Ok(name) = HeaderName::from_lowercase(b"x-mnote-web-owner") { headers.insert(name, HeaderValue::from_static("mnote-web")); } if let Ok(name) = HeaderName::from_lowercase(b"x-mnote-local-folder-events-owner") { headers.insert(name, HeaderValue::from_static("rust-web")); } Ok(( headers, Sse::new(stream).keep_alive( KeepAlive::new() .interval(Duration::from_secs(15)) .text("keepalive"), ), )) } fn parse_file_root_uri(root_uri: &str) -> Result { let trimmed = root_uri.trim(); let Some(path) = trimmed.strip_prefix("file://") else { return Err(WebError::bad_request_code( "local_folder_root_invalid", "本地文件夹 rootUri 必须是 file:// URI", )); }; Ok(PathBuf::from(path)) } fn local_markdown_relative_path_from_document_id(document_id: &str) -> Option { let trimmed = document_id.trim(); let encoded = trimmed.strip_prefix("local-md:")?; decode_local_id_segment(encoded).ok() } fn system_time_ms(time: SystemTime) -> u128 { time.duration_since(UNIX_EPOCH) .map(|duration| duration.as_millis()) .unwrap_or(0) } fn stream_event(event_name: &str, payload: &Value) -> SseEvent { let id = payload .get("revision") .and_then(|value| { value .as_str() .map(ToOwned::to_owned) .or_else(|| value.as_u64().map(|number| number.to_string())) }) .unwrap_or_else(|| "0".to_string()); SseEvent::default() .event(event_name) .id(id) .json_data(payload) .expect("SSE 事件必须可序列化") } #[cfg(test)] mod tests { use super::*; #[test] fn local_markdown_document_id_maps_to_relative_path() { assert_eq!( local_markdown_relative_path_from_document_id("local-md:docs~2FREADME.md").as_deref(), Some("docs/README.md") ); } #[test] fn parse_file_root_uri_requires_file_scheme() { assert!(parse_file_root_uri("file:///tmp/example").is_ok()); assert!(parse_file_root_uri("/tmp/example").is_err()); } }