724 lines
26 KiB
Rust
724 lines
26 KiB
Rust
use crate::app::AppState;
|
|
use crate::context::RequestContext;
|
|
use crate::error::WebError;
|
|
use crate::routes::local_folder_source::{
|
|
decode_local_id_segment, ensure_local_workspace_read_access_with_state,
|
|
load_local_folder_file_tree_snapshot, load_local_folder_page_tree_snapshot,
|
|
local_folder_watch_revision, local_workspace_id_from_root_uri,
|
|
};
|
|
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::stream;
|
|
use futures_util::StreamExt;
|
|
use serde::Deserialize;
|
|
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::{interval, timeout, MissedTickBehavior};
|
|
|
|
type BoxedEventStream =
|
|
Pin<Box<dyn futures_util::Stream<Item = Result<SseEvent, Infallible>> + Send>>;
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct LocalFolderEventsQuery {
|
|
pub root_uri: String,
|
|
pub document_id: Option<String>,
|
|
pub tree_live: Option<bool>,
|
|
}
|
|
|
|
pub async fn local_folder_events(
|
|
State(state): State<AppState>,
|
|
Extension(context): Extension<RequestContext>,
|
|
Query(query): Query<LocalFolderEventsQuery>,
|
|
) -> Result<(HeaderMap, Sse<BoxedEventStream>), WebError> {
|
|
let canonical_root =
|
|
ensure_local_workspace_read_access_with_state(&state, &context, &query.root_uri)
|
|
.map_err(|error| error.with_context(&context))?;
|
|
|
|
let (mut headers, stream): (HeaderMap, BoxedEventStream) = if query.tree_live.unwrap_or(false) {
|
|
build_tree_live_stream(state, context, canonical_root, query.root_uri).await?
|
|
} else {
|
|
build_document_events_stream(state, context, canonical_root, &query).await?
|
|
};
|
|
|
|
// Apply keepalive via the same type-erased stream path
|
|
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)))
|
|
}
|
|
|
|
/// Build the original document-level external-edit event stream (`ready` / `change`).
|
|
async fn build_document_events_stream(
|
|
state: AppState,
|
|
context: RequestContext,
|
|
canonical_root: PathBuf,
|
|
query: &LocalFolderEventsQuery,
|
|
) -> Result<(HeaderMap, BoxedEventStream), WebError> {
|
|
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() {
|
|
if !document_event_targets_relative_path(&payload, expected) {
|
|
continue;
|
|
}
|
|
}
|
|
return Some((
|
|
Ok(stream_event("change", &payload)),
|
|
(None, subscription, document_relative_path),
|
|
));
|
|
}
|
|
Err(RecvError::Lagged(_)) => continue,
|
|
Err(RecvError::Closed) => return None,
|
|
}
|
|
}
|
|
},
|
|
)
|
|
.boxed();
|
|
|
|
Ok((HeaderMap::new(), stream))
|
|
}
|
|
|
|
/// Build the tree live stream: emits `snapshot` (initial) and `resync` (on watcher change)
|
|
/// with full sidebar + file tree projections.
|
|
///
|
|
/// Reuses `LocalFolderWatcherRegistry` — no second watcher created.
|
|
/// No data is written to Convex command log.
|
|
async fn build_tree_live_stream(
|
|
state: AppState,
|
|
context: RequestContext,
|
|
canonical_root: PathBuf,
|
|
root_uri: String,
|
|
) -> Result<(HeaderMap, BoxedEventStream), WebError> {
|
|
let workspace_id = local_workspace_id_from_root_uri(&root_uri)
|
|
.map_err(|error| error.with_context(&context))?;
|
|
|
|
// Build initial snapshot
|
|
let sidebar_snapshot = load_local_folder_page_tree_snapshot(&root_uri)
|
|
.map_err(|error| error.with_context(&context))?;
|
|
let file_tree_snapshot = load_local_folder_file_tree_snapshot(&root_uri)
|
|
.map_err(|error| error.with_context(&context))?;
|
|
let revision =
|
|
local_folder_watch_revision(&root_uri).map_err(|error| error.with_context(&context))?;
|
|
|
|
let initial_payload = build_tree_snapshot_payload(
|
|
&root_uri,
|
|
&workspace_id,
|
|
&revision.revision,
|
|
"snapshot",
|
|
&sidebar_snapshot,
|
|
&file_tree_snapshot,
|
|
);
|
|
|
|
let subscription = match 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));
|
|
}
|
|
};
|
|
|
|
let stream = stream::unfold(
|
|
(Some(initial_payload), subscription, root_uri, workspace_id),
|
|
|(payload, mut subscription, root_uri, workspace_id)| async move {
|
|
if let Some(payload) = payload {
|
|
return Some((
|
|
Ok(stream_event("snapshot", &payload)),
|
|
(None, subscription, root_uri, workspace_id),
|
|
));
|
|
}
|
|
|
|
loop {
|
|
match subscription.receiver.recv().await {
|
|
Ok(watcher_payload) => {
|
|
let mut watcher_payloads = vec![watcher_payload];
|
|
loop {
|
|
match timeout(Duration::from_millis(120), subscription.receiver.recv())
|
|
.await
|
|
{
|
|
Ok(Ok(next_payload)) => watcher_payloads.push(next_payload),
|
|
Ok(Err(RecvError::Lagged(_))) => continue,
|
|
Ok(Err(RecvError::Closed)) => return None,
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
if let Some(batch_payload) = build_local_folder_watch_batch_payload(
|
|
&root_uri,
|
|
&workspace_id,
|
|
watcher_payloads,
|
|
) {
|
|
return Some((
|
|
Ok(stream_event("watch_batch", &batch_payload)),
|
|
(None, subscription, root_uri, workspace_id),
|
|
));
|
|
}
|
|
let error_payload = build_tree_live_error_payload(
|
|
&root_uri,
|
|
&workspace_id,
|
|
"tree_live_watch_batch_failed",
|
|
"local folder watcher batch payload missing paths",
|
|
);
|
|
return Some((
|
|
Ok(stream_event("tree_error", &error_payload)),
|
|
(None, subscription, root_uri, workspace_id),
|
|
));
|
|
}
|
|
Err(RecvError::Lagged(_)) => continue,
|
|
Err(RecvError::Closed) => return None,
|
|
}
|
|
}
|
|
},
|
|
)
|
|
.boxed();
|
|
|
|
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 == "." {
|
|
return String::new();
|
|
}
|
|
normalized
|
|
.rsplit_once('/')
|
|
.map(|(parent, _)| parent.to_string())
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
fn build_local_folder_watch_batch_payload(
|
|
root_uri: &str,
|
|
workspace_id: &str,
|
|
watcher_payloads: Vec<Value>,
|
|
) -> Option<Value> {
|
|
let revision = local_folder_watch_revision(root_uri).ok()?;
|
|
let mut changed_paths = Vec::new();
|
|
let mut affected_parents = Vec::new();
|
|
let mut event_kinds = Vec::new();
|
|
let mut seen_paths = std::collections::BTreeSet::new();
|
|
let mut seen_parents = std::collections::BTreeSet::new();
|
|
let mut seen_kinds = std::collections::BTreeSet::new();
|
|
for payload in watcher_payloads {
|
|
let relative_path = payload
|
|
.get("relativePath")
|
|
.and_then(Value::as_str)
|
|
.map(str::trim)
|
|
.filter(|value| !value.is_empty())?;
|
|
let event_kind = payload
|
|
.get("eventKind")
|
|
.and_then(Value::as_str)
|
|
.map(str::trim)
|
|
.filter(|value| !value.is_empty())
|
|
.unwrap_or("unknown");
|
|
if seen_paths.insert(relative_path.to_string()) {
|
|
changed_paths.push(json!({
|
|
"relativePath": relative_path,
|
|
"kind": event_kind,
|
|
}));
|
|
}
|
|
if seen_kinds.insert(event_kind.to_string()) {
|
|
event_kinds.push(event_kind.to_string());
|
|
}
|
|
let parent = parent_relative_path_for_watch_path(relative_path);
|
|
if seen_parents.insert(parent.clone()) {
|
|
affected_parents.push(json!({
|
|
"relativePath": parent,
|
|
"reason": "child-watch",
|
|
}));
|
|
}
|
|
}
|
|
if changed_paths.is_empty() {
|
|
return None;
|
|
}
|
|
Some(json!({
|
|
"schema": "mnote.local_folder_watch_batch.v1",
|
|
"kind": "watch_batch",
|
|
"sourceKind": "local_folder",
|
|
"rootUri": root_uri,
|
|
"workspaceId": workspace_id,
|
|
"revision": revision.revision,
|
|
"watchRevision": revision,
|
|
"changedPaths": changed_paths,
|
|
"affectedParents": affected_parents,
|
|
"eventKinds": event_kinds,
|
|
"fallbackResync": false,
|
|
}))
|
|
}
|
|
|
|
fn build_tree_live_error_payload(
|
|
root_uri: &str,
|
|
workspace_id: &str,
|
|
code: &str,
|
|
message: &str,
|
|
) -> Value {
|
|
json!({
|
|
"schema": "mnote.tree_live_error.v1",
|
|
"kind": "error",
|
|
"phase": "tree_live_resync",
|
|
"sourceKind": "local_folder",
|
|
"rootUri": root_uri,
|
|
"workspaceId": workspace_id,
|
|
"code": code,
|
|
"message": message,
|
|
"fallbackResync": true,
|
|
"revision": system_time_ms(SystemTime::now()).to_string(),
|
|
})
|
|
}
|
|
|
|
fn build_tree_snapshot_payload(
|
|
root_uri: &str,
|
|
workspace_id: &str,
|
|
revision: &str,
|
|
kind: &str,
|
|
sidebar_snapshot: &ProjectionSnapshot,
|
|
file_tree_snapshot: &ProjectionSnapshot,
|
|
) -> Value {
|
|
let dataset = json!({
|
|
"kernel_sidebar_projection": sidebar_snapshot.projection,
|
|
"kernelSidebarProjection": sidebar_snapshot.projection,
|
|
"kernel_file_tree_projection": file_tree_snapshot.projection,
|
|
"kernelFileTreeProjection": file_tree_snapshot.projection,
|
|
});
|
|
|
|
json!({
|
|
"kind": kind,
|
|
"revision": revision,
|
|
"stream": "workspace",
|
|
"projection": "sidebar_tree",
|
|
"scope": "workspace",
|
|
"sourceKind": "local_folder",
|
|
"rootUri": root_uri,
|
|
"workspaceId": workspace_id,
|
|
"data": {
|
|
"dataset": dataset,
|
|
"tree": sidebar_snapshot.projection,
|
|
},
|
|
})
|
|
}
|
|
|
|
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 document_event_targets_relative_path(payload: &Value, expected: &str) -> bool {
|
|
let relative_path = payload
|
|
.get("relativePath")
|
|
.and_then(Value::as_str)
|
|
.unwrap_or_default();
|
|
expected == relative_path
|
|
}
|
|
|
|
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::*;
|
|
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;
|
|
use tower::util::ServiceExt;
|
|
|
|
fn test_root(name: &str) -> std::path::PathBuf {
|
|
let root = std::env::temp_dir().join(format!(
|
|
"mnote-local-folder-events-{name}-{}-{}",
|
|
std::process::id(),
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|duration| duration.as_nanos())
|
|
.unwrap_or(0)
|
|
));
|
|
let _ = std::fs::remove_dir_all(&root);
|
|
std::fs::create_dir_all(&root).expect("create temp root");
|
|
root
|
|
}
|
|
|
|
fn app_with_local_workspace(root: &std::path::Path) -> axum::Router {
|
|
let root_uri = format!("file://{}", root.display());
|
|
initialize_local_workspace_for_actor("dev-user", &root_uri).expect("init local workspace");
|
|
build_app(AppState::new(AppConfig {
|
|
service_name: "mnote-web".into(),
|
|
service_version: "0.1.0".into(),
|
|
bind_addr: "127.0.0.1:0".into(),
|
|
public_bind_addr: "127.0.0.1:3000".into(),
|
|
legacy_next_base_url: Some("http://127.0.0.1:3100".into()),
|
|
enable_legacy_next_compat: false,
|
|
enable_debug_shell_routes: false,
|
|
enable_editor_actor: true,
|
|
hermes_base_path: "/api/hermes".into(),
|
|
compat_next_base_path: "/api/compat/next".into(),
|
|
convex_url: None,
|
|
convex_admin_key: None,
|
|
allow_dev_fixtures: false,
|
|
query_fixtures_json: None,
|
|
mutation_fixtures_json: None,
|
|
dev_user_id: "dev-user".into(),
|
|
dev_user_name: "开发用户".into(),
|
|
dev_user_email: "dev@mnote.local".into(),
|
|
}))
|
|
}
|
|
|
|
/// Build a root_uri query-parameter-safe by percent-encoding (no external crate).
|
|
fn encoded_root_uri(raw: &str) -> String {
|
|
raw.replace('%', "%25")
|
|
.replace(':', "%3A")
|
|
.replace('/', "%2F")
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tree_live_headers_include_mnote_web_owner() {
|
|
let root = test_root("tree-live-headers");
|
|
std::fs::write(root.join("test.md"), "# Test\n").expect("write test");
|
|
let root_uri = format!("file://{}", root.display());
|
|
let encoded = encoded_root_uri(&root_uri);
|
|
|
|
let app = app_with_local_workspace(&root);
|
|
let response = app
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri(format!(
|
|
"/api/local-folder/events?rootUri={encoded}&treeLive=true"
|
|
))
|
|
.header("x-mnote-actor-id", "dev-user")
|
|
.header("x-mnote-actor-type", "user")
|
|
.body(Body::empty())
|
|
.expect("request"),
|
|
)
|
|
.await
|
|
.expect("response");
|
|
|
|
assert_eq!(response.status(), 200);
|
|
assert_eq!(
|
|
response
|
|
.headers()
|
|
.get("x-mnote-web-owner")
|
|
.and_then(|v| v.to_str().ok()),
|
|
Some("mnote-web"),
|
|
"response should have x-mnote-web-owner header"
|
|
);
|
|
assert_eq!(
|
|
response
|
|
.headers()
|
|
.get("x-mnote-local-folder-events-owner")
|
|
.and_then(|v| v.to_str().ok()),
|
|
Some("rust-web"),
|
|
"response should have x-mnote-local-folder-events-owner header"
|
|
);
|
|
|
|
let _ = std::fs::remove_dir_all(root);
|
|
}
|
|
|
|
#[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 document_event_filter_rejects_resource_only_changes() {
|
|
let payload = json!({
|
|
"sourceKind": "local_folder",
|
|
"rootUri": "file:///test",
|
|
"relativePath": "docs/README/image.png",
|
|
"documentId": "",
|
|
"eventKind": "Create(File)",
|
|
"revision": 1,
|
|
});
|
|
|
|
assert!(
|
|
!document_event_targets_relative_path(&payload, "docs/README.md"),
|
|
"正文事件流不能把资源文件变化当作 Markdown 正文变化"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn build_tree_snapshot_payload_has_required_fields() {
|
|
let sidebar_projection = json!({
|
|
"projection": "page_tree",
|
|
"sourceKind": "local_folder",
|
|
"rootUri": "file:///test",
|
|
"watchRevision": "abc123",
|
|
"items": [],
|
|
});
|
|
let file_tree_projection = json!({
|
|
"projection": "file_tree",
|
|
"sourceKind": "local_folder",
|
|
"rootUri": "file:///test",
|
|
"watchRevision": "abc123",
|
|
"items": [],
|
|
});
|
|
let sidebar = ProjectionSnapshot {
|
|
dataset: json!({}),
|
|
projection: sidebar_projection,
|
|
};
|
|
let file_tree = ProjectionSnapshot {
|
|
dataset: json!({}),
|
|
projection: file_tree_projection,
|
|
};
|
|
|
|
let payload = build_tree_snapshot_payload(
|
|
"file:///test",
|
|
"local:test_workspace",
|
|
"rev_1",
|
|
"snapshot",
|
|
&sidebar,
|
|
&file_tree,
|
|
);
|
|
|
|
assert_eq!(payload["kind"], "snapshot");
|
|
assert_eq!(payload["stream"], "workspace");
|
|
assert_eq!(payload["projection"], "sidebar_tree");
|
|
assert_eq!(payload["sourceKind"], "local_folder");
|
|
assert_eq!(payload["rootUri"], "file:///test");
|
|
assert_eq!(payload["workspaceId"], "local:test_workspace");
|
|
assert_eq!(payload["revision"], "rev_1");
|
|
assert!(payload["data"]["dataset"]["kernel_sidebar_projection"].is_object());
|
|
assert!(payload["data"]["dataset"]["kernel_file_tree_projection"].is_object());
|
|
assert!(payload["data"]["dataset"]["kernelSidebarProjection"].is_object());
|
|
assert!(payload["data"]["dataset"]["kernelFileTreeProjection"].is_object());
|
|
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");
|
|
std::fs::create_dir_all(root.join("docs")).expect("create docs");
|
|
std::fs::write(root.join("docs/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 payload = build_local_folder_watch_batch_payload(
|
|
&root_uri,
|
|
&workspace_id,
|
|
vec![
|
|
json!({
|
|
"relativePath": "docs/README.md",
|
|
"eventKind": "Modify(Data)",
|
|
}),
|
|
json!({
|
|
"relativePath": "docs/New.md",
|
|
"eventKind": "Create(File)",
|
|
}),
|
|
],
|
|
)
|
|
.expect("watch batch payload");
|
|
|
|
assert_eq!(payload["schema"], "mnote.local_folder_watch_batch.v1");
|
|
assert_eq!(payload["kind"], "watch_batch");
|
|
assert_eq!(payload["fallbackResync"], false);
|
|
assert_eq!(payload["changedPaths"].as_array().map(Vec::len), Some(2));
|
|
assert!(
|
|
payload["affectedParents"]
|
|
.as_array()
|
|
.expect("affected parents")
|
|
.iter()
|
|
.any(|parent| parent["relativePath"].as_str() == Some("docs")
|
|
&& 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)")));
|
|
|
|
let _ = std::fs::remove_dir_all(root);
|
|
}
|
|
|
|
#[test]
|
|
fn tree_live_error_payload_is_structured() {
|
|
let payload = build_tree_live_error_payload(
|
|
"file:///test",
|
|
"local:test",
|
|
"tree_live_resync_failed",
|
|
"failed",
|
|
);
|
|
|
|
assert_eq!(payload["schema"], "mnote.tree_live_error.v1");
|
|
assert_eq!(payload["phase"], "tree_live_resync");
|
|
assert_eq!(payload["fallbackResync"], true);
|
|
assert_eq!(payload["code"], "tree_live_resync_failed");
|
|
}
|
|
}
|