feat(ui): global search service

[stefanceriu]

Introduce a reactive GlobalSearch service that will act as an entry point for results aggregated from multiple sources.

The only source for now is the SDK's per-room message search, merged across the filtered rooms by relevance score, with each hit's content and sender resolved via the timeline machinery. The design leaves room for further SearchResult kinds (people, rooms etc.) and for swapping in server-side sources without changing the interface.

[codecov]

Codecov Report

❌ Patch coverage is 89.40397% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 89.94%. Comparing base (97380ac) to head (e540702).
⚠️ Report is 13 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
crates/matrix-sdk-ui/src/search_service.rs	89.40%	13 Missing and 3 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #6695      +/-   ##
==========================================
+ Coverage   89.91%   89.94%   +0.03%     
==========================================
  Files         396      397       +1     
  Lines      110609   110760     +151     
  Branches   110609   110760     +151     
==========================================
+ Hits        99454    99624     +170     
+ Misses       7378     7369       -9     
+ Partials     3777     3767      -10     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[codspeed-hq]

Merging this PR will improve performance by 35.18%

⚡ 1 improved benchmark
✅ 49 untouched benchmarks

Performance Changes

	Mode	Benchmark	BASE	HEAD	Efficiency
⚡	Simulation	Restore session [memory store]	226.9 ms	167.9 ms	+35.18%

Tip

Curious why this is faster? Comment @codspeedbot explain why this is faster on this PR, or directly use the CodSpeed MCP with your agent.

---

_{Comparing stefan/globalSearchService (e540702) with main (805f981)}

[Hywan]

The PR description and commits mention GlobalSearch while the code has a SearchService. Who's the lier?

[Hywan]

I think this design can be improved: we should not involve two locks here.

Ideally, you want:

• SearchService::new(client: Client) so you get the SearchService and you can run queries with it,
• SearchService::query(&self, query: String) -> ObservableQuery

And then, on ObservableQuery, you can have the following methods:

• ObservableQuery::subscribe(&self) -> QueryResultSubscriber to subscribe to results, where QueryResultSubscriber is a thin type around eyeball_im::VectorSubscriber<QueryResult> exposing the following methods:
  • QueryResultSubscriber::values(&self) -> Vector<QueryResult> to get all current values,
  • QueryResultSubscriber::into_values_and_stream(self) -> (Vector<QueryResult>, eyeball_im::VectorSubscriberBatchedStream<QueryResult>), which is the equivalent of eyeball_im::VectorSubscriber::into_values_and_batched_stream.
• ObservableQuery::paginate(&self) to paginate the query further more
• ObservableQuery::pagination_state(&self) -> QueryPaginationState to get the pagination status of this query.

Now we have a design, how would you implement this? I initially wrote many lines of code but I ultimately got stuck on the same problem as you are I believe: paginating requires calling Stream::next(), which implies you have to store the stream somewhere.

Maybe we can loop over the stream in a task, and have a channel between the task and ObservableQuery. When calling ObservableQuery::paginate(), a message is sent over the channel, and the loop over the stream does a new iteration, such as:

fn task(
    client: Client,
    query_results: Arc<Mutex<ObservableVector<QueryResult>>>,
    pagination: Arc<Notify>,
    pagination_state: SharedObservable<QueryPaginationState>
) {
    let stream = client.search_messages(query).build_events();

    while let Some(query_results) = stream.next() {
        // add `query_results` on the shared observable vector
        pagination.notified().await;
    }
}

pub struct ObservableQuery {
    task: AbortOnDrop<()>,
    pagination: Arc<Notify>,
    pagination_state: SharedObservable<QueryPaginationState>,
    query_results: Arc<Mutex<ObservableVector<QueryResult>>>,
}

impl ObservableQuery {
    fn new(client: Client, query: String) -> Self {
        let pagination = Arc::new(Notify::new());
        let pagination_state = SharedObservable::new(QueryPaginationState::default());
        let query_results = Arc::new(Mutex::new(ObservableVector::new()));
        let task = task(client, query_results.clone(), pagination.clone());

        Self { … }
    }

    pub async fn paginate(&self) -> bool {
        match self.pagination_state.read().await  {
            PaginationState::Idle { end_reached } if end_reached.not()) => {
                self.pagination.notify_one();
                true
            }
            _ => false,
        }
    }
}

That way, you don't have to store the stream. You only have a single lock around the shared ObservableVector of QueryResults. You also no longer have an Option around the stream! And finally, the Client is passed directly to and owned by the task.

Also, notify_one (called by paginate) adds one permit every time it is called, ensuring the task will call next() exactly $1 + n$ where $n$ is the number of times paginate has been called. You already have this guarantee now.
Note that we can check the pagination state in paginate to avoid calling notify_one.

[stefanceriu]

We had a chat about this yesterday and agreed to start with a simplified version in which the streams are driven from background tasks and the task moves the pagination forward by listening to a Notify object.

I've done so in f0f9f27 but after seeing it in action I'm not so sure it made anything better:

• we still have 2 mutexes as the task itself needs one. results also gained an Arc
• paginate is now fire and forget and errors need to be surfaced from the listener. They also stop the stream so the recovery path now involves setting a new query, which isn't particularly ergonomic
• and generally there's just more things to think and reason about now for honestly not clear gains

Did I perhaps completely misunderstood your vision here? 🤔

[Hywan]

A couple of feedback.

To address the fact the pagination can return an error, we could do something a bit different: instead of using Notify, we can use a channel where the pagination sends ControlFlow::Continue(()) and then immediately waits to receive a ControlFlow too. If it receives a ControlFlow::Continue(()), then no error; otherwise if it receives a ControlFlow::Break(error), then it has to return the error. It will require 2 channels for that 🤔.

Edit: note that the design you adopted looks fine to me.

[Hywan]

I won't lie: this is NOT your fault, but I'm not super happy with the final result: more refactoring of the lower layers would be required to make it better. That's probably okay-ish with the current APIs though. It's still experimental and we can break it whenever we want. Let's iterate based on that!

Thanks for tackling this and for having tried a different approach!