From 4539a90069d2b4dc9545d839abec525abaf38a50 Mon Sep 17 00:00:00 2001 From: sufforest Date: Sat, 4 Jul 2026 06:25:32 -0700 Subject: [PATCH] feat: index-backed /search (jieba/CJK, ranked, visibility-filtered) rewrites POST /search to match against the inverted index instead of a substring scan over the last 1000 timeline events. the query is jieba- tokenized the same way the index is, so CJK searches work word-by-word; multi-term queries AND-intersect postings; order_by rank sums term frequency; the keys and filter.senders filters are honored. every hit AND its before/after context events pass the per-event history-visibility gate (the substring version gated neither, so context could leak pre-join / post-leave events); redacted and E2EE-room events are excluded. --- vela-api/src/directory/search.rs | 238 +++++++++++++++------------ vela-api/tests/search_index_query.rs | 198 ++++++++++++++++++++++ 2 files changed, 333 insertions(+), 103 deletions(-) create mode 100644 vela-api/tests/search_index_query.rs diff --git a/vela-api/src/directory/search.rs b/vela-api/src/directory/search.rs index 3d8f9af..dea37df 100644 --- a/vela-api/src/directory/search.rs +++ b/vela-api/src/directory/search.rs @@ -1,35 +1,37 @@ -//! `POST /_matrix/client/v3/search` — event search across joined rooms. +//! `POST /_matrix/client/v3/search` — event search across the caller's rooms. //! //! Spec: `client-server-api/#post_matrixclientv3search`. //! -//! MVP: linear case-insensitive substring scan over the timelines of -//! each of the caller's joined rooms. No inverted index — `SCAN_PER_ROOM` -//! bounds the per-room work. Supports `filter.rooms`, `filter.limit`, -//! `order_by` (`rank` or `recent`), `event_context` (before/after), -//! `next_batch` pagination, redaction filtering, and predecessor-chain -//! search. +//! Index-backed: matches against the `search_index` inverted index (see +//! `vela_store::search`), which is jieba-tokenized so Chinese/CJK text is +//! searched word-by-word. The query is tokenized the same way, then postings +//! are AND-intersected per room. Supports `keys`, `filter.rooms`, +//! `filter.senders`, `filter.limit`, `order_by` (`rank` by term frequency, or +//! `recent`), `event_context` (before/after), `next_batch` pagination, and +//! predecessor-chain search. Redacted events are filtered, and every hit is +//! checked against per-event history-visibility. //! -//! E2EE rooms are skipped: we hold only ciphertext for them, so the -//! server can't usefully match against the encrypted body. Clients are -//! expected to do their own local index for E2EE rooms. +//! E2EE rooms are skipped: we hold only ciphertext for them, so the server +//! can't usefully match. Clients index E2EE rooms locally. use crate::middleware::json::Json; use axum::extract::{Query, State}; use serde::Deserialize; use serde_json::{Value, json}; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use vela_core::error::VelaError; +use vela_store::search::{FIELD_BODY, FIELD_NAME, FIELD_TOPIC}; use crate::middleware::auth::AuthenticatedUser; use crate::middleware::error::ApiError; -use crate::room::messages::load_client_event; +use crate::room::messages::{TimelineReadGate, load_client_event}; use crate::router::AppState; -/// Per-room events we scan before giving up. Large enough that recent -/// chat history is covered; small enough that we don't burn cycles on -/// pathological rooms. -const SCAN_PER_ROOM: usize = 1000; +/// Per-token, per-room postings we consider. Bounds memory and pagination +/// depth: for a very common token in a huge room we look at the newest +/// `CANDIDATE_CAP` occurrences. Rare tokens (the usual case) are unaffected. +const CANDIDATE_CAP: usize = 1000; /// Default `filter.limit` when the caller doesn't supply one. const DEFAULT_LIMIT: usize = 10; /// Hard ceiling so a malicious `limit` can't make us serialize 10⁶ events. @@ -52,14 +54,14 @@ pub async fn post_search( .cloned() .unwrap_or_else(|| json!({})); - let search_term = room_events + // Tokenize the query exactly as the index did (jieba, CJK-aware). An + // empty tokenization (blank / punctuation-only term) yields no results. + let raw_term = room_events .get("search_term") .and_then(|v| v.as_str()) - .map(|s| s.to_lowercase()); - let Some(term) = search_term else { - return Ok(Json(empty_response())); - }; - if term.is_empty() { + .unwrap_or(""); + let q_tokens = vela_store::search::query_tokens(raw_term); + if q_tokens.is_empty() { return Ok(Json(empty_response())); } @@ -68,6 +70,8 @@ pub async fn post_search( .and_then(|v| v.as_str()) .unwrap_or("rank") .to_string(); + // Which of the searchable keys to match (spec `keys`; defaults to all). + let keys_mask = parse_keys_mask(&room_events); let filter = room_events.get("filter").cloned().unwrap_or(json!({})); let limit: usize = filter .get("limit") @@ -81,6 +85,12 @@ pub async fn post_search( .filter_map(|v| v.as_str().map(String::from)) .collect() }); + let filter_senders: Option> = + filter.get("senders").and_then(|v| v.as_array()).map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }); let context = room_events.get("event_context").cloned(); let context_before = context .as_ref() @@ -124,56 +134,72 @@ pub async fn post_search( let _ = collect_room_and_predecessors(&state, &room_id, &mut rooms_to_scan, &mut visited); } - // Collect every match across all rooms (we need the full count - // anyway for the response and for pagination). For each hit - // remember its stream_pos in its room so we can stitch - // before/after context. + // Collect every visible match across all rooms (we need the full count + // for the response and pagination). Per room, pull each query token's + // postings from the index and AND-intersect by event: an event matches + // iff every token appears in it (in an allowed key). `rank` is the summed + // term frequency. struct Hit { room_nid: u64, room_id: String, event: Value, stream_pos: u64, origin_server_ts: u64, + rank: u32, + } + struct Cand { + stream_pos: u64, + tf: u32, + matched: usize, } let mut all_hits: Vec = Vec::new(); for (room_nid, room_id) in rooms_to_scan { if is_room_encrypted(&state, room_nid) { continue; } - let entries = state - .db - .get_timeline_latest(room_nid, SCAN_PER_ROOM) - .map_err(|e| ApiError(VelaError::Store(e.to_string())))?; - let redacted = redacted_event_ids(&state, &entries); - for (pos, enid) in entries.iter().rev() { - let Some(ev) = load_client_event(&state, *enid, &room_id)? else { + // Per-event visibility gate for this caller (leave-cap + history + // visibility). Reflects membership; never 403s here. + let gate = TimelineReadGate::resolve_reader(&state, room_nid, user.user_nid)?; + + // A searchable event carries exactly one field, so a token appears at + // most once per event → one posting per (token, event). `matched` + // therefore counts distinct query tokens present. + let mut cand: HashMap = HashMap::new(); + for tok in &q_tokens { + for p in state.db.search_room_token(room_nid, tok, CANDIDATE_CAP) { + if (keys_mask & (1u8 << p.field)) == 0 { + continue; + } + let c = cand.entry(p.event_nid).or_insert(Cand { + stream_pos: p.stream_pos, + tf: 0, + matched: 0, + }); + c.tf += u32::from(p.tf); + c.matched += 1; + } + } + + for (enid, c) in cand { + if c.matched != q_tokens.len() { + continue; // AND: every query token must be present + } + // Redacted events must never surface. + if matches!(state.db.get_redacted_by(enid), Ok(Some(_))) { continue; - }; - let event_id = ev - .get("event_id") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - if redacted.contains(&event_id) { + } + // History-visibility + leave-cap for this specific event. + if !gate.permits(&state, room_nid, user.user_nid, enid, Some(c.stream_pos))? { continue; } - let body = ev - .pointer("/content/body") - .and_then(|v| v.as_str()) - .unwrap_or(""); - let body_lc = body.to_lowercase(); - // Token AND match: every whitespace-separated token in the - // search term must appear as a substring of the body. - // Synapse uses Postgres FTS (tsvector) which is closer to - // word-boundary matching; substring is a simpler MVP that - // still handles `"Message 4"` against `"Message number 4"` - // (token "4" is present even though "Message 4" verbatim - // isn't a substring). - let all_present = term - .split_whitespace() - .all(|t| !t.is_empty() && body_lc.contains(t)); - if !all_present { + let Some(ev) = load_client_event(&state, enid, &room_id)? else { continue; + }; + if let Some(senders) = &filter_senders { + let sender = ev.get("sender").and_then(|v| v.as_str()).unwrap_or(""); + if !senders.contains(sender) { + continue; + } } let ts = ev .get("origin_server_ts") @@ -183,21 +209,24 @@ pub async fn post_search( room_nid, room_id: room_id.clone(), event: ev, - stream_pos: *pos, + stream_pos: c.stream_pos, origin_server_ts: ts, + rank: c.tf, }); } } - // Order: spec defines `rank` (relevance-weighted; we have no - // relevance signal, so fall back to recent) and `recent`. Both - // sort newest first. - all_hits.sort_by(|a, b| { - b.origin_server_ts - .cmp(&a.origin_server_ts) - .then(b.stream_pos.cmp(&a.stream_pos)) - }); - let _ = order_by; // future: actually weight by token frequency for "rank". + // Order: `recent` = newest first; `rank` (default) = most term matches + // first, ties broken by recency. + if order_by == "recent" { + all_hits.sort_by(|a, b| { + b.origin_server_ts + .cmp(&a.origin_server_ts) + .then(b.stream_pos.cmp(&a.stream_pos)) + }); + } else { + all_hits.sort_by(|a, b| b.rank.cmp(&a.rank).then(b.stream_pos.cmp(&a.stream_pos))); + } let count = all_hits.len(); @@ -211,20 +240,23 @@ pub async fn post_search( let page_end = (page_start + limit).min(count); let page = &all_hits[page_start..page_end]; - let mut term_tokens: HashSet = HashSet::new(); - for t in term.split_whitespace() { - term_tokens.insert(t.to_string()); - } - let mut results: Vec = Vec::with_capacity(page.len()); for hit in page { let mut entry = json!({ - "rank": 1.0, + "rank": f64::from(hit.rank), "result": hit.event.clone(), }); if context.is_some() { + // The flanking context events must pass the SAME per-event + // visibility gate as the hit — otherwise a `joined`-visibility + // room leaks pre-join events (or a departed member leaks + // post-leave events) as "context". Re-resolve the gate for this + // hit's room (page is bounded by `limit`). + let gate = TimelineReadGate::resolve_reader(&state, hit.room_nid, user.user_nid)?; let ctx = build_event_context( &state, + &gate, + user.user_nid, hit.room_nid, &hit.room_id, hit.stream_pos, @@ -242,10 +274,9 @@ pub async fn post_search( let mut room_events_resp = serde_json::Map::new(); room_events_resp.insert("count".to_string(), json!(count)); room_events_resp.insert("results".to_string(), Value::Array(results)); - room_events_resp.insert( - "highlights".to_string(), - json!(term_tokens.into_iter().collect::>()), - ); + // Highlights: the query tokens clients should emphasize — the actual + // (jieba-segmented) tokens we matched on, so CJK words highlight too. + room_events_resp.insert("highlights".to_string(), json!(q_tokens)); // Spec-loose semantic: include `next_batch` whenever we returned // any results, so the client paginates one more call to confirm // "no more". Drop it on the first empty page — matches the @@ -326,8 +357,11 @@ fn collect_room_and_predecessors( /// Build the spec's `context` block for a hit: timeline events /// immediately before and after the match (no further filtering). The /// page is small (test caps at 2 each side) so a linear walk is fine. +#[allow(clippy::too_many_arguments)] fn build_event_context( state: &AppState, + gate: &TimelineReadGate, + user_nid: u64, room_nid: u64, room_id: &str, pivot_pos: u64, @@ -342,8 +376,12 @@ fn build_event_context( .get_timeline_before(room_nid, pivot_pos, before) .map_err(|e| ApiError(VelaError::Store(e.to_string())))?; // get_timeline_before returns chronological order; we want - // newest-first per spec. - for (_pos, enid) in entries.iter().rev() { + // newest-first per spec. Each event is visibility-gated so context + // can't leak events the caller may not see. + for (pos, enid) in entries.iter().rev() { + if !gate.permits(state, room_nid, user_nid, *enid, Some(*pos))? { + continue; + } if let Some(ev) = load_client_event(state, *enid, room_id)? { before_events.push(ev); } @@ -354,7 +392,10 @@ fn build_event_context( .db .get_timeline_range(room_nid, pivot_pos + 1, u64::MAX, after) .map_err(|e| ApiError(VelaError::Store(e.to_string())))?; - for (_pos, enid) in entries.iter() { + for (pos, enid) in entries.iter() { + if !gate.permits(state, room_nid, user_nid, *enid, Some(*pos))? { + continue; + } if let Some(ev) = load_client_event(state, *enid, room_id)? { after_events.push(ev); } @@ -368,33 +409,24 @@ fn build_event_context( })) } -/// Collect the set of event_ids that have been redacted in the given -/// timeline window. We look at `m.room.redaction` events and capture -/// their `redacts` field (or `content.redacts` for newer rooms). -fn redacted_event_ids(state: &AppState, entries: &[(u64, u64)]) -> HashSet { - let mut out: HashSet = HashSet::new(); - let Ok(Some(redaction_type_nid)) = state.db.get_nid("m.room.redaction") else { - return out; +/// Parse the spec `keys` array into a field bitmask (one bit per `FIELD_*`). +/// An absent, empty, or all-unrecognized `keys` means "all keys" — the spec +/// default. A hit is kept only if it matched in one of these keys. +fn parse_keys_mask(room_events: &Value) -> u8 { + let all = (1u8 << FIELD_BODY) | (1u8 << FIELD_NAME) | (1u8 << FIELD_TOPIC); + let Some(arr) = room_events.get("keys").and_then(|v| v.as_array()) else { + return all; }; - for (_pos, enid) in entries { - let Ok(Some((header, bytes))) = state.db.get_event(*enid) else { - continue; - }; - if header.type_nid != redaction_type_nid { - continue; - } - let Ok(json) = serde_json::from_slice::(&bytes) else { - continue; - }; - if let Some(target) = json - .pointer("/content/redacts") - .and_then(|v| v.as_str()) - .or_else(|| json.get("redacts").and_then(|v| v.as_str())) - { - out.insert(target.to_string()); + let mut mask = 0u8; + for k in arr { + match k.as_str() { + Some("content.body") => mask |= 1u8 << FIELD_BODY, + Some("content.name") => mask |= 1u8 << FIELD_NAME, + Some("content.topic") => mask |= 1u8 << FIELD_TOPIC, + _ => {} } } - out + if mask == 0 { all } else { mask } } /// True iff the room currently has an `m.room.encryption` state event. diff --git a/vela-api/tests/search_index_query.rs b/vela-api/tests/search_index_query.rs new file mode 100644 index 0000000..471b9d7 --- /dev/null +++ b/vela-api/tests/search_index_query.rs @@ -0,0 +1,198 @@ +//! `POST /v3/search` is index-backed: it matches against the jieba-tokenized +//! inverted index, so CJK text is searchable word-by-word, multi-term queries +//! are AND-intersected, and the `keys` filter is honored. + +mod common; + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use common::{Harness, read_json}; +use serde_json::{Value, json}; + +async fn search(harness: &Harness, token: &str, room_events: Value) -> Value { + let resp = harness + .request( + Request::post("/_matrix/client/v3/search") + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from( + json!({ "search_categories": { "room_events": room_events } }).to_string(), + )) + .unwrap(), + ) + .await; + assert_eq!(resp.status(), StatusCode::OK, "search returned non-200"); + read_json(resp).await +} + +fn count(resp: &Value) -> u64 { + resp["search_categories"]["room_events"]["count"] + .as_u64() + .unwrap() +} + +#[tokio::test] +async fn search_by_body_cjk_keys_and_and_semantics() { + let harness = Harness::new(); + harness.state.db.set_search_indexing_enabled(true); + let (_uid, token) = harness.register("alice", "password").await; + let room = harness + .create_room(&token, json!({"preset": "private_chat"})) + .await; + + let eid = harness + .send_message(&token, &room, "hello, 世界 world") + .await; + harness + .send_message(&token, &room, "unrelated chatter") + .await; + + let rooms = json!({ "rooms": [room] }); + + // Body search finds the exact message. + let r = search( + &harness, + &token, + json!({"search_term": "hello", "keys": ["content.body"], "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 1, "one hit for 'hello'"); + assert_eq!( + r["search_categories"]["room_events"]["results"][0]["result"]["event_id"], + json!(eid), + ); + + // CJK: 世界 was segmented into a token and is findable — the whole point. + let r = search( + &harness, + &token, + json!({"search_term": "世界", "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 1, "CJK word 世界 must match"); + + // Multi-term AND: both words present in the same event → match. + let r = search( + &harness, + &token, + json!({"search_term": "hello world", "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 1, "'hello world' both present → match"); + + // AND: a term where one token is absent → no match. + let r = search( + &harness, + &token, + json!({"search_term": "hello goodbye", "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 0, "'goodbye' absent → no match"); + + // An absent word → no hits. + let r = search( + &harness, + &token, + json!({"search_term": "goodbye", "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 0); + + // `keys` filter: the word lives in a body, not a room name, so + // restricting to content.name yields nothing. + let r = search( + &harness, + &token, + json!({"search_term": "hello", "keys": ["content.name"], "filter": rooms}), + ) + .await; + assert_eq!(count(&r), 0, "keys=content.name must exclude a body match"); +} + +/// A hit's `context` events must pass the same history-visibility gate as +/// the hit. Under `joined` visibility, a message sent before the searcher +/// joined must NOT leak into `context.events_before`. +#[tokio::test] +async fn context_events_are_visibility_gated() { + let harness = Harness::new(); + harness.state.db.set_search_indexing_enabled(true); + let (_a, alice) = harness.register("alice", "password").await; + let (_b, bob) = harness.register("bob", "password").await; + + let room = harness + .create_room( + &alice, + json!({ + "preset": "public_chat", + "initial_state": [{ + "type": "m.room.history_visibility", + "state_key": "", + "content": {"history_visibility": "joined"} + }] + }), + ) + .await; + + // Alice posts before Bob joins — Bob must never see this, even as context. + harness + .send_message(&alice, &room, "secretbeforejoin apple") + .await; + + harness.join(&bob, &room).await; + let hit = harness.send_message(&bob, &room, "findme apple").await; + + let r = search( + &harness, + &bob, + json!({ + "search_term": "findme", + "filter": {"rooms": [room]}, + "event_context": {"before_limit": 10, "after_limit": 0} + }), + ) + .await; + assert_eq!(count(&r), 1); + let res0 = &r["search_categories"]["room_events"]["results"][0]; + assert_eq!(res0["result"]["event_id"], json!(hit)); + // The pre-join message must be absent from the context. + let before = res0["context"]["events_before"] + .as_array() + .cloned() + .unwrap_or_default(); + for ev in &before { + let body = ev["content"]["body"].as_str().unwrap_or(""); + assert!( + !body.contains("secretbeforejoin"), + "pre-join event leaked into search context: {ev}" + ); + } +} + +#[tokio::test] +async fn rank_orders_by_term_frequency() { + let harness = Harness::new(); + harness.state.db.set_search_indexing_enabled(true); + let (_uid, token) = harness.register("bob", "password").await; + let room = harness + .create_room(&token, json!({"preset": "private_chat"})) + .await; + + harness.send_message(&token, &room, "apple once").await; + let dense = harness + .send_message(&token, &room, "apple apple apple") + .await; + + let r = search( + &harness, + &token, + json!({"search_term": "apple", "order_by": "rank", "filter": {"rooms": [room]}}), + ) + .await; + assert_eq!(count(&r), 2); + // Higher term frequency ranks first. + assert_eq!( + r["search_categories"]["room_events"]["results"][0]["result"]["event_id"], + json!(dense), + "the message with more occurrences should rank first" + ); +}