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

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::{SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast::error::RecvError;

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() {
                            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,
                }
            }
        },
    )
    .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 subscription = state
        .local_folder_watcher_registry()
        .subscribe(&canonical_root)
        .map_err(|error| WebError::internal(error).with_context(&context))?;

    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 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) => {
                        // Rebuild full snapshot on any filesystem change
                        if let Some(resync_payload) =
                            rebuild_tree_resync_payload(&root_uri, &workspace_id)
                        {
                            return Some((
                                Ok(stream_event("resync", &resync_payload)),
                                (None, subscription, root_uri, workspace_id),
                            ));
                        }
                        // Snapshot load failed — continue waiting for next change
                        continue;
                    }
                    Err(RecvError::Lagged(_)) => continue,
                    Err(RecvError::Closed) => return None,
                }
            }
        },
    )
    .boxed();

    Ok((HeaderMap::new(), stream))
}

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 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 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 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());
    }
}