sz_orm_es/
lib.rs

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

sz_orm_es/
lib.rs
