wsb 0.155.49

Workspace - A tool suite for file operations, version management, and development workflow automation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

use axum::{
    extract::{Query, State},
    response::sse::{Event, KeepAlive, Sse},
};
use serde::Deserialize;
use std::convert::Infallible;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use crate::web::search_sources::{CancelOnDrop, FileSource, SearchEvent, SessionSource};
use crate::web::state::WebState;

#[derive(Deserialize)]
pub struct SearchQuery {
    pub q: String,
    pub sort: Option<String>,
    pub project: Option<String>,
}

pub async fn global_search(
    State(state): State<WebState>,
    Query(params): Query<SearchQuery>,
) -> Sse<impl futures::Stream<Item = Result<Event, Infallible>>> {
    let cancelled = Arc::new(AtomicBool::new(false));
    let cancel_guard = CancelOnDrop(cancelled.clone());

    let (tx, mut rx) = tokio::sync::mpsc::channel::<SearchEvent>(256);

    // SessionSource is async — run as a regular tokio task.
    {
        let tx = tx.clone();
        let q = params.q.clone();
        let c = cancelled.clone();
        let source = SessionSource::new(state.db.clone(), params.sort.clone(), params.project.clone());
        tokio::spawn(async move { source.run(q, tx, c).await });
    }

    // FileSource is blocking (disk walk) — run in the blocking thread pool.
    {
        let tx = tx.clone();
        let q = params.q.clone();
        let c = cancelled.clone();
        let source = FileSource::new(state.workspace_root.clone());
        tokio::task::spawn_blocking(move || source.run(q, tx, c));
    }

    // Drop the original sender so rx closes when both sources finish.
    drop(tx);

    let stream = async_stream::stream! {
        let _guard = cancel_guard; // dropped when stream is dropped → sets cancelled
        while let Some(event) = rx.recv().await {
            match serde_json::to_string(&event) {
                Ok(json) => yield Ok(Event::default().event("result").data(json)),
                Err(e) => yield Ok(Event::default().event("error").data(e.to_string())),
            }
        }
        yield Ok(Event::default().event("end").data("{}"));
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}