sz_orm_es/
lib.rs

1//! # SZ-ORM ES — Elasticsearch 文档存储（**MOCK-ONLY，非生产可用**）
2//!
3//! ⚠️ **重要警告：本 crate 仅为内存 Mock 实现，未集成真实 Elasticsearch 客户端。**
4//!
5//! - 所有数据存储在进程内 `HashMap`，重启即丢失
6//! - 不支持分布式、副本、分片、持久化、TLS、认证等任何 ES 生产特性
7//! - 查询语义为简化版，不保证与真实 ES 行为一致
8//! - **请勿用于生产环境**——仅适用于单元测试与本地开发
9//!
10//! ## 替代方案
11//!
12//! 如需真实 ES 集成，请：
13//! 1. 实现 [`EsSync`] trait（接入 `elasticsearch` 官方 crate）
14//! 2. 通过 [`EsSyncManager::with_backend`] 注入自定义后端
15//!
16//! ## 主要类型
17//!
18//! - [`EsDocument`] — 文档模型
19//! - [`EsSearchRequest`] — 搜索请求
20//! - [`EsSync`] — 同步后端抽象（可被真实 ES 客户端实现）
21//! - [`InMemoryEsSync`] — **Mock 实现**（仅供测试）
22//! - [`EsSyncManager`] — 同步管理器（默认使用 Mock，可通过 `with_backend` 替换）
23
24pub mod extensions;
25
26use serde::{Deserialize, Serialize};
27use std::collections::HashMap;
28use std::sync::{Arc, RwLock};
29
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub struct EsDocument {
32    pub id: Option<String>,
33    pub index: String,
34    pub source: serde_json::Value,
35    pub timestamp: i64,
36}
37
38impl EsDocument {
39    pub fn new(index: impl Into<String>, source: serde_json::Value) -> Self {
40        Self {
41            id: None,
42            index: index.into(),
43            source,
44            timestamp: current_timestamp(),
45        }
46    }
47
48    pub fn with_id(mut self, id: impl Into<String>) -> Self {
49        self.id = Some(id.into());
50        self
51    }
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct EsSearchRequest {
56    pub index: String,
57    pub query: EsQuery,
58    pub from: usize,
59    pub size: usize,
60    pub sort: Vec<EsSort>,
61}
62
63impl EsSearchRequest {
64    pub fn new(index: impl Into<String>, query: EsQuery) -> Self {
65        Self {
66            index: index.into(),
67            query,
68            from: 0,
69            size: 10,
70            sort: Vec::new(),
71        }
72    }
73
74    pub fn with_pagination(mut self, from: usize, size: usize) -> Self {
75        self.from = from;
76        self.size = size;
77        self
78    }
79
80    pub fn with_sort(mut self, field: impl Into<String>, order: EsSortOrder) -> Self {
81        self.sort.push(EsSort {
82            field: field.into(),
83            order,
84        });
85        self
86    }
87}
88
89#[derive(Debug, Clone, Serialize, Deserialize)]
90pub struct EsSort {
91    pub field: String,
92    pub order: EsSortOrder,
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
96pub enum EsSortOrder {
97    Asc,
98    Desc,
99}
100
101#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
102#[serde(untagged)]
103pub enum EsQuery {
104    MatchAll,
105    Term(HashMap<String, serde_json::Value>),
106    Terms(HashMap<String, Vec<serde_json::Value>>),
107    Range(HashMap<String, EsRangeQuery>),
108    Bool(EsBoolQuery),
109}
110
111impl EsQuery {
112    pub fn match_all() -> Self {
113        EsQuery::MatchAll
114    }
115
116    pub fn term(field: impl Into<String>, value: serde_json::Value) -> Self {
117        let mut terms = HashMap::new();
118        terms.insert(field.into(), value);
119        EsQuery::Term(terms)
120    }
121
122    pub fn terms(field: impl Into<String>, values: Vec<serde_json::Value>) -> Self {
123        let mut terms = HashMap::new();
124        terms.insert(field.into(), values);
125        EsQuery::Terms(terms)
126    }
127
128    pub fn range(field: impl Into<String>, range: EsRangeQuery) -> Self {
129        let mut ranges = HashMap::new();
130        ranges.insert(field.into(), range);
131        EsQuery::Range(ranges)
132    }
133
134    pub fn must(queries: Vec<EsQuery>) -> Self {
135        EsQuery::Bool(EsBoolQuery {
136            must: Some(queries),
137            should: None,
138            filter: None,
139            must_not: None,
140            minimum_should_match: None,
141        })
142    }
143
144    pub fn should(queries: Vec<EsQuery>) -> Self {
145        EsQuery::Bool(EsBoolQuery {
146            must: None,
147            should: Some(queries),
148            filter: None,
149            must_not: None,
150            minimum_should_match: None,
151        })
152    }
153}
154
155#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
156pub struct EsBoolQuery {
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub must: Option<Vec<EsQuery>>,
159    #[serde(skip_serializing_if = "Option::is_none")]
160    pub should: Option<Vec<EsQuery>>,
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub filter: Option<Vec<EsQuery>>,
163    #[serde(skip_serializing_if = "Option::is_none")]
164    pub must_not: Option<Vec<EsQuery>>,
165    #[serde(skip_serializing_if = "Option::is_none")]
166    pub minimum_should_match: Option<usize>,
167}
168
169#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
170pub struct EsRangeQuery {
171    #[serde(skip_serializing_if = "Option::is_none")]
172    pub gt: Option<serde_json::Value>,
173    #[serde(skip_serializing_if = "Option::is_none")]
174    pub gte: Option<serde_json::Value>,
175    #[serde(skip_serializing_if = "Option::is_none")]
176    pub lt: Option<serde_json::Value>,
177    #[serde(skip_serializing_if = "Option::is_none")]
178    pub lte: Option<serde_json::Value>,
179}
180
181impl EsRangeQuery {
182    pub fn new() -> Self {
183        Self {
184            gt: None,
185            gte: None,
186            lt: None,
187            lte: None,
188        }
189    }
190
191    pub fn gt(mut self, value: serde_json::Value) -> Self {
192        self.gt = Some(value);
193        self
194    }
195
196    pub fn gte(mut self, value: serde_json::Value) -> Self {
197        self.gte = Some(value);
198        self
199    }
200
201    pub fn lt(mut self, value: serde_json::Value) -> Self {
202        self.lt = Some(value);
203        self
204    }
205
206    pub fn lte(mut self, value: serde_json::Value) -> Self {
207        self.lte = Some(value);
208        self
209    }
210}
211
212impl Default for EsRangeQuery {
213    fn default() -> Self {
214        Self::new()
215    }
216}
217
218#[derive(Debug, Clone, Serialize, Deserialize)]
219pub struct EsSearchResult {
220    pub total: usize,
221    pub hits: Vec<EsHit>,
222    pub took: i64,
223}
224
225#[derive(Debug, Clone, Serialize, Deserialize)]
226pub struct EsHit {
227    pub id: String,
228    pub score: f64,
229    pub source: serde_json::Value,
230}
231
232pub trait EsSync: Send + Sync {
233    fn sync_to_es(&self, documents: Vec<EsDocument>) -> Result<EsSyncResult, EsError>;
234    fn delete_from_es(&self, index: &str, ids: Vec<String>) -> Result<EsSyncResult, EsError>;
235    fn search(&self, request: EsSearchRequest) -> Result<EsSearchResult, EsError>;
236}
237
238#[derive(Debug, Clone, Serialize, Deserialize)]
239pub struct EsSyncResult {
240    pub indexed: usize,
241    pub failed: usize,
242    pub errors: Vec<String>,
243}
244
245impl EsSyncResult {
246    pub fn success(indexed: usize) -> Self {
247        Self {
248            indexed,
249            failed: 0,
250            errors: Vec::new(),
251        }
252    }
253
254    pub fn with_errors(indexed: usize, errors: Vec<String>) -> Self {
255        let failed = errors.len();
256        Self {
257            indexed,
258            failed,
259            errors,
260        }
261    }
262}
263
264pub struct EsSyncManager {
265    index_mappings: HashMap<String, HashMap<String, EsFieldType>>,
266    /// 后端实现（默认为 `InMemoryEsSync` Mock，可通过 `with_backend` 注入真实 ES 客户端）
267    backend: Box<dyn EsSync>,
268    /// Mock 后端引用（仅在 `new()` 默认构造时存在，用于支持 `count` 等 Mock 专属方法）
269    mock_backend: Option<InMemoryEsSync>,
270}
271
272impl EsSyncManager {
273    /// 创建默认 Manager（**使用 Mock 后端，仅供测试**）。
274    ///
275    /// ⚠️ **生产环境请使用 [`EsSyncManager::with_backend`] 注入真实 ES 客户端实现。**
276    pub fn new() -> Self {
277        let mock = InMemoryEsSync::new();
278        Self {
279            index_mappings: HashMap::new(),
280            backend: Box::new(mock.clone()),
281            mock_backend: Some(mock),
282        }
283    }
284
285    /// 使用指定后端创建 Manager（用于注入真实 ES 客户端）。
286    ///
287    /// # 示例
288    ///
289    /// ```ignore
290    /// use sz_orm_es::{EsSyncManager, EsSync};
291    ///
292    /// struct RealEsClient { /* elasticsearch::http::transport::Transport */ }
293    /// impl EsSync for RealEsClient { /* ... */ }
294    ///
295    /// let manager = EsSyncManager::with_backend(Box::new(RealEsClient { /* ... */ }));
296    /// ```
297    pub fn with_backend(backend: Box<dyn EsSync>) -> Self {
298        Self {
299            index_mappings: HashMap::new(),
300            backend,
301            mock_backend: None,
302        }
303    }
304
305    pub fn create_index(
306        &mut self,
307        index: impl Into<String>,
308        mapping: HashMap<String, EsFieldType>,
309    ) {
310        self.index_mappings.insert(index.into(), mapping);
311    }
312
313    pub fn get_mapping(&self, index: &str) -> Option<&HashMap<String, EsFieldType>> {
314        self.index_mappings.get(index)
315    }
316
317    pub fn sync_to_es(&self, documents: Vec<EsDocument>) -> Result<EsSyncResult, EsError> {
318        self.backend.sync_to_es(documents)
319    }
320
321    pub fn delete_from_es(&self, index: &str, ids: Vec<String>) -> Result<EsSyncResult, EsError> {
322        self.backend.delete_from_es(index, ids)
323    }
324
325    pub fn search(&self, request: EsSearchRequest) -> Result<EsSearchResult, EsError> {
326        self.backend.search(request)
327    }
328
329    /// Returns the total number of documents stored in `index`.
330    ///
331    /// ⚠️ 仅在默认 `new()` 构造（Mock 后端）时可用；使用 `with_backend` 注入自定义后端时返回 `EsError::BackendNotSupported`。
332    pub fn count(&self, index: &str) -> Result<usize, EsError> {
333        match &self.mock_backend {
334            Some(mock) => mock.count(index),
335            None => Err(EsError::BackendNotSupported {
336                operation: "count".to_string(),
337                reason: "current backend is not InMemoryEsSync; implement count on your EsSync impl".to_string(),
338            }),
339        }
340    }
341}
342
343impl Default for EsSyncManager {
344    fn default() -> Self {
345        Self::new()
346    }
347}
348
349/// In-memory `EsSync` implementation backed by a `HashMap<index, HashMap<id, EsDocument>>`.
350///
351/// ⚠️ **MOCK 实现——仅供单元测试与本地开发，请勿用于生产环境。**
352///
353/// - 数据存储在进程内 `HashMap`，重启即丢失
354/// - 不支持分布式、副本、分片、持久化等任何 ES 生产特性
355/// - 查询语义为简化版，不保证与真实 ES 行为一致
356///
357/// Suitable for unit tests and small in-process workloads.
358/// `search` supports `MatchAll`, `Term` (exact match), `Terms` (membership),
359/// `Range` (numeric/string bounds), and `Bool` (must/should/must_not).
360///
361/// `Clone` 通过 `Arc` 共享内部状态——克隆的是引用而非数据副本。
362#[derive(Clone)]
363pub struct InMemoryEsSync {
364    documents: Arc<RwLock<HashMap<String, HashMap<String, EsDocument>>>>,
365}
366
367/// `MockEsSync` 是 [`InMemoryEsSync`] 的语义化别名，明确表达"测试用 Mock"含义。
368///
369/// 在新代码中推荐使用 `MockEsSync` 替代 `InMemoryEsSync`，以让"非生产可用"的含义更清晰。
370pub type MockEsSync = InMemoryEsSync;
371
372impl InMemoryEsSync {
373    pub fn new() -> Self {
374        Self {
375            documents: Arc::new(RwLock::new(HashMap::new())),
376        }
377    }
378
379    pub fn count(&self, index: &str) -> Result<usize, EsError> {
380        let docs = self
381            .documents
382            .read()
383            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
384        Ok(docs.get(index).map(|m| m.len()).unwrap_or(0))
385    }
386
387    fn generate_id(doc: &EsDocument, counter: usize) -> String {
388        doc.id
389            .clone()
390            .unwrap_or_else(|| format!("auto-{}-{}", doc.timestamp, counter))
391    }
392
393    fn matches_query(doc: &EsDocument, query: &EsQuery) -> bool {
394        match query {
395            EsQuery::MatchAll => true,
396            EsQuery::Term(terms) => {
397                if let Some(obj) = doc.source.as_object() {
398                    terms.iter().all(|(k, v)| obj.get(k) == Some(v))
399                } else {
400                    false
401                }
402            }
403            EsQuery::Terms(terms) => {
404                let Some(obj) = doc.source.as_object() else {
405                    return false;
406                };
407                terms.iter().all(|(k, values)| {
408                    if let Some(v) = obj.get(k) {
409                        values.contains(v)
410                    } else {
411                        false
412                    }
413                })
414            }
415            EsQuery::Range(ranges) => {
416                let Some(obj) = doc.source.as_object() else {
417                    return false;
418                };
419                ranges.iter().all(|(field, range)| match obj.get(field) {
420                    Some(value) => range_match(value, range),
421                    None => false,
422                })
423            }
424            EsQuery::Bool(b) => bool_match(doc, b),
425        }
426    }
427}
428
429impl Default for InMemoryEsSync {
430    fn default() -> Self {
431        Self::new()
432    }
433}
434
435#[allow(clippy::neg_cmp_op_on_partial_ord)]
436fn range_match(value: &serde_json::Value, range: &EsRangeQuery) -> bool {
437    match (value.as_f64(), value.as_str()) {
438        (Some(num), _) => {
439            if let Some(gt) = range.gt.as_ref().and_then(|v| v.as_f64()) {
440                if !(num > gt) {
441                    return false;
442                }
443            }
444            if let Some(gte) = range.gte.as_ref().and_then(|v| v.as_f64()) {
445                if !(num >= gte) {
446                    return false;
447                }
448            }
449            if let Some(lt) = range.lt.as_ref().and_then(|v| v.as_f64()) {
450                if !(num < lt) {
451                    return false;
452                }
453            }
454            if let Some(lte) = range.lte.as_ref().and_then(|v| v.as_f64()) {
455                if !(num <= lte) {
456                    return false;
457                }
458            }
459            true
460        }
461        (_, Some(s)) => {
462            if let Some(gt) = range.gt.as_ref().and_then(|v| v.as_str()) {
463                if s <= gt {
464                    return false;
465                }
466            }
467            if let Some(gte) = range.gte.as_ref().and_then(|v| v.as_str()) {
468                if s < gte {
469                    return false;
470                }
471            }
472            if let Some(lt) = range.lt.as_ref().and_then(|v| v.as_str()) {
473                if s >= lt {
474                    return false;
475                }
476            }
477            if let Some(lte) = range.lte.as_ref().and_then(|v| v.as_str()) {
478                if s > lte {
479                    return false;
480                }
481            }
482            true
483        }
484        _ => false,
485    }
486}
487
488fn bool_match(doc: &EsDocument, b: &EsBoolQuery) -> bool {
489    if let Some(must) = &b.must {
490        if !must.iter().all(|q| InMemoryEsSync::matches_query(doc, q)) {
491            return false;
492        }
493    }
494    if let Some(must_not) = &b.must_not {
495        if must_not
496            .iter()
497            .any(|q| InMemoryEsSync::matches_query(doc, q))
498        {
499            return false;
500        }
501    }
502    if let Some(filter) = &b.filter {
503        if !filter.iter().all(|q| InMemoryEsSync::matches_query(doc, q)) {
504            return false;
505        }
506    }
507    if let Some(should) = &b.should {
508        let matched = should
509            .iter()
510            .filter(|q| InMemoryEsSync::matches_query(doc, q))
511            .count();
512        let min_match = b.minimum_should_match.unwrap_or(1);
513        if matched < min_match {
514            return false;
515        }
516    }
517    true
518}
519
520impl EsSync for InMemoryEsSync {
521    fn sync_to_es(&self, documents: Vec<EsDocument>) -> Result<EsSyncResult, EsError> {
522        let mut store = self
523            .documents
524            .write()
525            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
526        let mut indexed = 0usize;
527        let mut errors: Vec<String> = Vec::new();
528        for (i, doc) in documents.into_iter().enumerate() {
529            if doc.index.is_empty() {
530                errors.push(format!("document {} has empty index", i));
531                continue;
532            }
533            let id = Self::generate_id(&doc, i);
534            let map = store.entry(doc.index.clone()).or_default();
535            map.insert(id, doc);
536            indexed += 1;
537        }
538        if errors.is_empty() {
539            Ok(EsSyncResult::success(indexed))
540        } else {
541            Ok(EsSyncResult::with_errors(indexed, errors))
542        }
543    }
544
545    fn delete_from_es(&self, index: &str, ids: Vec<String>) -> Result<EsSyncResult, EsError> {
546        let mut store = self
547            .documents
548            .write()
549            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
550        let map = store
551            .get_mut(index)
552            .ok_or_else(|| EsError::IndexNotFound(index.to_string()))?;
553        let mut deleted = 0usize;
554        let mut errors: Vec<String> = Vec::new();
555        for id in ids {
556            if map.remove(&id).is_some() {
557                deleted += 1;
558            } else {
559                errors.push(format!("document not found: {}", id));
560            }
561        }
562        if errors.is_empty() {
563            Ok(EsSyncResult::success(deleted))
564        } else {
565            Ok(EsSyncResult::with_errors(deleted, errors))
566        }
567    }
568
569    fn search(&self, request: EsSearchRequest) -> Result<EsSearchResult, EsError> {
570        let store = self
571            .documents
572            .read()
573            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
574        let Some(map) = store.get(&request.index) else {
575            return Err(EsError::IndexNotFound(request.index.clone()));
576        };
577
578        let start = std::time::Instant::now();
579        let mut hits: Vec<(String, EsDocument)> = map
580            .iter()
581            .filter(|(_, doc)| Self::matches_query(doc, &request.query))
582            .map(|(id, doc)| (id.clone(), doc.clone()))
583            .collect();
584
585        // Sort by fields in `request.sort`.
586        for sort in request.sort.iter().rev() {
587            let field = &sort.field;
588            let asc = matches!(sort.order, EsSortOrder::Asc);
589            hits.sort_by(|a, b| {
590                let av =
591                    a.1.source
592                        .get(field)
593                        .cloned()
594                        .unwrap_or(serde_json::Value::Null);
595                let bv =
596                    b.1.source
597                        .get(field)
598                        .cloned()
599                        .unwrap_or(serde_json::Value::Null);
600                let ord = compare_values(&av, &bv);
601                if asc {
602                    ord
603                } else {
604                    ord.reverse()
605                }
606            });
607        }
608
609        let total = hits.len();
610        let paged = hits
611            .into_iter()
612            .skip(request.from)
613            .take(request.size)
614            .map(|(id, doc)| EsHit {
615                id,
616                score: 1.0,
617                source: doc.source,
618            })
619            .collect::<Vec<_>>();
620
621        Ok(EsSearchResult {
622            total,
623            hits: paged,
624            took: start.elapsed().as_millis() as i64,
625        })
626    }
627}
628
629fn compare_values(a: &serde_json::Value, b: &serde_json::Value) -> std::cmp::Ordering {
630    use std::cmp::Ordering;
631    match (a.as_f64(), b.as_f64()) {
632        (Some(x), Some(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
633        _ => a.to_string().cmp(&b.to_string()),
634    }
635}
636
637#[derive(Debug, Clone, Serialize, Deserialize)]
638pub enum EsFieldType {
639    Text,
640    Keyword,
641    Integer,
642    Long,
643    Float,
644    Double,
645    Boolean,
646    Date,
647    Object,
648    Nested,
649    Ip,
650    GeoPoint,
651}
652
653#[derive(Debug)]
654pub enum EsError {
655    ConnectionFailed(String),
656    IndexNotFound(String),
657    DocumentNotFound(String),
658    MappingError(String),
659    QueryError(String),
660    SyncError(String),
661    /// 当前后端不支持的操作（如 `EsSyncManager::with_backend` 注入自定义后端后调用 Mock 专属方法）
662    BackendNotSupported {
663        operation: String,
664        reason: String,
665    },
666}
667
668impl std::fmt::Display for EsError {
669    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
670        match self {
671            EsError::ConnectionFailed(msg) => write!(f, "Connection failed: {}", msg),
672            EsError::IndexNotFound(idx) => write!(f, "Index not found: {}", idx),
673            EsError::DocumentNotFound(id) => write!(f, "Document not found: {}", id),
674            EsError::MappingError(msg) => write!(f, "Mapping error: {}", msg),
675            EsError::QueryError(msg) => write!(f, "Query error: {}", msg),
676            EsError::SyncError(msg) => write!(f, "Sync error: {}", msg),
677            EsError::BackendNotSupported { operation, reason } => {
678                write!(f, "Backend not supported: {} ({})", operation, reason)
679            }
680        }
681    }
682}
683
684impl std::error::Error for EsError {}
685
686impl serde::Serialize for EsError {
687    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
688    where
689        S: serde::Serializer,
690    {
691        serializer.serialize_str(&self.to_string())
692    }
693}
694
695fn current_timestamp() -> i64 {
696    use std::time::{SystemTime, UNIX_EPOCH};
697    SystemTime::now()
698        .duration_since(UNIX_EPOCH)
699        .unwrap_or_default()
700        .as_millis() as i64
701}
702
703#[cfg(test)]
704mod tests {
705    use super::*;
706
707    #[test]
708    fn test_es_document_new() {
709        let doc = EsDocument::new("test-index", serde_json::json!({"name": "test"}));
710        assert_eq!(doc.index, "test-index");
711        assert_eq!(doc.source["name"], "test");
712        assert!(doc.id.is_none());
713    }
714
715    #[test]
716    fn test_es_document_with_id() {
717        let doc = EsDocument::new("test-index", serde_json::json!({})).with_id("doc1");
718        assert_eq!(doc.id, Some("doc1".to_string()));
719    }
720
721    #[test]
722    fn test_es_search_request_new() {
723        let query = EsQuery::match_all();
724        let request = EsSearchRequest::new("test-index", query);
725        assert_eq!(request.index, "test-index");
726        assert_eq!(request.from, 0);
727        assert_eq!(request.size, 10);
728    }
729
730    #[test]
731    fn test_es_search_request_pagination() {
732        let query = EsQuery::match_all();
733        let request = EsSearchRequest::new("test", query).with_pagination(20, 50);
734        assert_eq!(request.from, 20);
735        assert_eq!(request.size, 50);
736    }
737
738    #[test]
739    fn test_es_search_request_sort() {
740        let query = EsQuery::match_all();
741        let request = EsSearchRequest::new("test", query).with_sort("date", EsSortOrder::Desc);
742        assert_eq!(request.sort.len(), 1);
743        assert_eq!(request.sort[0].field, "date");
744        assert_eq!(request.sort[0].order, EsSortOrder::Desc);
745    }
746
747    #[test]
748    fn test_es_query_match_all() {
749        let query = EsQuery::match_all();
750        assert_eq!(query, EsQuery::MatchAll);
751    }
752
753    #[test]
754    fn test_es_query_term() {
755        let query = EsQuery::term("status", serde_json::json!("active"));
756        assert_eq!(
757            query,
758            EsQuery::Term(std::collections::HashMap::from([(
759                "status".to_string(),
760                serde_json::json!("active")
761            )]))
762        );
763    }
764
765    #[test]
766    fn test_es_query_terms() {
767        let query = EsQuery::terms("tags", vec![serde_json::json!("a"), serde_json::json!("b")]);
768        assert!(matches!(query, EsQuery::Terms(_)));
769    }
770
771    #[test]
772    fn test_es_query_range() {
773        let range = EsRangeQuery::new().gte(serde_json::json!(100));
774        let query = EsQuery::range("price", range);
775        assert!(matches!(query, EsQuery::Range(_)));
776    }
777
778    #[test]
779    fn test_es_query_bool_must() {
780        let queries = vec![
781            EsQuery::term("status", serde_json::json!("active")),
782            EsQuery::term("type", serde_json::json!("post")),
783        ];
784        let query = EsQuery::must(queries);
785        assert!(matches!(query, EsQuery::Bool(_)));
786    }
787
788    #[test]
789    fn test_es_range_query() {
790        let range = EsRangeQuery::new()
791            .gte(serde_json::json!(0))
792            .lt(serde_json::json!(100));
793
794        assert!(range.gte.is_some());
795        assert!(range.lt.is_some());
796    }
797
798    #[test]
799    fn test_es_sync_result_success() {
800        let result = EsSyncResult::success(10);
801        assert_eq!(result.indexed, 10);
802        assert_eq!(result.failed, 0);
803        assert!(result.errors.is_empty());
804    }
805
806    #[test]
807    fn test_es_sync_result_with_errors() {
808        let errors = vec!["error1".to_string(), "error2".to_string()];
809        let result = EsSyncResult::with_errors(8, errors.clone());
810        assert_eq!(result.indexed, 8);
811        assert_eq!(result.failed, 2);
812        assert_eq!(result.errors, errors);
813    }
814
815    #[test]
816    fn test_es_sync_manager_new() {
817        let manager = EsSyncManager::new();
818        assert!(manager.index_mappings.is_empty());
819    }
820
821    #[test]
822    fn test_es_sync_manager_create_index() {
823        let mut manager = EsSyncManager::new();
824        let mut mapping = HashMap::new();
825        mapping.insert("title".to_string(), EsFieldType::Text);
826        mapping.insert("count".to_string(), EsFieldType::Integer);
827
828        manager.create_index("test-index", mapping);
829
830        let retrieved = manager.get_mapping("test-index");
831        assert!(retrieved.is_some());
832        assert_eq!(retrieved.unwrap().len(), 2);
833    }
834
835    #[test]
836    fn test_es_sync_manager_get_mapping_not_found() {
837        let manager = EsSyncManager::new();
838        let mapping = manager.get_mapping("nonexistent");
839        assert!(mapping.is_none());
840    }
841
842    #[test]
843    fn test_inmemory_sync_and_count() {
844        let sync = InMemoryEsSync::new();
845        let docs = vec![
846            EsDocument::new("users", serde_json::json!({"name": "alice"})).with_id("u1"),
847            EsDocument::new("users", serde_json::json!({"name": "bob"})).with_id("u2"),
848        ];
849        let result = sync.sync_to_es(docs).unwrap();
850        assert_eq!(result.indexed, 2);
851        assert_eq!(sync.count("users").unwrap(), 2);
852    }
853
854    #[test]
855    fn test_inmemory_sync_replaces_by_id() {
856        let sync = InMemoryEsSync::new();
857        sync.sync_to_es(vec![
858            EsDocument::new("idx", serde_json::json!({"v": 1})).with_id("d1")
859        ])
860        .unwrap();
861        sync.sync_to_es(vec![
862            EsDocument::new("idx", serde_json::json!({"v": 2})).with_id("d1")
863        ])
864        .unwrap();
865        assert_eq!(sync.count("idx").unwrap(), 1);
866
867        let req = EsSearchRequest::new("idx", EsQuery::match_all());
868        let res = sync.search(req).unwrap();
869        assert_eq!(res.hits[0].source["v"], 2);
870    }
871
872    #[test]
873    fn test_inmemory_sync_rejects_empty_index() {
874        let sync = InMemoryEsSync::new();
875        let result = sync
876            .sync_to_es(vec![EsDocument::new("", serde_json::json!({}))])
877            .unwrap();
878        assert_eq!(result.indexed, 0);
879        assert_eq!(result.failed, 1);
880    }
881
882    #[test]
883    fn test_inmemory_delete_real() {
884        let sync = InMemoryEsSync::new();
885        sync.sync_to_es(vec![
886            EsDocument::new("i", serde_json::json!({"k": "a"})).with_id("1"),
887            EsDocument::new("i", serde_json::json!({"k": "b"})).with_id("2"),
888        ])
889        .unwrap();
890
891        let result = sync.delete_from_es("i", vec!["1".to_string()]).unwrap();
892        assert_eq!(result.indexed, 1);
893        assert_eq!(sync.count("i").unwrap(), 1);
894
895        let err = sync.delete_from_es("missing", vec!["x".to_string()]);
896        assert!(matches!(err, Err(EsError::IndexNotFound(_))));
897    }
898
899    #[test]
900    fn test_inmemory_delete_missing_records_error() {
901        let sync = InMemoryEsSync::new();
902        sync.sync_to_es(vec![
903            EsDocument::new("i", serde_json::json!({"k": "a"})).with_id("1")
904        ])
905        .unwrap();
906        let r = sync
907            .delete_from_es("i", vec!["1".to_string(), "999".to_string()])
908            .unwrap();
909        assert_eq!(r.indexed, 1);
910        assert_eq!(r.failed, 1);
911    }
912
913    #[test]
914    fn test_inmemory_search_match_all() {
915        let sync = InMemoryEsSync::new();
916        sync.sync_to_es(vec![
917            EsDocument::new("i", serde_json::json!({"n": 1})).with_id("1"),
918            EsDocument::new("i", serde_json::json!({"n": 2})).with_id("2"),
919        ])
920        .unwrap();
921        let req = EsSearchRequest::new("i", EsQuery::match_all());
922        let res = sync.search(req).unwrap();
923        assert_eq!(res.total, 2);
924        assert_eq!(res.hits.len(), 2);
925    }
926
927    #[test]
928    fn test_inmemory_search_term_filter() {
929        let sync = InMemoryEsSync::new();
930        sync.sync_to_es(vec![
931            EsDocument::new("i", serde_json::json!({"status": "active"})).with_id("1"),
932            EsDocument::new("i", serde_json::json!({"status": "inactive"})).with_id("2"),
933        ])
934        .unwrap();
935        let req = EsSearchRequest::new("i", EsQuery::term("status", serde_json::json!("active")));
936        let res = sync.search(req).unwrap();
937        assert_eq!(res.total, 1);
938        assert_eq!(res.hits[0].source["status"], "active");
939    }
940
941    #[test]
942    fn test_inmemory_search_terms_filter() {
943        let sync = InMemoryEsSync::new();
944        sync.sync_to_es(vec![
945            EsDocument::new("i", serde_json::json!({"tag": "a"})).with_id("1"),
946            EsDocument::new("i", serde_json::json!({"tag": "b"})).with_id("2"),
947            EsDocument::new("i", serde_json::json!({"tag": "c"})).with_id("3"),
948        ])
949        .unwrap();
950        let req = EsSearchRequest::new(
951            "i",
952            EsQuery::terms("tag", vec![serde_json::json!("a"), serde_json::json!("c")]),
953        );
954        let res = sync.search(req).unwrap();
955        assert_eq!(res.total, 2);
956    }
957
958    #[test]
959    fn test_inmemory_search_range_filter() {
960        let sync = InMemoryEsSync::new();
961        sync.sync_to_es(vec![
962            EsDocument::new("i", serde_json::json!({"age": 10})).with_id("1"),
963            EsDocument::new("i", serde_json::json!({"age": 20})).with_id("2"),
964            EsDocument::new("i", serde_json::json!({"age": 30})).with_id("3"),
965        ])
966        .unwrap();
967        let req = EsSearchRequest::new(
968            "i",
969            EsQuery::range("age", EsRangeQuery::new().gte(serde_json::json!(20))),
970        );
971        let res = sync.search(req).unwrap();
972        assert_eq!(res.total, 2);
973    }
974
975    #[test]
976    fn test_inmemory_search_bool_query() {
977        let sync = InMemoryEsSync::new();
978        sync.sync_to_es(vec![
979            EsDocument::new("i", serde_json::json!({"status": "active", "n": 5})).with_id("1"),
980            EsDocument::new("i", serde_json::json!({"status": "active", "n": 50})).with_id("2"),
981            EsDocument::new("i", serde_json::json!({"status": "inactive", "n": 5})).with_id("3"),
982        ])
983        .unwrap();
984        // status=active AND n>=10
985        let bool = EsQuery::must(vec![
986            EsQuery::term("status", serde_json::json!("active")),
987            EsQuery::range("n", EsRangeQuery::new().gte(serde_json::json!(10))),
988        ]);
989        let req = EsSearchRequest::new("i", bool);
990        let res = sync.search(req).unwrap();
991        assert_eq!(res.total, 1);
992        assert_eq!(res.hits[0].source["n"], 50);
993    }
994
995    #[test]
996    fn test_inmemory_search_pagination_and_sort() {
997        let sync = InMemoryEsSync::new();
998        let docs: Vec<EsDocument> = (1..=5)
999            .map(|i| EsDocument::new("i", serde_json::json!({"v": i})).with_id(format!("d{}", i)))
1000            .collect();
1001        sync.sync_to_es(docs).unwrap();
1002
1003        let req = EsSearchRequest::new("i", EsQuery::match_all())
1004            .with_pagination(0, 2)
1005            .with_sort("v", EsSortOrder::Desc);
1006        let res = sync.search(req).unwrap();
1007        assert_eq!(res.total, 5);
1008        assert_eq!(res.hits.len(), 2);
1009        // Descending sort means the highest values come first.
1010        assert_eq!(res.hits[0].source["v"], 5);
1011        assert_eq!(res.hits[1].source["v"], 4);
1012
1013        // Now test pagination: skip the first, take the next two.
1014        let req2 = EsSearchRequest::new("i", EsQuery::match_all())
1015            .with_pagination(1, 2)
1016            .with_sort("v", EsSortOrder::Desc);
1017        let res2 = sync.search(req2).unwrap();
1018        assert_eq!(res2.hits.len(), 2);
1019        assert_eq!(res2.hits[0].source["v"], 4);
1020        assert_eq!(res2.hits[1].source["v"], 3);
1021
1022        // Ascending order should reverse the result.
1023        let req3 = EsSearchRequest::new("i", EsQuery::match_all())
1024            .with_pagination(0, 2)
1025            .with_sort("v", EsSortOrder::Asc);
1026        let res3 = sync.search(req3).unwrap();
1027        assert_eq!(res3.hits[0].source["v"], 1);
1028        assert_eq!(res3.hits[1].source["v"], 2);
1029    }
1030
1031    #[test]
1032    fn test_inmemory_search_missing_index_errors() {
1033        let sync = InMemoryEsSync::new();
1034        let req = EsSearchRequest::new("missing", EsQuery::match_all());
1035        let err = sync.search(req);
1036        assert!(matches!(err, Err(EsError::IndexNotFound(_))));
1037    }
1038
1039    #[test]
1040    fn test_sync_manager_delegates_to_backend() {
1041        let mut manager = EsSyncManager::new();
1042        manager.create_index("idx", HashMap::new());
1043
1044        let r = manager
1045            .sync_to_es(vec![
1046                EsDocument::new("idx", serde_json::json!({"k": "v"})).with_id("d1")
1047            ])
1048            .unwrap();
1049        assert_eq!(r.indexed, 1);
1050        assert_eq!(manager.count("idx").unwrap(), 1);
1051
1052        let req = EsSearchRequest::new("idx", EsQuery::match_all());
1053        let res = manager.search(req).unwrap();
1054        assert_eq!(res.total, 1);
1055
1056        let r = manager
1057            .delete_from_es("idx", vec!["d1".to_string()])
1058            .unwrap();
1059        assert_eq!(r.indexed, 1);
1060        assert_eq!(manager.count("idx").unwrap(), 0);
1061    }
1062
1063    #[test]
1064    fn test_es_sync_trait_object_dyn() {
1065        // Ensures the trait is object-safe and usable via dynamic dispatch.
1066        let sync: Box<dyn EsSync> = Box::new(InMemoryEsSync::new());
1067        sync.sync_to_es(vec![
1068            EsDocument::new("i", serde_json::json!({"k": 1})).with_id("1")
1069        ])
1070        .unwrap();
1071        let req = EsSearchRequest::new("i", EsQuery::match_all());
1072        let res = sync.search(req).unwrap();
1073        assert_eq!(res.total, 1);
1074    }
1075
1076    /// 验证：`with_backend` 注入自定义后端后，`count` 应返回 `BackendNotSupported`
1077    /// （因为 mock_backend 为 None）。
1078    ///
1079    /// 这确保用户在注入真实 ES 客户端时，不会被静默地回退到 Mock 行为。
1080    #[test]
1081    fn test_with_backend_count_returns_not_supported() {
1082        // 一个假后端：sync_to_es / delete_from_es / search 都返回 Ok（空结果）
1083        struct StubBackend;
1084        impl EsSync for StubBackend {
1085            fn sync_to_es(
1086                &self,
1087                _documents: Vec<EsDocument>,
1088            ) -> Result<EsSyncResult, EsError> {
1089                Ok(EsSyncResult::success(0))
1090            }
1091            fn delete_from_es(
1092                &self,
1093                _index: &str,
1094                _ids: Vec<String>,
1095            ) -> Result<EsSyncResult, EsError> {
1096                Ok(EsSyncResult::success(0))
1097            }
1098            fn search(
1099                &self,
1100                _request: EsSearchRequest,
1101            ) -> Result<EsSearchResult, EsError> {
1102                Ok(EsSearchResult {
1103                    total: 0,
1104                    hits: Vec::new(),
1105                    took: 0,
1106                })
1107            }
1108        }
1109
1110        let manager = EsSyncManager::with_backend(Box::new(StubBackend));
1111        let result = manager.count("any_index");
1112        assert!(
1113            matches!(result, Err(EsError::BackendNotSupported { .. })),
1114            "with_backend 注入自定义后端后 count 应返回 BackendNotSupported，实际: {:?}",
1115            result
1116        );
1117
1118        // 但 sync_to_es / delete / search 应正常委托给自定义后端
1119        let sync_result = manager
1120            .sync_to_es(vec![EsDocument::new("i", serde_json::json!({}))])
1121            .unwrap();
1122        assert_eq!(sync_result.indexed, 0);
1123    }
1124
1125    /// 验证：默认 `new()` 构造的 Manager 仍可使用 `count`（Mock 后端可用）
1126    #[test]
1127    fn test_default_new_count_works_with_mock() {
1128        let manager = EsSyncManager::new();
1129        // 空索引应返回 0 而非错误
1130        let count = manager.count("empty_idx").unwrap();
1131        assert_eq!(count, 0);
1132    }
1133
1134    /// 验证：`MockEsSync` 别名与 `InMemoryEsSync` 等价
1135    #[test]
1136    fn test_mock_es_sync_alias_equivalence() {
1137        let mock: MockEsSync = MockEsSync::new();
1138        let _: InMemoryEsSync = mock; // 类型应等价
1139        let mock2 = InMemoryEsSync::new();
1140        let _clone: InMemoryEsSync = mock2.clone(); // Clone 应可用
1141    }
1142}
sz_orm_es/lib.rs

sz_orm_es/
lib.rs
