0508 1.2s轮询同步
This commit is contained in:
lix-2026
@@ -0,0 +1,256 @@
|
||||
use crate::context::RequestContext;
|
||||
use crate::error::WebError;
|
||||
use crate::routes::local_folder_source::decode_local_id_segment;
|
||||
use axum::extract::{Extension, Query};
|
||||
use axum::http::{HeaderMap, HeaderName, HeaderValue};
|
||||
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
|
||||
use futures_util::stream;
|
||||
use notify::event::ModifyKind;
|
||||
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
use std::convert::Infallible;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct LocalFolderEventsQuery {
|
||||
pub root_uri: String,
|
||||
pub document_id: Option<String>,
|
||||
}
|
||||
|
||||
pub async fn local_folder_events(
|
||||
Extension(context): Extension<RequestContext>,
|
||||
Query(query): Query<LocalFolderEventsQuery>,
|
||||
) -> Result<
|
||||
(
|
||||
HeaderMap,
|
||||
Sse<impl futures_util::Stream<Item = Result<SseEvent, Infallible>>>,
|
||||
),
|
||||
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 (sender, receiver) = mpsc::unbounded_channel::<Value>();
|
||||
let (watch_sender, mut watch_receiver) =
|
||||
mpsc::unbounded_channel::<notify::Result<Event>>();
|
||||
let mut watcher = RecommendedWatcher::new(
|
||||
move |result| {
|
||||
let _ = watch_sender.send(result);
|
||||
},
|
||||
Config::default(),
|
||||
)
|
||||
.map_err(|error| {
|
||||
WebError::internal(format!("本地文件事件监听启动失败: {error}")).with_context(&context)
|
||||
})?;
|
||||
watcher
|
||||
.watch(&canonical_root, RecursiveMode::Recursive)
|
||||
.map_err(|error| {
|
||||
WebError::internal(format!("本地文件夹监听失败: {error}")).with_context(&context)
|
||||
})?;
|
||||
|
||||
let root_for_task = canonical_root.clone();
|
||||
let root_uri = query.root_uri.clone();
|
||||
tokio::spawn(async move {
|
||||
let _watcher = watcher;
|
||||
while let Some(result) = watch_receiver.recv().await {
|
||||
let Ok(event) = result else {
|
||||
continue;
|
||||
};
|
||||
if !should_emit_event_kind(&event.kind) {
|
||||
continue;
|
||||
}
|
||||
for path in event.paths {
|
||||
if !is_markdown_path(&path) {
|
||||
continue;
|
||||
}
|
||||
let Some(relative_path) = relative_path_string(&root_for_task, &path) else {
|
||||
continue;
|
||||
};
|
||||
if let Some(expected) = document_relative_path.as_deref() {
|
||||
if expected != relative_path {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
let payload = json!({
|
||||
"sourceKind": "local_folder",
|
||||
"rootUri": root_uri,
|
||||
"relativePath": relative_path,
|
||||
"documentId": format!("local-md:{}", encode_local_id_segment(&relative_path)),
|
||||
"eventKind": format!("{:?}", event.kind),
|
||||
"revision": event_revision(&path),
|
||||
});
|
||||
if sender.send(payload).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let initial = json!({
|
||||
"sourceKind": "local_folder",
|
||||
"rootUri": query.root_uri,
|
||||
"documentId": query.document_id,
|
||||
"revision": system_time_ms(SystemTime::now()),
|
||||
});
|
||||
let stream = stream::unfold((Some(initial), receiver), |(initial, mut receiver)| async move {
|
||||
if let Some(payload) = initial {
|
||||
return Some((Ok(stream_event("ready", &payload)), (None, receiver)));
|
||||
}
|
||||
receiver
|
||||
.recv()
|
||||
.await
|
||||
.map(|payload| (Ok(stream_event("change", &payload)), (None, receiver)))
|
||||
});
|
||||
|
||||
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<PathBuf, WebError> {
|
||||
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<String> {
|
||||
let trimmed = document_id.trim();
|
||||
let encoded = trimmed.strip_prefix("local-md:")?;
|
||||
decode_local_id_segment(encoded).ok()
|
||||
}
|
||||
|
||||
fn relative_path_string(root: &Path, path: &Path) -> Option<String> {
|
||||
path.strip_prefix(root)
|
||||
.ok()
|
||||
.map(|relative| relative.to_string_lossy().replace('\\', "/"))
|
||||
.filter(|relative| !relative.is_empty())
|
||||
}
|
||||
|
||||
fn is_markdown_path(path: &Path) -> bool {
|
||||
path.extension()
|
||||
.and_then(|extension| extension.to_str())
|
||||
.map(|extension| {
|
||||
extension.eq_ignore_ascii_case("md") || extension.eq_ignore_ascii_case("markdown")
|
||||
})
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
fn should_emit_event_kind(kind: &EventKind) -> bool {
|
||||
match kind {
|
||||
EventKind::Create(_) | EventKind::Remove(_) => true,
|
||||
EventKind::Modify(modify_kind) => matches!(
|
||||
modify_kind,
|
||||
ModifyKind::Any | ModifyKind::Data(_) | ModifyKind::Name(_) | ModifyKind::Metadata(_)
|
||||
),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_local_id_segment(value: &str) -> String {
|
||||
let mut encoded = String::with_capacity(value.len());
|
||||
for byte in value.as_bytes() {
|
||||
let character = *byte as char;
|
||||
if character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | '-') {
|
||||
encoded.push(character);
|
||||
} else {
|
||||
encoded.push('~');
|
||||
encoded.push_str(&format!("{byte:02X}"));
|
||||
}
|
||||
}
|
||||
encoded
|
||||
}
|
||||
|
||||
fn event_revision(path: &Path) -> u128 {
|
||||
std::fs::metadata(path)
|
||||
.ok()
|
||||
.and_then(|metadata| metadata.modified().ok())
|
||||
.map(system_time_ms)
|
||||
.unwrap_or_else(|| system_time_ms(SystemTime::now()))
|
||||
}
|
||||
|
||||
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 markdown_path_filter_accepts_markdown_files_only() {
|
||||
assert!(is_markdown_path(Path::new("README.md")));
|
||||
assert!(is_markdown_path(Path::new("README.markdown")));
|
||||
assert!(!is_markdown_path(Path::new("image.png")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_kind_filter_ignores_access_events() {
|
||||
assert!(should_emit_event_kind(&EventKind::Create(notify::event::CreateKind::File)));
|
||||
assert!(should_emit_event_kind(&EventKind::Modify(ModifyKind::Data(
|
||||
notify::event::DataChange::Content,
|
||||
))));
|
||||
assert!(!should_emit_event_kind(&EventKind::Access(
|
||||
notify::event::AccessKind::Read,
|
||||
)));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user