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:
338                    "current backend is not InMemoryEsSync; implement count on your EsSync impl"
339                        .to_string(),
340            }),
341        }
342    }
343}
344
345impl Default for EsSyncManager {
346    fn default() -> Self {
347        Self::new()
348    }
349}
350
351/// In-memory `EsSync` implementation backed by a `HashMap<index, HashMap<id, EsDocument>>`.
352///
353/// ⚠️ **MOCK 实现——仅供单元测试与本地开发，请勿用于生产环境。**
354///
355/// - 数据存储在进程内 `HashMap`，重启即丢失
356/// - 不支持分布式、副本、分片、持久化等任何 ES 生产特性
357/// - 查询语义为简化版，不保证与真实 ES 行为一致
358///
359/// Suitable for unit tests and small in-process workloads.
360/// `search` supports `MatchAll`, `Term` (exact match), `Terms` (membership),
361/// `Range` (numeric/string bounds), and `Bool` (must/should/must_not).
362///
363/// `Clone` 通过 `Arc` 共享内部状态——克隆的是引用而非数据副本。
364#[derive(Clone)]
365pub struct InMemoryEsSync {
366    documents: Arc<RwLock<HashMap<String, HashMap<String, EsDocument>>>>,
367}
368
369/// `MockEsSync` 是 [`InMemoryEsSync`] 的语义化别名，明确表达"测试用 Mock"含义。
370///
371/// 在新代码中推荐使用 `MockEsSync` 替代 `InMemoryEsSync`，以让"非生产可用"的含义更清晰。
372pub type MockEsSync = InMemoryEsSync;
373
374impl InMemoryEsSync {
375    pub fn new() -> Self {
376        Self {
377            documents: Arc::new(RwLock::new(HashMap::new())),
378        }
379    }
380
381    pub fn count(&self, index: &str) -> Result<usize, EsError> {
382        let docs = self
383            .documents
384            .read()
385            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
386        Ok(docs.get(index).map(|m| m.len()).unwrap_or(0))
387    }
388
389    fn generate_id(doc: &EsDocument, counter: usize) -> String {
390        doc.id
391            .clone()
392            .unwrap_or_else(|| format!("auto-{}-{}", doc.timestamp, counter))
393    }
394
395    fn matches_query(doc: &EsDocument, query: &EsQuery) -> bool {
396        match query {
397            EsQuery::MatchAll => true,
398            EsQuery::Term(terms) => {
399                if let Some(obj) = doc.source.as_object() {
400                    terms.iter().all(|(k, v)| obj.get(k) == Some(v))
401                } else {
402                    false
403                }
404            }
405            EsQuery::Terms(terms) => {
406                let Some(obj) = doc.source.as_object() else {
407                    return false;
408                };
409                terms.iter().all(|(k, values)| {
410                    if let Some(v) = obj.get(k) {
411                        values.contains(v)
412                    } else {
413                        false
414                    }
415                })
416            }
417            EsQuery::Range(ranges) => {
418                let Some(obj) = doc.source.as_object() else {
419                    return false;
420                };
421                ranges.iter().all(|(field, range)| match obj.get(field) {
422                    Some(value) => range_match(value, range),
423                    None => false,
424                })
425            }
426            EsQuery::Bool(b) => bool_match(doc, b),
427        }
428    }
429}
430
431impl Default for InMemoryEsSync {
432    fn default() -> Self {
433        Self::new()
434    }
435}
436
437#[allow(clippy::neg_cmp_op_on_partial_ord)]
438fn range_match(value: &serde_json::Value, range: &EsRangeQuery) -> bool {
439    match (value.as_f64(), value.as_str()) {
440        (Some(num), _) => {
441            if let Some(gt) = range.gt.as_ref().and_then(|v| v.as_f64()) {
442                if !(num > gt) {
443                    return false;
444                }
445            }
446            if let Some(gte) = range.gte.as_ref().and_then(|v| v.as_f64()) {
447                if !(num >= gte) {
448                    return false;
449                }
450            }
451            if let Some(lt) = range.lt.as_ref().and_then(|v| v.as_f64()) {
452                if !(num < lt) {
453                    return false;
454                }
455            }
456            if let Some(lte) = range.lte.as_ref().and_then(|v| v.as_f64()) {
457                if !(num <= lte) {
458                    return false;
459                }
460            }
461            true
462        }
463        (_, Some(s)) => {
464            if let Some(gt) = range.gt.as_ref().and_then(|v| v.as_str()) {
465                if s <= gt {
466                    return false;
467                }
468            }
469            if let Some(gte) = range.gte.as_ref().and_then(|v| v.as_str()) {
470                if s < gte {
471                    return false;
472                }
473            }
474            if let Some(lt) = range.lt.as_ref().and_then(|v| v.as_str()) {
475                if s >= lt {
476                    return false;
477                }
478            }
479            if let Some(lte) = range.lte.as_ref().and_then(|v| v.as_str()) {
480                if s > lte {
481                    return false;
482                }
483            }
484            true
485        }
486        _ => false,
487    }
488}
489
490fn bool_match(doc: &EsDocument, b: &EsBoolQuery) -> bool {
491    if let Some(must) = &b.must {
492        if !must.iter().all(|q| InMemoryEsSync::matches_query(doc, q)) {
493            return false;
494        }
495    }
496    if let Some(must_not) = &b.must_not {
497        if must_not
498            .iter()
499            .any(|q| InMemoryEsSync::matches_query(doc, q))
500        {
501            return false;
502        }
503    }
504    if let Some(filter) = &b.filter {
505        if !filter.iter().all(|q| InMemoryEsSync::matches_query(doc, q)) {
506            return false;
507        }
508    }
509    if let Some(should) = &b.should {
510        let matched = should
511            .iter()
512            .filter(|q| InMemoryEsSync::matches_query(doc, q))
513            .count();
514        let min_match = b.minimum_should_match.unwrap_or(1);
515        if matched < min_match {
516            return false;
517        }
518    }
519    true
520}
521
522impl EsSync for InMemoryEsSync {
523    fn sync_to_es(&self, documents: Vec<EsDocument>) -> Result<EsSyncResult, EsError> {
524        let mut store = self
525            .documents
526            .write()
527            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
528        let mut indexed = 0usize;
529        let mut errors: Vec<String> = Vec::new();
530        for (i, doc) in documents.into_iter().enumerate() {
531            if doc.index.is_empty() {
532                errors.push(format!("document {} has empty index", i));
533                continue;
534            }
535            let id = Self::generate_id(&doc, i);
536            let map = store.entry(doc.index.clone()).or_default();
537            map.insert(id, doc);
538            indexed += 1;
539        }
540        if errors.is_empty() {
541            Ok(EsSyncResult::success(indexed))
542        } else {
543            Ok(EsSyncResult::with_errors(indexed, errors))
544        }
545    }
546
547    fn delete_from_es(&self, index: &str, ids: Vec<String>) -> Result<EsSyncResult, EsError> {
548        let mut store = self
549            .documents
550            .write()
551            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
552        let map = store
553            .get_mut(index)
554            .ok_or_else(|| EsError::IndexNotFound(index.to_string()))?;
555        let mut deleted = 0usize;
556        let mut errors: Vec<String> = Vec::new();
557        for id in ids {
558            if map.remove(&id).is_some() {
559                deleted += 1;
560            } else {
561                errors.push(format!("document not found: {}", id));
562            }
563        }
564        if errors.is_empty() {
565            Ok(EsSyncResult::success(deleted))
566        } else {
567            Ok(EsSyncResult::with_errors(deleted, errors))
568        }
569    }
570
571    fn search(&self, request: EsSearchRequest) -> Result<EsSearchResult, EsError> {
572        let store = self
573            .documents
574            .read()
575            .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
576        let Some(map) = store.get(&request.index) else {
577            return Err(EsError::IndexNotFound(request.index.clone()));
578        };
579
580        let start = std::time::Instant::now();
581        let mut hits: Vec<(String, EsDocument)> = map
582            .iter()
583            .filter(|(_, doc)| Self::matches_query(doc, &request.query))
584            .map(|(id, doc)| (id.clone(), doc.clone()))
585            .collect();
586
587        // Sort by fields in `request.sort`.
588        for sort in request.sort.iter().rev() {
589            let field = &sort.field;
590            let asc = matches!(sort.order, EsSortOrder::Asc);
591            hits.sort_by(|a, b| {
592                let av =
593                    a.1.source
594                        .get(field)
595                        .cloned()
596                        .unwrap_or(serde_json::Value::Null);
597                let bv =
598                    b.1.source
599                        .get(field)
600                        .cloned()
601                        .unwrap_or(serde_json::Value::Null);
602                let ord = compare_values(&av, &bv);
603                if asc {
604                    ord
605                } else {
606                    ord.reverse()
607                }
608            });
609        }
610
611        let total = hits.len();
612        let paged = hits
613            .into_iter()
614            .skip(request.from)
615            .take(request.size)
616            .map(|(id, doc)| EsHit {
617                id,
618                score: 1.0,
619                source: doc.source,
620            })
621            .collect::<Vec<_>>();
622
623        Ok(EsSearchResult {
624            total,
625            hits: paged,
626            took: start.elapsed().as_millis() as i64,
627        })
628    }
629}
630
631fn compare_values(a: &serde_json::Value, b: &serde_json::Value) -> std::cmp::Ordering {
632    use std::cmp::Ordering;
633    match (a.as_f64(), b.as_f64()) {
634        (Some(x), Some(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
635        _ => a.to_string().cmp(&b.to_string()),
636    }
637}
638
639#[derive(Debug, Clone, Serialize, Deserialize)]
640pub enum EsFieldType {
641    Text,
642    Keyword,
643    Integer,
644    Long,
645    Float,
646    Double,
647    Boolean,
648    Date,
649    Object,
650    Nested,
651    Ip,
652    GeoPoint,
653}
654
655#[derive(Debug)]
656pub enum EsError {
657    ConnectionFailed(String),
658    IndexNotFound(String),
659    DocumentNotFound(String),
660    MappingError(String),
661    QueryError(String),
662    SyncError(String),
663    /// 当前后端不支持的操作（如 `EsSyncManager::with_backend` 注入自定义后端后调用 Mock 专属方法）
664    BackendNotSupported {
665        operation: String,
666        reason: String,
667    },
668}
669
670impl std::fmt::Display for EsError {
671    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
672        match self {
673            EsError::ConnectionFailed(msg) => write!(f, "Connection failed: {}", msg),
674            EsError::IndexNotFound(idx) => write!(f, "Index not found: {}", idx),
675            EsError::DocumentNotFound(id) => write!(f, "Document not found: {}", id),
676            EsError::MappingError(msg) => write!(f, "Mapping error: {}", msg),
677            EsError::QueryError(msg) => write!(f, "Query error: {}", msg),
678            EsError::SyncError(msg) => write!(f, "Sync error: {}", msg),
679            EsError::BackendNotSupported { operation, reason } => {
680                write!(f, "Backend not supported: {} ({})", operation, reason)
681            }
682        }
683    }
684}
685
686impl std::error::Error for EsError {}
687
688impl serde::Serialize for EsError {
689    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
690    where
691        S: serde::Serializer,
692    {
693        serializer.serialize_str(&self.to_string())
694    }
695}
696
697fn current_timestamp() -> i64 {
698    use std::time::{SystemTime, UNIX_EPOCH};
699    SystemTime::now()
700        .duration_since(UNIX_EPOCH)
701        .unwrap_or_default()
702        .as_millis() as i64
703}
704
705#[cfg(test)]
706mod tests {
707    use super::*;
708
709    #[test]
710    fn test_es_document_new() {
711        let doc = EsDocument::new("test-index", serde_json::json!({"name": "test"}));
712        assert_eq!(doc.index, "test-index");
713        assert_eq!(doc.source["name"], "test");
714        assert!(doc.id.is_none());
715    }
716
717    #[test]
718    fn test_es_document_with_id() {
719        let doc = EsDocument::new("test-index", serde_json::json!({})).with_id("doc1");
720        assert_eq!(doc.id, Some("doc1".to_string()));
721    }
722
723    #[test]
724    fn test_es_search_request_new() {
725        let query = EsQuery::match_all();
726        let request = EsSearchRequest::new("test-index", query);
727        assert_eq!(request.index, "test-index");
728        assert_eq!(request.from, 0);
729        assert_eq!(request.size, 10);
730    }
731
732    #[test]
733    fn test_es_search_request_pagination() {
734        let query = EsQuery::match_all();
735        let request = EsSearchRequest::new("test", query).with_pagination(20, 50);
736        assert_eq!(request.from, 20);
737        assert_eq!(request.size, 50);
738    }
739
740    #[test]
741    fn test_es_search_request_sort() {
742        let query = EsQuery::match_all();
743        let request = EsSearchRequest::new("test", query).with_sort("date", EsSortOrder::Desc);
744        assert_eq!(request.sort.len(), 1);
745        assert_eq!(request.sort[0].field, "date");
746        assert_eq!(request.sort[0].order, EsSortOrder::Desc);
747    }
748
749    #[test]
750    fn test_es_query_match_all() {
751        let query = EsQuery::match_all();
752        assert_eq!(query, EsQuery::MatchAll);
753    }
754
755    #[test]
756    fn test_es_query_term() {
757        let query = EsQuery::term("status", serde_json::json!("active"));
758        assert_eq!(
759            query,
760            EsQuery::Term(std::collections::HashMap::from([(
761                "status".to_string(),
762                serde_json::json!("active")
763            )]))
764        );
765    }
766
767    #[test]
768    fn test_es_query_terms() {
769        let query = EsQuery::terms("tags", vec![serde_json::json!("a"), serde_json::json!("b")]);
770        assert!(matches!(query, EsQuery::Terms(_)));
771    }
772
773    #[test]
774    fn test_es_query_range() {
775        let range = EsRangeQuery::new().gte(serde_json::json!(100));
776        let query = EsQuery::range("price", range);
777        assert!(matches!(query, EsQuery::Range(_)));
778    }
779
780    #[test]
781    fn test_es_query_bool_must() {
782        let queries = vec![
783            EsQuery::term("status", serde_json::json!("active")),
784            EsQuery::term("type", serde_json::json!("post")),
785        ];
786        let query = EsQuery::must(queries);
787        assert!(matches!(query, EsQuery::Bool(_)));
788    }
789
790    #[test]
791    fn test_es_range_query() {
792        let range = EsRangeQuery::new()
793            .gte(serde_json::json!(0))
794            .lt(serde_json::json!(100));
795
796        assert!(range.gte.is_some());
797        assert!(range.lt.is_some());
798    }
799
800    #[test]
801    fn test_es_sync_result_success() {
802        let result = EsSyncResult::success(10);
803        assert_eq!(result.indexed, 10);
804        assert_eq!(result.failed, 0);
805        assert!(result.errors.is_empty());
806    }
807
808    #[test]
809    fn test_es_sync_result_with_errors() {
810        let errors = vec!["error1".to_string(), "error2".to_string()];
811        let result = EsSyncResult::with_errors(8, errors.clone());
812        assert_eq!(result.indexed, 8);
813        assert_eq!(result.failed, 2);
814        assert_eq!(result.errors, errors);
815    }
816
817    #[test]
818    fn test_es_sync_manager_new() {
819        let manager = EsSyncManager::new();
820        assert!(manager.index_mappings.is_empty());
821    }
822
823    #[test]
824    fn test_es_sync_manager_create_index() {
825        let mut manager = EsSyncManager::new();
826        let mut mapping = HashMap::new();
827        mapping.insert("title".to_string(), EsFieldType::Text);
828        mapping.insert("count".to_string(), EsFieldType::Integer);
829
830        manager.create_index("test-index", mapping);
831
832        let retrieved = manager.get_mapping("test-index");
833        assert!(retrieved.is_some());
834        assert_eq!(retrieved.unwrap().len(), 2);
835    }
836
837    #[test]
838    fn test_es_sync_manager_get_mapping_not_found() {
839        let manager = EsSyncManager::new();
840        let mapping = manager.get_mapping("nonexistent");
841        assert!(mapping.is_none());
842    }
843
844    #[test]
845    fn test_inmemory_sync_and_count() {
846        let sync = InMemoryEsSync::new();
847        let docs = vec![
848            EsDocument::new("users", serde_json::json!({"name": "alice"})).with_id("u1"),
849            EsDocument::new("users", serde_json::json!({"name": "bob"})).with_id("u2"),
850        ];
851        let result = sync.sync_to_es(docs).unwrap();
852        assert_eq!(result.indexed, 2);
853        assert_eq!(sync.count("users").unwrap(), 2);
854    }
855
856    #[test]
857    fn test_inmemory_sync_replaces_by_id() {
858        let sync = InMemoryEsSync::new();
859        sync.sync_to_es(vec![
860            EsDocument::new("idx", serde_json::json!({"v": 1})).with_id("d1")
861        ])
862        .unwrap();
863        sync.sync_to_es(vec![
864            EsDocument::new("idx", serde_json::json!({"v": 2})).with_id("d1")
865        ])
866        .unwrap();
867        assert_eq!(sync.count("idx").unwrap(), 1);
868
869        let req = EsSearchRequest::new("idx", EsQuery::match_all());
870        let res = sync.search(req).unwrap();
871        assert_eq!(res.hits[0].source["v"], 2);
872    }
873
874    #[test]
875    fn test_inmemory_sync_rejects_empty_index() {
876        let sync = InMemoryEsSync::new();
877        let result = sync
878            .sync_to_es(vec![EsDocument::new("", serde_json::json!({}))])
879            .unwrap();
880        assert_eq!(result.indexed, 0);
881        assert_eq!(result.failed, 1);
882    }
883
884    #[test]
885    fn test_inmemory_delete_real() {
886        let sync = InMemoryEsSync::new();
887        sync.sync_to_es(vec![
888            EsDocument::new("i", serde_json::json!({"k": "a"})).with_id("1"),
889            EsDocument::new("i", serde_json::json!({"k": "b"})).with_id("2"),
890        ])
891        .unwrap();
892
893        let result = sync.delete_from_es("i", vec!["1".to_string()]).unwrap();
894        assert_eq!(result.indexed, 1);
895        assert_eq!(sync.count("i").unwrap(), 1);
896
897        let err = sync.delete_from_es("missing", vec!["x".to_string()]);
898        assert!(matches!(err, Err(EsError::IndexNotFound(_))));
899    }
900
901    #[test]
902    fn test_inmemory_delete_missing_records_error() {
903        let sync = InMemoryEsSync::new();
904        sync.sync_to_es(vec![
905            EsDocument::new("i", serde_json::json!({"k": "a"})).with_id("1")
906        ])
907        .unwrap();
908        let r = sync
909            .delete_from_es("i", vec!["1".to_string(), "999".to_string()])
910            .unwrap();
911        assert_eq!(r.indexed, 1);
912        assert_eq!(r.failed, 1);
913    }
914
915    #[test]
916    fn test_inmemory_search_match_all() {
917        let sync = InMemoryEsSync::new();
918        sync.sync_to_es(vec![
919            EsDocument::new("i", serde_json::json!({"n": 1})).with_id("1"),
920            EsDocument::new("i", serde_json::json!({"n": 2})).with_id("2"),
921        ])
922        .unwrap();
923        let req = EsSearchRequest::new("i", EsQuery::match_all());
924        let res = sync.search(req).unwrap();
925        assert_eq!(res.total, 2);
926        assert_eq!(res.hits.len(), 2);
927    }
928
929    #[test]
930    fn test_inmemory_search_term_filter() {
931        let sync = InMemoryEsSync::new();
932        sync.sync_to_es(vec![
933            EsDocument::new("i", serde_json::json!({"status": "active"})).with_id("1"),
934            EsDocument::new("i", serde_json::json!({"status": "inactive"})).with_id("2"),
935        ])
936        .unwrap();
937        let req = EsSearchRequest::new("i", EsQuery::term("status", serde_json::json!("active")));
938        let res = sync.search(req).unwrap();
939        assert_eq!(res.total, 1);
940        assert_eq!(res.hits[0].source["status"], "active");
941    }
942
943    #[test]
944    fn test_inmemory_search_terms_filter() {
945        let sync = InMemoryEsSync::new();
946        sync.sync_to_es(vec![
947            EsDocument::new("i", serde_json::json!({"tag": "a"})).with_id("1"),
948            EsDocument::new("i", serde_json::json!({"tag": "b"})).with_id("2"),
949            EsDocument::new("i", serde_json::json!({"tag": "c"})).with_id("3"),
950        ])
951        .unwrap();
952        let req = EsSearchRequest::new(
953            "i",
954            EsQuery::terms("tag", vec![serde_json::json!("a"), serde_json::json!("c")]),
955        );
956        let res = sync.search(req).unwrap();
957        assert_eq!(res.total, 2);
958    }
959
960    #[test]
961    fn test_inmemory_search_range_filter() {
962        let sync = InMemoryEsSync::new();
963        sync.sync_to_es(vec![
964            EsDocument::new("i", serde_json::json!({"age": 10})).with_id("1"),
965            EsDocument::new("i", serde_json::json!({"age": 20})).with_id("2"),
966            EsDocument::new("i", serde_json::json!({"age": 30})).with_id("3"),
967        ])
968        .unwrap();
969        let req = EsSearchRequest::new(
970            "i",
971            EsQuery::range("age", EsRangeQuery::new().gte(serde_json::json!(20))),
972        );
973        let res = sync.search(req).unwrap();
974        assert_eq!(res.total, 2);
975    }
976
977    #[test]
978    fn test_inmemory_search_bool_query() {
979        let sync = InMemoryEsSync::new();
980        sync.sync_to_es(vec![
981            EsDocument::new("i", serde_json::json!({"status": "active", "n": 5})).with_id("1"),
982            EsDocument::new("i", serde_json::json!({"status": "active", "n": 50})).with_id("2"),
983            EsDocument::new("i", serde_json::json!({"status": "inactive", "n": 5})).with_id("3"),
984        ])
985        .unwrap();
986        // status=active AND n>=10
987        let bool = EsQuery::must(vec![
988            EsQuery::term("status", serde_json::json!("active")),
989            EsQuery::range("n", EsRangeQuery::new().gte(serde_json::json!(10))),
990        ]);
991        let req = EsSearchRequest::new("i", bool);
992        let res = sync.search(req).unwrap();
993        assert_eq!(res.total, 1);
994        assert_eq!(res.hits[0].source["n"], 50);
995    }
996
997    #[test]
998    fn test_inmemory_search_pagination_and_sort() {
999        let sync = InMemoryEsSync::new();
1000        let docs: Vec<EsDocument> = (1..=5)
1001            .map(|i| EsDocument::new("i", serde_json::json!({"v": i})).with_id(format!("d{}", i)))
1002            .collect();
1003        sync.sync_to_es(docs).unwrap();
1004
1005        let req = EsSearchRequest::new("i", EsQuery::match_all())
1006            .with_pagination(0, 2)
1007            .with_sort("v", EsSortOrder::Desc);
1008        let res = sync.search(req).unwrap();
1009        assert_eq!(res.total, 5);
1010        assert_eq!(res.hits.len(), 2);
1011        // Descending sort means the highest values come first.
1012        assert_eq!(res.hits[0].source["v"], 5);
1013        assert_eq!(res.hits[1].source["v"], 4);
1014
1015        // Now test pagination: skip the first, take the next two.
1016        let req2 = EsSearchRequest::new("i", EsQuery::match_all())
1017            .with_pagination(1, 2)
1018            .with_sort("v", EsSortOrder::Desc);
1019        let res2 = sync.search(req2).unwrap();
1020        assert_eq!(res2.hits.len(), 2);
1021        assert_eq!(res2.hits[0].source["v"], 4);
1022        assert_eq!(res2.hits[1].source["v"], 3);
1023
1024        // Ascending order should reverse the result.
1025        let req3 = EsSearchRequest::new("i", EsQuery::match_all())
1026            .with_pagination(0, 2)
1027            .with_sort("v", EsSortOrder::Asc);
1028        let res3 = sync.search(req3).unwrap();
1029        assert_eq!(res3.hits[0].source["v"], 1);
1030        assert_eq!(res3.hits[1].source["v"], 2);
1031    }
1032
1033    #[test]
1034    fn test_inmemory_search_missing_index_errors() {
1035        let sync = InMemoryEsSync::new();
1036        let req = EsSearchRequest::new("missing", EsQuery::match_all());
1037        let err = sync.search(req);
1038        assert!(matches!(err, Err(EsError::IndexNotFound(_))));
1039    }
1040
1041    #[test]
1042    fn test_sync_manager_delegates_to_backend() {
1043        let mut manager = EsSyncManager::new();
1044        manager.create_index("idx", HashMap::new());
1045
1046        let r = manager
1047            .sync_to_es(vec![
1048                EsDocument::new("idx", serde_json::json!({"k": "v"})).with_id("d1")
1049            ])
1050            .unwrap();
1051        assert_eq!(r.indexed, 1);
1052        assert_eq!(manager.count("idx").unwrap(), 1);
1053
1054        let req = EsSearchRequest::new("idx", EsQuery::match_all());
1055        let res = manager.search(req).unwrap();
1056        assert_eq!(res.total, 1);
1057
1058        let r = manager
1059            .delete_from_es("idx", vec!["d1".to_string()])
1060            .unwrap();
1061        assert_eq!(r.indexed, 1);
1062        assert_eq!(manager.count("idx").unwrap(), 0);
1063    }
1064
1065    #[test]
1066    fn test_es_sync_trait_object_dyn() {
1067        // Ensures the trait is object-safe and usable via dynamic dispatch.
1068        let sync: Box<dyn EsSync> = Box::new(InMemoryEsSync::new());
1069        sync.sync_to_es(vec![
1070            EsDocument::new("i", serde_json::json!({"k": 1})).with_id("1")
1071        ])
1072        .unwrap();
1073        let req = EsSearchRequest::new("i", EsQuery::match_all());
1074        let res = sync.search(req).unwrap();
1075        assert_eq!(res.total, 1);
1076    }
1077
1078    /// 验证：`with_backend` 注入自定义后端后，`count` 应返回 `BackendNotSupported`
1079    /// （因为 mock_backend 为 None）。
1080    ///
1081    /// 这确保用户在注入真实 ES 客户端时，不会被静默地回退到 Mock 行为。
1082    #[test]
1083    fn test_with_backend_count_returns_not_supported() {
1084        // 一个假后端：sync_to_es / delete_from_es / search 都返回 Ok（空结果）
1085        struct StubBackend;
1086        impl EsSync for StubBackend {
1087            fn sync_to_es(&self, _documents: Vec<EsDocument>) -> Result<EsSyncResult, EsError> {
1088                Ok(EsSyncResult::success(0))
1089            }
1090            fn delete_from_es(
1091                &self,
1092                _index: &str,
1093                _ids: Vec<String>,
1094            ) -> Result<EsSyncResult, EsError> {
1095                Ok(EsSyncResult::success(0))
1096            }
1097            fn search(&self, _request: EsSearchRequest) -> Result<EsSearchResult, EsError> {
1098                Ok(EsSearchResult {
1099                    total: 0,
1100                    hits: Vec::new(),
1101                    took: 0,
1102                })
1103            }
1104        }
1105
1106        let manager = EsSyncManager::with_backend(Box::new(StubBackend));
1107        let result = manager.count("any_index");
1108        assert!(
1109            matches!(result, Err(EsError::BackendNotSupported { .. })),
1110            "with_backend 注入自定义后端后 count 应返回 BackendNotSupported，实际: {:?}",
1111            result
1112        );
1113
1114        // 但 sync_to_es / delete / search 应正常委托给自定义后端
1115        let sync_result = manager
1116            .sync_to_es(vec![EsDocument::new("i", serde_json::json!({}))])
1117            .unwrap();
1118        assert_eq!(sync_result.indexed, 0);
1119    }
1120
1121    /// 验证：默认 `new()` 构造的 Manager 仍可使用 `count`（Mock 后端可用）
1122    #[test]
1123    fn test_default_new_count_works_with_mock() {
1124        let manager = EsSyncManager::new();
1125        // 空索引应返回 0 而非错误
1126        let count = manager.count("empty_idx").unwrap();
1127        assert_eq!(count, 0);
1128    }
1129
1130    /// 验证：`MockEsSync` 别名与 `InMemoryEsSync` 等价
1131    #[test]
1132    fn test_mock_es_sync_alias_equivalence() {
1133        let mock: MockEsSync = MockEsSync::new();
1134        let _: InMemoryEsSync = mock; // 类型应等价
1135        let mock2 = InMemoryEsSync::new();
1136        let _clone: InMemoryEsSync = mock2.clone(); // Clone 应可用
1137    }
1138}
sz_orm_es/lib.rs

sz_orm_es/
lib.rs
