Skip to content

feat: index-backed /search (jieba/CJK, ranked, visibility-filtered) #239

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sufforest merged 1 commit into from
Jul 4, 2026
Merged

feat: index-backed /search (jieba/CJK, ranked, visibility-filtered) #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
238 changes: 135 additions & 103 deletions vela-api/src/directory/search.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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()));
}

Expand All @@ -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")
Expand All @@ -81,6 +85,12 @@ pub async fn post_search(
.filter_map(|v| v.as_str().map(String::from))
.collect()
});
let filter_senders: Option<HashSet<String>> =
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()
Expand Down Expand Up @@ -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<Hit> = 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<u64, Cand> = 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")
Expand All @@ -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();

Expand All @@ -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<String> = HashSet::new();
for t in term.split_whitespace() {
term_tokens.insert(t.to_string());
}

let mut results: Vec<Value> = 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,
Expand All @@ -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::<Vec<_>>()),
);
// 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
Expand Down Expand Up @@ -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,
Expand All @@ -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);
}
Expand All @@ -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);
}
Expand All @@ -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<String> {
let mut out: HashSet<String> = 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::<Value>(&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.
Expand Down
Loading
Loading