mf_search/
service.rs

1use crate::backend::SqliteBackend;
2use crate::indexer::mutations_from_step;
3use crate::model::IndexDoc;
4use anyhow::Result;
5use mf_model::node_pool::NodePool;
6use mf_transform::step::Step;
7use std::sync::Arc;
8use mf_state::transaction::Transaction;
9
10/// 外部可投递的事件
11#[derive(Debug, Clone)]
12pub enum IndexEvent {
13    /// 增量：来自单个 step（可选地提供变更前的 `NodePool` 以支持删除子树精确处理）
14    StepApplied {
15        pool_before: Option<Arc<NodePool>>,
16        pool_after: Arc<NodePool>,
17        step: Arc<dyn Step>,
18    },
19    /// 增量：来自一整个事务（可选地提供 before 池）
20    TransactionCommitted {
21        pool_before: Option<Arc<NodePool>>,
22        pool_after: Arc<NodePool>,
23        steps: Vec<Arc<dyn Step>>,
24    },
25    /// 全量重建
26    Rebuild { pool: Arc<NodePool>, scope: RebuildScope },
27}
28
29#[derive(Debug, Clone, Copy)]
30pub enum RebuildScope {
31    Full,
32}
33
34/// 索引服务：桥接 `Transaction/Step` 与后端
35pub struct IndexService {
36    backend: Arc<SqliteBackend>,
37}
38
39impl IndexService {
40    pub fn new(backend: Arc<SqliteBackend>) -> Self {
41        Self { backend }
42    }
43
44    /// 处理事件（调度后端执行）
45    pub async fn handle(
46        &self,
47        event: IndexEvent,
48    ) -> Result<()> {
49        match event {
50            IndexEvent::StepApplied { pool_before, pool_after, step } => {
51                let pool_b = pool_before.as_deref().unwrap_or(&pool_after);
52                let muts = mutations_from_step(pool_b, &pool_after, &step);
53                self.backend.apply(muts).await
54            },
55            IndexEvent::TransactionCommitted {
56                pool_before,
57                pool_after,
58                steps,
59            } => {
60                let pool_b = pool_before.as_deref().unwrap_or(&pool_after);
61                // 合并事务中所有 step 的增量（可能有覆盖）
62                let mut all = Vec::new();
63                for s in &steps {
64                    all.extend(mutations_from_step(pool_b, &pool_after, s));
65                }
66                self.backend.apply(all).await
67            },
68            IndexEvent::Rebuild { pool, scope: RebuildScope::Full } => {
69                // 并行/顺序遍历整个池，构造文档集合
70                let mut docs: Vec<IndexDoc> = Vec::new();
71                for shard in &pool.get_inner().nodes {
72                    for node in shard.values() {
73                        docs.push(IndexDoc::from_node(&pool, node));
74                    }
75                }
76                self.backend.rebuild_all(docs).await
77            },
78        }
79    }
80}
81
82/// 搜索服务：提供高层查询接口
83pub struct SearchService {
84    backend: Arc<SqliteBackend>,
85}
86
87impl SearchService {
88    pub fn new(backend: Arc<SqliteBackend>) -> Self {
89        Self { backend }
90    }
91
92    /// 简单查询：返回节点 ID 列表
93    pub async fn search(
94        &self,
95        query: crate::backend::SearchQuery,
96    ) -> Result<Vec<String>> {
97        self.backend.search_ids(query).await
98    }
99
100    /// 查询并返回完整文档
101    pub async fn search_docs(
102        &self,
103        query: crate::backend::SearchQuery,
104    ) -> Result<Vec<IndexDoc>> {
105        self.backend.search_docs(query).await
106    }
107
108    /// 全文搜索
109    pub async fn search_text(
110        &self,
111        text: &str,
112        limit: usize,
113    ) -> Result<Vec<String>> {
114        self.backend
115            .search_ids(crate::backend::SearchQuery {
116                text: Some(text.to_string()),
117                limit,
118                ..Default::default()
119            })
120            .await
121    }
122
123    /// 全文搜索（返回完整文档）
124    pub async fn search_text_docs(
125        &self,
126        text: &str,
127        limit: usize,
128    ) -> Result<Vec<IndexDoc>> {
129        self.backend
130            .search_docs(crate::backend::SearchQuery {
131                text: Some(text.to_string()),
132                limit,
133                ..Default::default()
134            })
135            .await
136    }
137
138    /// 查询子树（递归）
139    pub async fn query_descendants(
140        &self,
141        parent_id: &str,
142        limit: usize,
143    ) -> Result<Vec<String>> {
144        self.backend
145            .search_ids(crate::backend::SearchQuery {
146                parent_id: Some(parent_id.to_string()),
147                include_descendants: true,
148                limit,
149                ..Default::default()
150            })
151            .await
152    }
153
154    /// 按类型查询
155    pub async fn query_by_type(
156        &self,
157        node_type: &str,
158        limit: usize,
159    ) -> Result<Vec<String>> {
160        self.backend
161            .search_ids(crate::backend::SearchQuery {
162                node_type: Some(node_type.to_string()),
163                limit,
164                ..Default::default()
165            })
166            .await
167    }
168}
169
170#[allow(dead_code)]
171pub fn event_from_transaction(
172    pool_after: Arc<NodePool>,
173    tr: &Transaction,
174) -> IndexEvent {
175    let steps: Vec<Arc<dyn Step>> = tr.steps.iter().cloned().collect();
176    IndexEvent::TransactionCommitted { pool_before: None, pool_after, steps }
177}
mf_search/service.rs

mf_search/
service.rs
