refactor(server): indexer & worker & sync perf (#15504)

packages/frontend/native/index.d.ts

@@ -154,6 +154,7 @@ export declare class DocStoragePool {
   getDocClock(universalId: string, docId: string): Promise<DocClock | null>
   getDocIndexedClock(universalId: string, docId: string): Promise<DocIndexedClock | null>
   setDocIndexedClock(universalId: string, docId: string, indexedClock: Date, indexerVersion: number): Promise<void>
+  setDocIndexedClocks(universalId: string, clocks: Array<DocIndexedClock>): Promise<void>
   clearDocIndexedClock(universalId: string, docId: string): Promise<void>
   getBlob(universalId: string, key: string): Promise<Blob | null>
   setBlob(universalId: string, blob: SetBlob): Promise<void>
@@ -172,13 +173,13 @@ export declare class DocStoragePool {
   clearClocks(universalId: string): Promise<void>
   setBlobUploadedAt(universalId: string, peer: string, blobId: string, uploadedAt?: Date | undefined | null): Promise<void>
   getBlobUploadedAt(universalId: string, peer: string, blobId: string): Promise<Date | null>
-  ftsAddDocument(id: string, indexName: string, docId: string, text: string, index: boolean): Promise<void>
-  ftsFlushIndex(id: string): Promise<void>
-  ftsIndexVersion(): Promise<number>
-  ftsDeleteDocument(id: string, indexName: string, docId: string): Promise<void>
-  ftsGetDocument(id: string, indexName: string, docId: string): Promise<string | null>
-  ftsSearch(id: string, indexName: string, query: string): Promise<Array<NativeSearchHit>>
-  ftsGetMatches(id: string, indexName: string, docId: string, query: string): Promise<Array<NativeMatch>>
+  indexUpsert(id: string, table: string, document: NativeIndexDocument): Promise<void>
+  indexFlush(id: string): Promise<void>
+  indexVersion(): Promise<number>
+  indexDelete(id: string, table: string, docId: string): Promise<void>
+  indexSearch(id: string, table: string, query: NativeIndexQuery, options: NativeIndexSearchOptions): Promise<NativeIndexSearchResult>
+  indexAggregate(id: string, table: string, query: NativeIndexQuery, field: string, limit: number, offset: number, hits?: NativeIndexSearchOptions | undefined | null): Promise<NativeIndexAggregateResult>
+  indexDeleteByQuery(id: string, table: string, query: NativeIndexQuery): Promise<number>
 }
 
 export interface Blob {
@@ -237,15 +238,69 @@ export interface NativeCrawlResult {
   summary: string
 }
 
-export interface NativeMatch {
-  start: number
-  end: number
+export interface NativeIndexAggregateResult {
+  total: number
+  buckets: Array<NativeIndexBucket>
 }
 
-export interface NativeSearchHit {
+export interface NativeIndexBucket {
+  key: string
+  count: number
+  score: number
+  hits: Array<NativeIndexHit>
+}
+
+export interface NativeIndexDocument {
+  id: string
+  fields: Array<NativeIndexField>
+}
+
+export interface NativeIndexField {
+  field: string
+  values: Array<string>
+}
+
+export interface NativeIndexHighlight {
+  field: string
+  values: Array<NativeIndexHighlightValue>
+}
+
+export interface NativeIndexHighlightValue {
+  valueIndex: number
+  spans: Array<NativeIndexSpan>
+}
+
+export interface NativeIndexHit {
   id: string
   score: number
-  terms: Array<string>
+  fields: Array<NativeIndexField>
+  highlights: Array<NativeIndexHighlight>
+}
+
+export interface NativeIndexQuery {
+  kind: string
+  field?: string
+  value?: string
+  occur?: string
+  clauses?: Array<NativeIndexQuery>
+  boost?: number
+}
+
+export interface NativeIndexSearchOptions {
+  limit: number
+  offset: number
+  fields: Array<string>
+  highlights: Array<string>
+}
+
+export interface NativeIndexSearchResult {
+  total: number
+  hits: Array<NativeIndexHit>
+}
+
+export interface NativeIndexSpan {
+  start: number
+  end: number
 }
 
 export interface SetBlob {

packages/frontend/native/nbstore/src/error.rs

@@ -12,8 +12,12 @@ pub enum Error {
   ConnectionInProgress,
   #[error("Invalid operation")]
   InvalidOperation,
+  #[error("Index is rebuilding")]
+  IndexNotReady,
   #[error("Serialization Error: {0}")]
   Serialization(String),
   #[error(transparent)]
+  Indexer(#[from] memory_indexer::Error),
+  #[error(transparent)]
   Parse(#[from] ParseError),
 }

packages/frontend/native/nbstore/src/indexer.rs

@@ -1,353 +0,0 @@
-use affine_doc_loader::{BlockInfo, CrawlResult, ParseError, parse_doc_from_binary};
-use memory_indexer::{SearchHit, SnapshotData};
-use napi_derive::napi;
-use serde::Serialize;
-use sqlx::Row;
-use y_octo::merge_updates_v1;
-
-// Increment this whenever there is a breaking change in the index format or how
-// updates are applied
-const NBSTORE_INDEXER_VERSION: u32 = 1;
-
-use super::{
-  error::{Error, Result},
-  storage::SqliteDocStorage,
-};
-
-#[napi(object)]
-#[derive(Debug, Serialize)]
-pub struct NativeBlockInfo {
-  pub block_id: String,
-  pub flavour: String,
-  pub content: Option<Vec<String>>,
-  pub blob: Option<Vec<String>>,
-  pub ref_doc_id: Option<Vec<String>>,
-  pub ref_info: Option<Vec<String>>,
-  pub parent_flavour: Option<String>,
-  pub parent_block_id: Option<String>,
-  pub additional: Option<String>,
-}
-
-impl From<BlockInfo> for NativeBlockInfo {
-  fn from(value: BlockInfo) -> Self {
-    Self {
-      block_id: value.block_id,
-      flavour: value.flavour,
-      content: value.content,
-      blob: value.blob,
-      ref_doc_id: value.ref_doc_id,
-      ref_info: value.ref_info,
-      parent_flavour: value.parent_flavour,
-      parent_block_id: value.parent_block_id,
-      additional: value.additional,
-    }
-  }
-}
-
-#[napi(object)]
-#[derive(Debug, Serialize)]
-pub struct NativeCrawlResult {
-  pub blocks: Vec<NativeBlockInfo>,
-  pub title: String,
-  pub summary: String,
-}
-
-impl From<CrawlResult> for NativeCrawlResult {
-  fn from(value: CrawlResult) -> Self {
-    Self {
-      blocks: value.blocks.into_iter().map(Into::into).collect(),
-      title: value.title,
-      summary: value.summary,
-    }
-  }
-}
-
-#[napi(object)]
-#[derive(Debug, Serialize)]
-pub struct NativeSearchHit {
-  pub id: String,
-  pub score: f64,
-  pub terms: Vec<String>,
-}
-
-impl From<SearchHit> for NativeSearchHit {
-  fn from(value: SearchHit) -> Self {
-    Self {
-      id: value.doc_id,
-      score: value.score,
-      terms: value.matched_terms.into_iter().map(|t| t.term).collect(),
-    }
-  }
-}
-
-#[napi(object)]
-#[derive(Debug, Serialize)]
-pub struct NativeMatch {
-  pub start: u32,
-  pub end: u32,
-}
-
-impl From<(u32, u32)> for NativeMatch {
-  fn from(value: (u32, u32)) -> Self {
-    Self {
-      start: value.0,
-      end: value.1,
-    }
-  }
-}
-
-impl SqliteDocStorage {
-  pub async fn crawl_doc_data(&self, doc_id: &str) -> Result<NativeCrawlResult> {
-    let doc_bin = self.load_doc_binary(doc_id).await?.ok_or(ParseError::DocNotFound)?;
-
-    let result = parse_doc_from_binary(doc_bin, doc_id.to_string())?;
-    Ok(result.into())
-  }
-
-  async fn load_doc_binary(&self, doc_id: &str) -> Result<Option<Vec<u8>>> {
-    let snapshot = self.get_doc_snapshot(doc_id.to_string()).await?;
-    let mut updates = self.get_doc_updates(doc_id.to_string()).await?;
-
-    if snapshot.is_none() && updates.is_empty() {
-      return Ok(None);
-    }
-
-    updates.sort_by_key(|a| a.timestamp);
-
-    let mut segments = Vec::with_capacity(snapshot.as_ref().map(|_| 1).unwrap_or(0) + updates.len());
-    if let Some(record) = snapshot {
-      segments.push(record.bin.to_vec());
-    }
-    segments.extend(updates.into_iter().map(|update| update.bin.to_vec()));
-
-    merge_updates(segments).map(Some)
-  }
-
-  pub async fn init_index(&self) -> Result<()> {
-    let snapshots = sqlx::query("SELECT index_name, data FROM idx_snapshots")
-      .fetch_all(&self.pool)
-      .await?;
-
-    {
-      let mut index = self.index.write().await;
-      let config = bincode::config::standard();
-      for row in snapshots {
-        let index_name: String = row.get("index_name");
-        let data: Vec<u8> = row.get("data");
-        if let Ok(decompressed) = zstd::stream::decode_all(std::io::Cursor::new(&data))
-          && let Ok((snapshot, _)) = bincode::serde::decode_from_slice::<SnapshotData, _>(&decompressed, config)
-        {
-          index.load_snapshot(&index_name, snapshot);
-        }
-      }
-    }
-
-    Ok(())
-  }
-
-  async fn compact_index(&self, index_name: &str) -> Result<()> {
-    let snapshot_data = {
-      let index = self.index.read().await;
-      index.get_snapshot_data(index_name)
-    };
-
-    if let Some(data) = snapshot_data {
-      let blob = bincode::serde::encode_to_vec(&data, bincode::config::standard())
-        .map_err(|e| Error::Serialization(e.to_string()))?;
-      let compressed =
-        zstd::stream::encode_all(std::io::Cursor::new(&blob), 4).map_err(|e| Error::Serialization(e.to_string()))?;
-
-      let mut tx = self.pool.begin().await?;
-
-      sqlx::query("INSERT OR REPLACE INTO idx_snapshots (index_name, data) VALUES (?, ?)")
-        .bind(index_name)
-        .bind(compressed)
-        .execute(&mut *tx)
-        .await?;
-
-      tx.commit().await?;
-    }
-    Ok(())
-  }
-
-  pub async fn flush_index(&self) -> Result<()> {
-    let (dirty_docs, deleted_docs) = {
-      let mut index = self.index.write().await;
-      index.take_dirty_and_deleted()
-    };
-
-    if dirty_docs.is_empty() && deleted_docs.is_empty() {
-      return Ok(());
-    }
-
-    let mut modified_indices = std::collections::HashSet::new();
-    for index_name in deleted_docs.keys() {
-      modified_indices.insert(index_name.clone());
-    }
-    for (index_name, _, _, _) in &dirty_docs {
-      modified_indices.insert(index_name.clone());
-    }
-
-    for index_name in modified_indices {
-      self.compact_index(&index_name).await?;
-    }
-
-    Ok(())
-  }
-
-  pub fn index_version() -> u32 {
-    memory_indexer::InMemoryIndex::snapshot_version() + NBSTORE_INDEXER_VERSION
-  }
-
-  pub async fn fts_add(&self, index_name: &str, doc_id: &str, text: &str, index: bool) -> Result<()> {
-    let mut idx = self.index.write().await;
-    idx.add_doc(index_name, doc_id, text, index);
-    Ok(())
-  }
-
-  pub async fn fts_delete(&self, index_name: &str, doc_id: &str) -> Result<()> {
-    let mut idx = self.index.write().await;
-    idx.remove_doc(index_name, doc_id);
-    Ok(())
-  }
-
-  pub async fn fts_get(&self, index_name: &str, doc_id: &str) -> Result<Option<String>> {
-    let idx = self.index.read().await;
-    Ok(idx.get_doc(index_name, doc_id))
-  }
-
-  pub async fn fts_search(&self, index_name: &str, query: &str) -> Result<Vec<NativeSearchHit>> {
-    let idx = self.index.read().await;
-    Ok(idx.search_hits(index_name, query).into_iter().map(Into::into).collect())
-  }
-
-  pub async fn fts_get_matches(&self, index_name: &str, doc_id: &str, query: &str) -> Result<Vec<NativeMatch>> {
-    let idx = self.index.read().await;
-    Ok(
-      idx
-        .get_matches(index_name, doc_id, query)
-        .into_iter()
-        .map(Into::into)
-        .collect(),
-    )
-  }
-
-  pub async fn fts_get_matches_for_terms(
-    &self,
-    index_name: &str,
-    doc_id: &str,
-    terms: Vec<String>,
-  ) -> Result<Vec<NativeMatch>> {
-    let idx = self.index.read().await;
-    Ok(
-      idx
-        .get_matches_for_terms(index_name, doc_id, &terms)
-        .into_iter()
-        .map(Into::into)
-        .collect(),
-    )
-  }
-}
-
-fn merge_updates(mut segments: Vec<Vec<u8>>) -> Result<Vec<u8>> {
-  if segments.is_empty() {
-    return Err(ParseError::DocNotFound.into());
-  }
-
-  if segments.len() == 1 {
-    return segments.pop().ok_or(ParseError::DocNotFound.into());
-  }
-
-  let update = merge_updates_v1(segments).map_err(|_| ParseError::InvalidBinary)?;
-  let buffer = update
-    .encode_v1()
-    .map_err(|err| ParseError::ParserError(err.to_string()))?;
-
-  Ok(buffer)
-}
-
-#[cfg(test)]
-mod tests {
-  use std::path::{Path, PathBuf};
-
-  use affine_doc_loader::ParseError;
-  use assert_json_diff::assert_json_eq;
-  use chrono::Utc;
-  use serde_json::Value;
-  use tokio::fs;
-  use uuid::Uuid;
-
-  use super::{super::error::Error, *};
-
-  const DEMO_BIN: &[u8] = include_bytes!("../../../../common/native/fixtures/demo.ydoc");
-  const DEMO_JSON: &[u8] = include_bytes!("../../../../common/native/fixtures/demo.ydoc.json");
-
-  fn temp_workspace_dir() -> PathBuf {
-    std::env::temp_dir().join(format!("affine-native-{}", Uuid::new_v4()))
-  }
-
-  async fn init_db(path: &Path) -> SqliteDocStorage {
-    fs::create_dir_all(path.parent().unwrap()).await.unwrap();
-    let storage = SqliteDocStorage::new(path.to_string_lossy().into_owned());
-    storage.connect().await.unwrap();
-    storage
-  }
-
-  async fn cleanup(path: &Path) {
-    let _ = fs::remove_dir_all(path.parent().unwrap()).await;
-  }
-
-  #[tokio::test]
-  async fn parse_demo_snapshot_matches_fixture() {
-    let base = temp_workspace_dir();
-    fs::create_dir_all(&base).await.unwrap();
-    let db_path = base.join("storage.db");
-
-    let storage = init_db(&db_path).await;
-    sqlx::query(r#"INSERT INTO snapshots (doc_id, data, updated_at) VALUES (?, ?, ?)"#)
-      .bind("demo-doc")
-      .bind(DEMO_BIN)
-      .bind(Utc::now().naive_utc())
-      .execute(&storage.pool)
-      .await
-      .unwrap();
-    sqlx::query(r#"INSERT INTO updates (doc_id, data, created_at) VALUES (?, ?, ?)"#)
-      .bind("demo-doc")
-      .bind(&[0, 0][..])
-      .bind(Utc::now().naive_utc())
-      .execute(&storage.pool)
-      .await
-      .unwrap();
-
-    let result = storage.crawl_doc_data("demo-doc").await.unwrap();
-
-    let mut expected: Value = serde_json::from_slice(DEMO_JSON).unwrap();
-    let mut actual = serde_json::to_value(&result).unwrap();
-    for document in [&mut expected, &mut actual] {
-      for block in document["blocks"].as_array_mut().unwrap() {
-        if let Some(additional) = block["additional"].as_str() {
-          block["additional"] = serde_json::from_str(additional).unwrap();
-        }
-      }
-    }
-    assert_json_eq!(expected, actual);
-
-    storage.close().await;
-    cleanup(&db_path).await;
-  }
-
-  #[tokio::test]
-  async fn missing_doc_returns_error() {
-    let base = temp_workspace_dir();
-    fs::create_dir_all(&base).await.unwrap();
-    let db_path = base.join("storage.db");
-
-    let storage = init_db(&db_path).await;
-
-    let err = storage.crawl_doc_data("absent-doc").await.unwrap_err();
-    assert!(matches!(err, Error::Parse(ParseError::DocNotFound)));
-
-    storage.close().await;
-    cleanup(&db_path).await;
-  }
-}

packages/frontend/native/nbstore/src/indexer/crawl.rs

@@ -0,0 +1,39 @@
+use affine_doc_loader::{ParseError, parse_doc_from_binary};
+use y_octo::merge_updates_v1;
+
+use super::{NativeCrawlResult, SqliteDocStorage, error::Result};
+
+impl SqliteDocStorage {
+  pub async fn crawl_doc_data(&self, doc_id: &str) -> Result<NativeCrawlResult> {
+    let doc_bin = self.load_doc_binary(doc_id).await?.ok_or(ParseError::DocNotFound)?;
+    Ok(parse_doc_from_binary(doc_bin, doc_id.to_string())?.into())
+  }
+
+  async fn load_doc_binary(&self, doc_id: &str) -> Result<Option<Vec<u8>>> {
+    let snapshot = self.get_doc_snapshot(doc_id.to_string()).await?;
+    let mut updates = self.get_doc_updates(doc_id.to_string()).await?;
+    if snapshot.is_none() && updates.is_empty() {
+      return Ok(None);
+    }
+    updates.sort_by_key(|update| update.timestamp);
+    let mut segments = Vec::with_capacity(snapshot.as_ref().map(|_| 1).unwrap_or(0) + updates.len());
+    if let Some(record) = snapshot {
+      segments.push(record.bin.to_vec());
+    }
+    segments.extend(updates.into_iter().map(|update| update.bin.to_vec()));
+    merge_updates(segments).map(Some)
+  }
+}
+
+fn merge_updates(mut segments: Vec<Vec<u8>>) -> Result<Vec<u8>> {
+  if segments.is_empty() {
+    return Err(ParseError::DocNotFound.into());
+  }
+  if segments.len() == 1 {
+    return segments.pop().ok_or(ParseError::DocNotFound.into());
+  }
+  let update = merge_updates_v1(segments).map_err(|_| ParseError::InvalidBinary)?;
+  update
+    .encode_v1()
+    .map_err(|error| ParseError::ParserError(error.to_string()).into())
+}

packages/frontend/native/nbstore/src/indexer/dto.rs

@@ -0,0 +1,142 @@
+use affine_doc_loader::{BlockInfo, CrawlResult};
+use napi_derive::napi;
+use serde::{Deserialize, Serialize};
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeBlockInfo {
+  pub block_id: String,
+  pub flavour: String,
+  pub content: Option<Vec<String>>,
+  pub blob: Option<Vec<String>>,
+  pub ref_doc_id: Option<Vec<String>>,
+  pub ref_info: Option<Vec<String>>,
+  pub parent_flavour: Option<String>,
+  pub parent_block_id: Option<String>,
+  pub additional: Option<String>,
+}
+
+impl From<BlockInfo> for NativeBlockInfo {
+  fn from(value: BlockInfo) -> Self {
+    Self {
+      block_id: value.block_id,
+      flavour: value.flavour,
+      content: value.content,
+      blob: value.blob,
+      ref_doc_id: value.ref_doc_id,
+      ref_info: value.ref_info,
+      parent_flavour: value.parent_flavour,
+      parent_block_id: value.parent_block_id,
+      additional: value.additional,
+    }
+  }
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeCrawlResult {
+  pub blocks: Vec<NativeBlockInfo>,
+  pub title: String,
+  pub summary: String,
+}
+
+impl From<CrawlResult> for NativeCrawlResult {
+  fn from(value: CrawlResult) -> Self {
+    Self {
+      blocks: value.blocks.into_iter().map(Into::into).collect(),
+      title: value.title,
+      summary: value.summary,
+    }
+  }
+}
+
+#[napi(object)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NativeIndexField {
+  pub field: String,
+  pub values: Vec<String>,
+}
+
+#[napi(object)]
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NativeIndexDocument {
+  pub id: String,
+  pub fields: Vec<NativeIndexField>,
+}
+
+#[napi(object)]
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NativeIndexQuery {
+  pub kind: String,
+  pub field: Option<String>,
+  pub value: Option<String>,
+  pub occur: Option<String>,
+  pub clauses: Option<Vec<NativeIndexQuery>>,
+  pub boost: Option<f64>,
+}
+
+#[napi(object)]
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NativeIndexSearchOptions {
+  pub limit: u32,
+  pub offset: u32,
+  pub fields: Vec<String>,
+  pub highlights: Vec<String>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexSpan {
+  pub start: u32,
+  pub end: u32,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexHighlightValue {
+  pub value_index: u32,
+  pub spans: Vec<NativeIndexSpan>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexHighlight {
+  pub field: String,
+  pub values: Vec<NativeIndexHighlightValue>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexHit {
+  pub id: String,
+  pub score: f64,
+  pub fields: Vec<NativeIndexField>,
+  pub highlights: Vec<NativeIndexHighlight>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexSearchResult {
+  pub total: u32,
+  pub hits: Vec<NativeIndexHit>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexBucket {
+  pub key: String,
+  pub count: u32,
+  pub score: f64,
+  pub hits: Vec<NativeIndexHit>,
+}
+
+#[napi(object)]
+#[derive(Debug, Serialize)]
+pub struct NativeIndexAggregateResult {
+  pub total: u32,
+  pub buckets: Vec<NativeIndexBucket>,
+}

packages/frontend/native/nbstore/src/indexer/mod.rs

@@ -0,0 +1,71 @@
+mod crawl;
+mod dto;
+mod persistence;
+mod table;
+
+use std::sync::atomic::{AtomicBool, Ordering};
+
+pub use dto::{
+  NativeBlockInfo, NativeCrawlResult, NativeIndexAggregateResult, NativeIndexBucket, NativeIndexDocument,
+  NativeIndexField, NativeIndexHighlight, NativeIndexHighlightValue, NativeIndexHit, NativeIndexQuery,
+  NativeIndexSearchOptions, NativeIndexSearchResult, NativeIndexSpan,
+};
+use error::{Error, Result};
+use memory_indexer::MemoryIndex;
+pub(super) use table::{TableIndex, string_values};
+
+pub(super) use super::{DocIndexedClock, error, storage::SqliteDocStorage};
+
+const NBSTORE_INDEXER_VERSION: u32 = 7;
+
+pub struct IndexManager {
+  pub(super) doc: TableIndex,
+  pub(super) block: TableIndex,
+  pub(super) ready: AtomicBool,
+}
+
+impl IndexManager {
+  pub fn new() -> Self {
+    Self {
+      doc: TableIndex::doc(),
+      block: TableIndex::block(),
+      ready: AtomicBool::new(true),
+    }
+  }
+
+  pub(super) fn table(&self, table: &str) -> Result<&TableIndex> {
+    match table {
+      "doc" => Ok(&self.doc),
+      "block" => Ok(&self.block),
+      _ => Err(Error::Serialization(format!("unknown index table {table}"))),
+    }
+  }
+
+  pub(super) fn tables(&self) -> [(&'static str, &TableIndex); 2] {
+    [("doc", &self.doc), ("block", &self.block)]
+  }
+
+  pub(super) fn ensure_ready(&self) -> Result<()> {
+    self
+      .ready
+      .load(Ordering::Acquire)
+      .then_some(())
+      .ok_or(Error::IndexNotReady)
+  }
+
+  pub(super) async fn reset(&self) {
+    for (_, table) in self.tables() {
+      *table.index.write().await = MemoryIndex::new(table.schema.clone());
+    }
+    self.ready.store(false, Ordering::Release);
+  }
+}
+
+impl Default for IndexManager {
+  fn default() -> Self {
+    Self::new()
+  }
+}
+
+#[cfg(test)]
+mod tests;

packages/frontend/native/nbstore/src/indexer/persistence.rs

@@ -0,0 +1,193 @@
+use std::sync::atomic::Ordering;
+
+use memory_indexer::{Document, MemoryIndex, TermsAggregation, Value};
+use sqlx::Row;
+
+use super::{
+  DocIndexedClock, NBSTORE_INDEXER_VERSION, NativeIndexAggregateResult, NativeIndexBucket, NativeIndexDocument,
+  NativeIndexQuery, NativeIndexSearchOptions, NativeIndexSearchResult, SqliteDocStorage, error::Result, string_values,
+};
+
+impl SqliteDocStorage {
+  pub async fn init_index(&self) -> Result<()> {
+    let snapshots = sqlx::query("SELECT index_name, data FROM idx_snapshots WHERE index_name IN ('doc', 'block')")
+      .fetch_all(&self.pool)
+      .await?;
+    let mut corrupted = false;
+    for row in snapshots {
+      let name: String = row.get("index_name");
+      let data: Vec<u8> = row.get("data");
+      let table = self.indexes.table(&name)?;
+      if let Ok(index) = MemoryIndex::from_checkpoint(table.schema.clone(), &data) {
+        *table.index.write().await = index;
+      } else {
+        corrupted = true;
+      }
+    }
+    if corrupted {
+      self.indexes.reset().await;
+      let mut tx = self.pool.begin().await?;
+      sqlx::query("DELETE FROM idx_snapshots").execute(&mut *tx).await?;
+      sqlx::query("DELETE FROM indexer_sync").execute(&mut *tx).await?;
+      tx.commit().await?;
+    }
+    sqlx::query("DELETE FROM idx_snapshots WHERE index_name NOT IN ('doc', 'block')")
+      .execute(&self.pool)
+      .await?;
+    Ok(())
+  }
+
+  pub async fn flush_index(&self) -> Result<()> {
+    let mut checkpoints = Vec::new();
+    for (name, table) in self.indexes.tables() {
+      let index = table.index.read().await;
+      if index.has_unpersisted_changes() {
+        checkpoints.push((name, index.checkpoint()?));
+      }
+    }
+    if checkpoints.is_empty() {
+      return Ok(());
+    }
+    let mut tx = self.pool.begin().await?;
+    for (name, checkpoint) in &checkpoints {
+      sqlx::query("INSERT OR REPLACE INTO idx_snapshots (index_name, data) VALUES (?, ?)")
+        .bind(name)
+        .bind(&checkpoint.bytes)
+        .execute(&mut *tx)
+        .await?;
+    }
+    tx.commit().await?;
+    for (name, checkpoint) in checkpoints {
+      self
+        .indexes
+        .table(name)?
+        .index
+        .write()
+        .await
+        .mark_checkpoint_persisted(checkpoint.sequence)?;
+    }
+    Ok(())
+  }
+
+  pub async fn commit_indexed_clocks(&self, clocks: &[DocIndexedClock]) -> Result<()> {
+    let mut checkpoints = Vec::new();
+    for (name, table) in self.indexes.tables() {
+      checkpoints.push((name, table.index.read().await.checkpoint()?));
+    }
+    let mut tx = self.pool.begin().await?;
+    for (name, checkpoint) in &checkpoints {
+      sqlx::query("INSERT OR REPLACE INTO idx_snapshots (index_name, data) VALUES (?, ?)")
+        .bind(name)
+        .bind(&checkpoint.bytes)
+        .execute(&mut *tx)
+        .await?;
+    }
+    for clock in clocks {
+      sqlx::query(
+        r#"INSERT INTO indexer_sync (doc_id, indexed_clock, indexer_version)
+           VALUES ($1, $2, $3)
+           ON CONFLICT(doc_id)
+           DO UPDATE SET indexed_clock=$2, indexer_version=$3"#,
+      )
+      .bind(&clock.doc_id)
+      .bind(clock.timestamp)
+      .bind(clock.indexer_version)
+      .execute(&mut *tx)
+      .await?;
+    }
+    tx.commit().await?;
+    for (name, checkpoint) in checkpoints {
+      self
+        .indexes
+        .table(name)?
+        .index
+        .write()
+        .await
+        .mark_checkpoint_persisted(checkpoint.sequence)?;
+    }
+    self.indexes.ready.store(true, Ordering::Release);
+    Ok(())
+  }
+
+  pub fn index_version() -> u32 {
+    NBSTORE_INDEXER_VERSION
+  }
+
+  pub async fn index_upsert(&self, table: &str, document: NativeIndexDocument) -> Result<()> {
+    let table = self.indexes.table(table)?;
+    let mut index_document = Document::new(document.id);
+    for field in document.fields {
+      let field_id = table.field(&field.field)?;
+      index_document.add_values(field_id, field.values.into_iter().map(Value::String));
+    }
+    table.index.write().await.upsert(index_document)?;
+    Ok(())
+  }
+
+  pub async fn index_delete(&self, table: &str, id: &str) -> Result<()> {
+    self.indexes.table(table)?.index.write().await.delete(id);
+    Ok(())
+  }
+
+  pub async fn index_search(
+    &self,
+    table: &str,
+    query: NativeIndexQuery,
+    options: NativeIndexSearchOptions,
+  ) -> Result<NativeIndexSearchResult> {
+    self.indexes.ensure_ready()?;
+    let table = self.indexes.table(table)?;
+    let query = table.compile_query(query)?;
+    let result = table
+      .index
+      .read()
+      .await
+      .search(&query, table.compile_options(options)?)?;
+    Ok(NativeIndexSearchResult {
+      total: result.total as u32,
+      hits: result.hits.into_iter().map(|hit| table.hit(hit)).collect(),
+    })
+  }
+
+  pub async fn index_aggregate(
+    &self,
+    table: &str,
+    query: NativeIndexQuery,
+    field: &str,
+    limit: u32,
+    offset: u32,
+    hits: Option<NativeIndexSearchOptions>,
+  ) -> Result<NativeIndexAggregateResult> {
+    self.indexes.ensure_ready()?;
+    let table = self.indexes.table(table)?;
+    let query = table.compile_query(query)?;
+    let result = table.index.read().await.aggregate(
+      &query,
+      TermsAggregation {
+        field: table.field(field)?,
+        limit: limit as usize,
+        offset: offset as usize,
+        top_hits: hits.map(|options| table.compile_options(options)).transpose()?,
+      },
+    )?;
+    Ok(NativeIndexAggregateResult {
+      total: result.total as u32,
+      buckets: result
+        .buckets
+        .into_iter()
+        .map(|bucket| NativeIndexBucket {
+          key: string_values(vec![bucket.key]).pop().unwrap_or_default(),
+          count: bucket.count as u32,
+          score: bucket.max_score as f64,
+          hits: bucket.hits.into_iter().map(|hit| table.hit(hit)).collect(),
+        })
+        .collect(),
+    })
+  }
+
+  pub async fn index_delete_by_query(&self, table: &str, query: NativeIndexQuery) -> Result<u32> {
+    let table = self.indexes.table(table)?;
+    let query = table.compile_query(query)?;
+    Ok(table.index.write().await.delete_by_query(&query)? as u32)
+  }
+}

packages/frontend/native/nbstore/src/indexer/table.rs

@@ -0,0 +1,223 @@
+use std::collections::HashMap;
+
+use memory_indexer::{
+  FieldId, FieldOptions, FieldType, Highlight, MemoryIndex, PositionEncoding, Query, Schema, SearchMode, SearchOptions,
+  TextOptions, Value,
+};
+use tokio::sync::RwLock;
+
+use super::{
+  NativeIndexField, NativeIndexHighlight, NativeIndexHighlightValue, NativeIndexHit, NativeIndexQuery,
+  NativeIndexSearchOptions, NativeIndexSpan,
+  error::{Error, Result},
+};
+
+pub(crate) struct TableIndex {
+  pub(super) schema: Schema,
+  fields: HashMap<String, FieldId>,
+  pub(super) index: RwLock<MemoryIndex>,
+}
+
+impl TableIndex {
+  pub(super) fn doc() -> Self {
+    let mut schema = Schema::builder().position_encoding(PositionEncoding::Utf16);
+    let fields = HashMap::from([
+      ("docId".into(), schema.keyword("docId", FieldOptions::indexed_stored())),
+      (
+        "title".into(),
+        schema.text("title", text_options(), FieldOptions::indexed_stored()),
+      ),
+      (
+        "summary".into(),
+        schema.keyword("summary", FieldOptions::new().stored()),
+      ),
+    ]);
+    Self::from_schema(schema, fields)
+  }
+
+  pub(super) fn block() -> Self {
+    let mut schema = Schema::builder().position_encoding(PositionEncoding::Utf16);
+    let fields = HashMap::from([
+      ("docId".into(), schema.keyword("docId", keyword_options())),
+      ("blockId".into(), schema.keyword("blockId", keyword_options())),
+      (
+        "content".into(),
+        schema.text("content", text_options(), FieldOptions::indexed_stored().multi_value()),
+      ),
+      ("flavour".into(), schema.keyword("flavour", keyword_options())),
+      ("blob".into(), schema.keyword("blob", keyword_options())),
+      ("refDocId".into(), schema.keyword("refDocId", keyword_options())),
+      (
+        "ref".into(),
+        schema.keyword("ref", FieldOptions::new().stored().multi_value()),
+      ),
+      (
+        "parentFlavour".into(),
+        schema.keyword("parentFlavour", keyword_options()),
+      ),
+      (
+        "parentBlockId".into(),
+        schema.keyword("parentBlockId", keyword_options()),
+      ),
+      (
+        "additional".into(),
+        schema.keyword("additional", FieldOptions::new().stored().multi_value()),
+      ),
+      (
+        "markdownPreview".into(),
+        schema.keyword("markdownPreview", FieldOptions::new().stored().multi_value()),
+      ),
+    ]);
+    Self::from_schema(schema, fields)
+  }
+
+  fn from_schema(builder: memory_indexer::SchemaBuilder, fields: HashMap<String, FieldId>) -> Self {
+    let schema = builder.build().expect("static nbstore index schema must be valid");
+    Self {
+      index: RwLock::new(MemoryIndex::new(schema.clone())),
+      schema,
+      fields,
+    }
+  }
+
+  pub(super) fn field(&self, name: &str) -> Result<FieldId> {
+    self
+      .fields
+      .get(name)
+      .copied()
+      .ok_or_else(|| Error::Serialization(format!("unknown index field {name}")))
+  }
+
+  pub(super) fn compile_query(&self, query: NativeIndexQuery) -> Result<Query> {
+    match query.kind.as_str() {
+      "match" => {
+        let field = self.field(query.field.as_deref().unwrap_or_default())?;
+        let value = query.value.unwrap_or_default();
+        match self.schema.field(field).map(|field| &field.field_type) {
+          Some(FieldType::Text(_)) => Ok(Query::text(field, value, SearchMode::Auto)),
+          Some(FieldType::Keyword) => Ok(Query::term(field, value)),
+          _ => Err(Error::Serialization("match requires Text or Keyword field".into())),
+        }
+      }
+      "exists" => Ok(Query::Exists(self.field(query.field.as_deref().unwrap_or_default())?)),
+      "all" => Ok(Query::All),
+      "boost" => {
+        let clause = query
+          .clauses
+          .unwrap_or_default()
+          .pop()
+          .ok_or_else(|| Error::Serialization("boost requires one clause".into()))?;
+        Ok(Query::Boost {
+          query: Box::new(self.compile_query(clause)?),
+          factor: query.boost.unwrap_or(1.0) as f32,
+        })
+      }
+      "boolean" => {
+        let clauses = query
+          .clauses
+          .unwrap_or_default()
+          .into_iter()
+          .map(|query| self.compile_query(query))
+          .collect::<Result<Vec<_>>>()?;
+        let (must, should, must_not) = match query.occur.as_deref() {
+          Some("must") => (clauses, vec![], vec![]),
+          Some("should") => (vec![], clauses, vec![]),
+          Some("must_not") => (vec![], vec![], clauses),
+          _ => return Err(Error::Serialization("invalid boolean occurrence".into())),
+        };
+        Ok(Query::boolean(must, should, must_not))
+      }
+      kind => Err(Error::Serialization(format!("unknown query kind {kind}"))),
+    }
+  }
+
+  pub(super) fn compile_options(&self, options: NativeIndexSearchOptions) -> Result<SearchOptions> {
+    Ok(SearchOptions {
+      limit: options.limit as usize,
+      offset: options.offset as usize,
+      after: None,
+      sort: vec![],
+      stored_fields: options
+        .fields
+        .iter()
+        .map(|field| self.field(field))
+        .collect::<Result<Vec<_>>>()?,
+      highlight_fields: options
+        .highlights
+        .iter()
+        .map(|field| self.field(field))
+        .collect::<Result<Vec<_>>>()?,
+    })
+  }
+
+  pub(super) fn hit(&self, hit: memory_indexer::SearchHit) -> NativeIndexHit {
+    NativeIndexHit {
+      id: hit.id,
+      score: hit.score as f64,
+      fields: hit
+        .fields
+        .into_iter()
+        .map(|(field, values)| NativeIndexField {
+          field: self.field_name(field),
+          values: string_values(values),
+        })
+        .collect(),
+      highlights: group_highlights(self, hit.highlights),
+    }
+  }
+
+  fn field_name(&self, field: FieldId) -> String {
+    self
+      .schema
+      .field(field)
+      .expect("result field belongs to schema")
+      .name
+      .clone()
+  }
+}
+
+pub(crate) fn string_values(values: Vec<Value>) -> Vec<String> {
+  values
+    .into_iter()
+    .filter_map(|value| match value {
+      Value::String(value) => Some(value),
+      Value::I64(_) | Value::Bool(_) => None,
+    })
+    .collect()
+}
+
+fn keyword_options() -> FieldOptions {
+  FieldOptions::indexed_stored().multi_value()
+}
+
+fn text_options() -> TextOptions {
+  TextOptions::multilingual()
+    .with_pinyin()
+    .with_prefix()
+    .with_fuzzy()
+    .with_positions()
+}
+
+fn group_highlights(table: &TableIndex, highlights: Vec<Highlight>) -> Vec<NativeIndexHighlight> {
+  let mut grouped: Vec<NativeIndexHighlight> = Vec::new();
+  for highlight in highlights {
+    let field = table.field_name(highlight.field);
+    let values = NativeIndexHighlightValue {
+      value_index: highlight.value_index as u32,
+      spans: highlight
+        .spans
+        .into_iter()
+        .map(|(start, end)| NativeIndexSpan { start, end })
+        .collect(),
+    };
+    if let Some(existing) = grouped.iter_mut().find(|item| item.field == field) {
+      existing.values.push(values);
+    } else {
+      grouped.push(NativeIndexHighlight {
+        field,
+        values: vec![values],
+      });
+    }
+  }
+  grouped
+}

packages/frontend/native/nbstore/src/indexer/tests.rs

@@ -0,0 +1,318 @@
+use std::path::{Path, PathBuf};
+
+use affine_doc_loader::ParseError;
+use assert_json_diff::assert_json_eq;
+use chrono::Utc;
+use serde_json::Value;
+use tokio::fs;
+use uuid::Uuid;
+
+use super::{
+  super::{DocIndexedClock, error::Error, storage::SqliteDocStorage},
+  NativeIndexDocument, NativeIndexField, NativeIndexQuery, NativeIndexSearchOptions,
+};
+
+const DEMO_BIN: &[u8] = include_bytes!("../../../../../common/native/fixtures/demo.ydoc");
+const DEMO_JSON: &[u8] = include_bytes!("../../../../../common/native/fixtures/demo.ydoc.json");
+
+fn temp_workspace_dir() -> PathBuf {
+  std::env::temp_dir().join(format!("affine-native-{}", Uuid::new_v4()))
+}
+
+async fn init_db(path: &Path) -> SqliteDocStorage {
+  fs::create_dir_all(path.parent().unwrap()).await.unwrap();
+  let storage = SqliteDocStorage::new(path.to_string_lossy().into_owned());
+  storage.connect().await.unwrap();
+  storage
+}
+
+async fn cleanup(path: &Path) {
+  let _ = fs::remove_dir_all(path.parent().unwrap()).await;
+}
+
+fn query(kind: &str, field: Option<&str>, value: Option<&str>) -> NativeIndexQuery {
+  NativeIndexQuery {
+    kind: kind.into(),
+    field: field.map(Into::into),
+    value: value.map(Into::into),
+    occur: None,
+    clauses: None,
+    boost: None,
+  }
+}
+
+fn options(fields: &[&str], highlights: &[&str]) -> NativeIndexSearchOptions {
+  NativeIndexSearchOptions {
+    limit: 10,
+    offset: 0,
+    fields: fields.iter().map(|value| (*value).into()).collect(),
+    highlights: highlights.iter().map(|value| (*value).into()).collect(),
+  }
+}
+
+fn document(id: &str, fields: &[(&str, &[&str])]) -> NativeIndexDocument {
+  NativeIndexDocument {
+    id: id.into(),
+    fields: fields
+      .iter()
+      .map(|(field, values)| NativeIndexField {
+        field: (*field).into(),
+        values: values.iter().map(|value| (*value).into()).collect(),
+      })
+      .collect(),
+  }
+}
+
+#[tokio::test]
+async fn parse_demo_snapshot_matches_fixture() {
+  let base = temp_workspace_dir();
+  fs::create_dir_all(&base).await.unwrap();
+  let db_path = base.join("storage.db");
+  let storage = init_db(&db_path).await;
+  sqlx::query(r#"INSERT INTO snapshots (doc_id, data, updated_at) VALUES (?, ?, ?)"#)
+    .bind("demo-doc")
+    .bind(DEMO_BIN)
+    .bind(Utc::now().naive_utc())
+    .execute(&storage.pool)
+    .await
+    .unwrap();
+  sqlx::query(r#"INSERT INTO updates (doc_id, data, created_at) VALUES (?, ?, ?)"#)
+    .bind("demo-doc")
+    .bind(&[0, 0][..])
+    .bind(Utc::now().naive_utc())
+    .execute(&storage.pool)
+    .await
+    .unwrap();
+  let result = storage.crawl_doc_data("demo-doc").await.unwrap();
+  let mut expected: Value = serde_json::from_slice(DEMO_JSON).unwrap();
+  let mut actual = serde_json::to_value(&result).unwrap();
+  for document in [&mut expected, &mut actual] {
+    for block in document["blocks"].as_array_mut().unwrap() {
+      if let Some(additional) = block["additional"].as_str() {
+        block["additional"] = serde_json::from_str(additional).unwrap();
+      }
+    }
+  }
+  assert_json_eq!(expected, actual);
+  storage.close().await;
+  cleanup(&db_path).await;
+}
+
+#[tokio::test]
+async fn missing_doc_returns_error() {
+  let db_path = temp_workspace_dir().join("storage.db");
+  let storage = init_db(&db_path).await;
+  let error = storage.crawl_doc_data("absent-doc").await.unwrap_err();
+  assert!(matches!(error, Error::Parse(ParseError::DocNotFound)));
+  storage.close().await;
+  cleanup(&db_path).await;
+}
+
+#[tokio::test]
+async fn index_tables_support_terminal_queries_and_restart() {
+  let db_path = temp_workspace_dir().join("storage.db");
+  let storage = init_db(&db_path).await;
+  storage
+    .index_upsert(
+      "doc",
+      document(
+        "doc-1",
+        &[
+          ("docId", &["doc-1"]),
+          ("title", &["Rust 搜索"]),
+          ("summary", &["stored summary"]),
+        ],
+      ),
+    )
+    .await
+    .unwrap();
+  storage
+    .index_upsert(
+      "block",
+      document(
+        "block-1",
+        &[
+          ("docId", &["doc-1"]),
+          ("blockId", &["block-1"]),
+          ("content", &["hello world", "你好搜索"]),
+          ("flavour", &["affine:paragraph"]),
+        ],
+      ),
+    )
+    .await
+    .unwrap();
+  storage
+    .index_upsert(
+      "block",
+      document(
+        "block-2",
+        &[
+          ("docId", &["doc-10"]),
+          ("blockId", &["block-2"]),
+          ("content", &["hello unrelated"]),
+          ("flavour", &["affine:code"]),
+        ],
+      ),
+    )
+    .await
+    .unwrap();
+
+  let text = storage
+    .index_search(
+      "block",
+      query("match", Some("content"), Some("搜索")),
+      options(&["content"], &["content"]),
+    )
+    .await
+    .unwrap();
+  assert_eq!(text.total, 1);
+  assert_eq!(text.hits[0].id, "block-1");
+  assert!(!text.hits[0].highlights[0].values[0].spans.is_empty());
+
+  let exact = storage
+    .index_search(
+      "block",
+      query("match", Some("docId"), Some("doc-1")),
+      options(&["docId"], &[]),
+    )
+    .await
+    .unwrap();
+  assert_eq!(exact.total, 1);
+  assert_eq!(exact.hits[0].fields[0].values, ["doc-1"]);
+
+  let aggregate = storage
+    .index_aggregate(
+      "block",
+      query("all", None, None),
+      "flavour",
+      10,
+      0,
+      Some(options(&["blockId"], &[])),
+    )
+    .await
+    .unwrap();
+  assert_eq!(aggregate.total, 2);
+  assert_eq!(aggregate.buckets.len(), 2);
+
+  let clock = DocIndexedClock {
+    doc_id: "doc-1".into(),
+    timestamp: Utc::now().naive_utc(),
+    indexer_version: SqliteDocStorage::index_version() as i64,
+  };
+  storage.commit_indexed_clocks(&[clock]).await.unwrap();
+  storage.close().await;
+
+  let restored = init_db(&db_path).await;
+  let result = restored
+    .index_search(
+      "block",
+      query("match", Some("content"), Some("hello")),
+      options(&[], &[]),
+    )
+    .await
+    .unwrap();
+  assert_eq!(result.total, 2);
+  assert_eq!(
+    restored
+      .index_delete_by_query("block", query("match", Some("docId"), Some("doc-1")))
+      .await
+      .unwrap(),
+    1
+  );
+  restored.close().await;
+  cleanup(&db_path).await;
+}
+
+#[tokio::test]
+async fn corrupt_checkpoint_requires_rebuild_and_clears_clocks() {
+  let db_path = temp_workspace_dir().join("storage.db");
+  let storage = init_db(&db_path).await;
+  storage
+    .index_upsert(
+      "doc",
+      document("doc-1", &[("docId", &["doc-1"]), ("title", &["title"])]),
+    )
+    .await
+    .unwrap();
+  let clock = DocIndexedClock {
+    doc_id: "doc-1".into(),
+    timestamp: Utc::now().naive_utc(),
+    indexer_version: SqliteDocStorage::index_version() as i64,
+  };
+  storage.commit_indexed_clocks(&[clock]).await.unwrap();
+  sqlx::query("UPDATE idx_snapshots SET data = x'00' WHERE index_name = 'doc'")
+    .execute(&storage.pool)
+    .await
+    .unwrap();
+  storage.close().await;
+
+  let restored = init_db(&db_path).await;
+  assert!(matches!(
+    restored
+      .index_search("doc", query("all", None, None), options(&[], &[]))
+      .await,
+    Err(Error::IndexNotReady)
+  ));
+  assert!(restored.get_doc_indexed_clock("doc-1".into()).await.unwrap().is_none());
+  restored
+    .index_upsert(
+      "doc",
+      document("doc-1", &[("docId", &["doc-1"]), ("title", &["rebuilt"])]),
+    )
+    .await
+    .unwrap();
+  restored
+    .commit_indexed_clocks(&[DocIndexedClock {
+      doc_id: "doc-1".into(),
+      timestamp: Utc::now().naive_utc(),
+      indexer_version: SqliteDocStorage::index_version() as i64,
+    }])
+    .await
+    .unwrap();
+  assert_eq!(
+    restored
+      .index_search("doc", query("all", None, None), options(&[], &[]))
+      .await
+      .unwrap()
+      .total,
+    1
+  );
+  restored.close().await;
+  cleanup(&db_path).await;
+}
+
+#[tokio::test]
+async fn failed_atomic_commit_keeps_index_dirty_and_clock_unadvanced() {
+  let db_path = temp_workspace_dir().join("storage.db");
+  let storage = init_db(&db_path).await;
+  storage
+    .index_upsert(
+      "doc",
+      document("doc-1", &[("docId", &["doc-1"]), ("title", &["title"])]),
+    )
+    .await
+    .unwrap();
+  sqlx::query("DROP TABLE idx_snapshots")
+    .execute(&storage.pool)
+    .await
+    .unwrap();
+  let clock = DocIndexedClock {
+    doc_id: "doc-1".into(),
+    timestamp: Utc::now().naive_utc(),
+    indexer_version: SqliteDocStorage::index_version() as i64,
+  };
+  assert!(storage.commit_indexed_clocks(&[clock]).await.is_err());
+  assert!(
+    storage
+      .indexes
+      .table("doc")
+      .unwrap()
+      .index
+      .read()
+      .await
+      .has_unpersisted_changes()
+  );
+  assert!(storage.get_doc_indexed_clock("doc-1".into()).await.unwrap().is_none());
+  storage.close().await;
+  cleanup(&db_path).await;
+}

packages/frontend/native/nbstore/src/lib.rs

@@ -220,6 +220,12 @@ impl DocStoragePool {
     Ok(())
   }
 
+  #[napi]
+  pub async fn set_doc_indexed_clocks(&self, universal_id: String, clocks: Vec<DocIndexedClock>) -> Result<()> {
+    self.get(universal_id).await?.commit_indexed_clocks(&clocks).await?;
+    Ok(())
+  }
+
   #[napi]
   pub async fn clear_doc_indexed_clock(&self, universal_id: String, doc_id: String) -> Result<()> {
     self.get(universal_id).await?.clear_doc_indexed_clock(doc_id).await?;
@@ -410,65 +416,72 @@ impl DocStoragePool {
   }
 
   #[napi]
-  pub async fn fts_add_document(
-    &self,
-    id: String,
-    index_name: String,
-    doc_id: String,
-    text: String,
-    index: bool,
-  ) -> Result<()> {
+  pub async fn index_upsert(&self, id: String, table: String, document: indexer::NativeIndexDocument) -> Result<()> {
     let storage = self.pool.get(id).await?;
-    storage.fts_add(&index_name, &doc_id, &text, index).await?;
+    storage.index_upsert(&table, document).await?;
     Ok(())
   }
 
   #[napi]
-  pub async fn fts_flush_index(&self, id: String) -> Result<()> {
+  pub async fn index_flush(&self, id: String) -> Result<()> {
     let storage = self.pool.get(id).await?;
     storage.flush_index().await?;
     Ok(())
   }
 
   #[napi]
-  pub async fn fts_index_version(&self) -> Result<u32> {
+  pub async fn index_version(&self) -> Result<u32> {
     Ok(SqliteDocStorage::index_version())
   }
 
   #[napi]
-  pub async fn fts_delete_document(&self, id: String, index_name: String, doc_id: String) -> Result<()> {
+  pub async fn index_delete(&self, id: String, table: String, doc_id: String) -> Result<()> {
     let storage = self.pool.get(id).await?;
-    storage.fts_delete(&index_name, &doc_id).await?;
+    storage.index_delete(&table, &doc_id).await?;
     Ok(())
   }
 
   #[napi]
-  pub async fn fts_get_document(&self, id: String, index_name: String, doc_id: String) -> Result<Option<String>> {
+  pub async fn index_search(
+    &self,
+    id: String,
+    table: String,
+    query: indexer::NativeIndexQuery,
+    options: indexer::NativeIndexSearchOptions,
+  ) -> Result<indexer::NativeIndexSearchResult> {
     let storage = self.pool.get(id).await?;
-    Ok(storage.fts_get(&index_name, &doc_id).await?)
+    Ok(storage.index_search(&table, query, options).await?)
   }
 
   #[napi]
-  pub async fn fts_search(
+  #[allow(clippy::too_many_arguments)]
+  pub async fn index_aggregate(
     &self,
     id: String,
-    index_name: String,
-    query: String,
-  ) -> Result<Vec<indexer::NativeSearchHit>> {
+    table: String,
+    query: indexer::NativeIndexQuery,
+    field: String,
+    limit: u32,
+    offset: u32,
+    hits: Option<indexer::NativeIndexSearchOptions>,
+  ) -> Result<indexer::NativeIndexAggregateResult> {
     let storage = self.pool.get(id).await?;
-    Ok(storage.fts_search(&index_name, &query).await?)
+    Ok(
+      storage
+        .index_aggregate(&table, query, &field, limit, offset, hits)
+        .await?,
+    )
   }
 
   #[napi]
-  pub async fn fts_get_matches(
+  pub async fn index_delete_by_query(
     &self,
     id: String,
-    index_name: String,
-    doc_id: String,
-    query: String,
-  ) -> Result<Vec<indexer::NativeMatch>> {
+    table: String,
+    query: indexer::NativeIndexQuery,
+  ) -> Result<u32> {
     let storage = self.pool.get(id).await?;
-    Ok(storage.fts_get_matches(&index_name, &doc_id, &query).await?)
+    Ok(storage.index_delete_by_query(&table, query).await?)
   }
 }
 

packages/frontend/native/nbstore/src/storage.rs

@@ -4,20 +4,18 @@ use affine_schema::{
   get_migrator,
   import_validation::{V2_IMPORT_SCHEMA_RULES, validate_import_schema, validate_required_schema},
 };
-use memory_indexer::InMemoryIndex;
 use sqlx::{
   Pool, Row,
   migrate::{MigrateDatabase, Migration, Migrator},
   sqlite::{Sqlite, SqliteConnectOptions, SqlitePoolOptions},
 };
-use tokio::sync::RwLock;
 
-use super::error::Result;
+use super::{error::Result, indexer::IndexManager};
 
 pub struct SqliteDocStorage {
   pub pool: Pool<Sqlite>,
   path: String,
-  pub index: Arc<RwLock<InMemoryIndex>>,
+  pub indexes: Arc<IndexManager>,
 }
 
 impl SqliteDocStorage {
@@ -26,7 +24,7 @@ impl SqliteDocStorage {
 
     let mut pool_options = SqlitePoolOptions::new();
 
-    let index = Arc::new(RwLock::new(InMemoryIndex::default()));
+    let indexes = Arc::new(IndexManager::new());
 
     if path == ":memory:" {
       pool_options = pool_options
@@ -38,7 +36,7 @@ impl SqliteDocStorage {
       Self {
         pool: pool_options.connect_lazy_with(sqlite_options),
         path,
-        index,
+        indexes,
       }
     } else {
       Self {
@@ -46,7 +44,7 @@ impl SqliteDocStorage {
           .max_connections(4)
           .connect_lazy_with(sqlite_options.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)),
         path,
-        index,
+        indexes,
       }
     }
   }