sync_engine/coordinator/api.rs
1//! V1.1 API: Query and batch operations.
2//!
3//! This module contains the higher-level API methods added in V1.1:
4//! - `contains()` - Fast probabilistic existence check
5//! - `len()` / `is_empty()` - L1 cache size queries
6//! - `status()` - Detailed sync status
7//! - `get_many()` - Parallel batch fetch
8//! - `submit_many()` - Batch upsert
9//! - `delete_many()` - Batch delete
10//! - `get_or_insert_with()` - Cache-aside pattern
11
12use std::sync::atomic::Ordering;
13use tokio::task::JoinSet;
14use tracing::{debug, info, warn, error};
15
16use crate::storage::traits::StorageError;
17use crate::sync_item::SyncItem;
18use crate::merkle::MerkleBatch;
19
20use super::{SyncEngine, ItemStatus, BatchResult};
21
22impl SyncEngine {
23 // ═══════════════════════════════════════════════════════════════════════════
24 // API: Query & Batch Operations
25 // ═══════════════════════════════════════════════════════════════════════════
26
27 /// Check if an item exists across all tiers.
28 ///
29 /// Checks in order: L1 cache → Redis EXISTS → L3 Cuckoo filter → SQL query.
30 /// If found in SQL and Cuckoo filter was untrusted, updates the filter.
31 ///
32 /// # Returns
33 /// - `true` → item definitely exists in at least one tier
34 /// - `false` → item does not exist (authoritative)
35 ///
36 /// # Example
37 ///
38 /// ```rust,no_run
39 /// # use sync_engine::SyncEngine;
40 /// # async fn example(engine: &SyncEngine) {
41 /// if engine.contains("user.123").await {
42 /// let item = engine.get("user.123").await;
43 /// } else {
44 /// println!("Not found");
45 /// }
46 /// # }
47 /// ```
48 pub async fn contains(&self, id: &str) -> bool {
49 // L1: In-memory cache (definitive, sync)
50 if self.l1_cache.contains_key(id) {
51 return true;
52 }
53
54 // L2: Redis EXISTS (async, authoritative for Redis tier)
55 if let Some(ref l2) = self.l2_store {
56 if l2.exists(id).await.unwrap_or(false) {
57 return true;
58 }
59 }
60
61 // L3: Cuckoo filter check (sync, probabilistic)
62 if self.l3_filter.is_trusted() {
63 // Filter is trusted - use it for fast negative
64 if !self.l3_filter.should_check_l3(id) {
65 return false; // Definitely not in L3
66 }
67 }
68
69 // L3: SQL query (async, ground truth)
70 if let Some(ref l3) = self.l3_store {
71 if l3.exists(id).await.unwrap_or(false) {
72 // Found in SQL - update Cuckoo if it was untrusted
73 if !self.l3_filter.is_trusted() {
74 self.l3_filter.insert(id);
75 }
76 return true;
77 }
78 }
79
80 false
81 }
82
83 /// Fast sync check if item might exist (L1 + Cuckoo only).
84 ///
85 /// Use this when you need a quick probabilistic check without async.
86 /// For authoritative check, use `contains()` instead.
87 #[must_use]
88 #[inline]
89 pub fn contains_fast(&self, id: &str) -> bool {
90 self.l1_cache.contains_key(id) || self.l3_filter.should_check_l3(id)
91 }
92
93 /// Get the current count of items in L1 cache.
94 #[must_use]
95 #[inline]
96 pub fn len(&self) -> usize {
97 self.l1_cache.len()
98 }
99
100 /// Check if L1 cache is empty.
101 #[must_use]
102 #[inline]
103 pub fn is_empty(&self) -> bool {
104 self.l1_cache.is_empty()
105 }
106
107 /// Get the sync status of an item.
108 ///
109 /// Returns detailed state information about where an item exists
110 /// and its sync status across tiers.
111 ///
112 /// # Example
113 ///
114 /// ```rust,no_run
115 /// # use sync_engine::{SyncEngine, ItemStatus};
116 /// # async fn example(engine: &SyncEngine) {
117 /// match engine.status("order.456").await {
118 /// ItemStatus::Synced { in_l1, in_l2, in_l3 } => {
119 /// println!("Synced: L1={}, L2={}, L3={}", in_l1, in_l2, in_l3);
120 /// }
121 /// ItemStatus::Pending => println!("Queued for sync"),
122 /// ItemStatus::Missing => println!("Not found"),
123 /// }
124 /// # }
125 /// ```
126 pub async fn status(&self, id: &str) -> ItemStatus {
127 let in_l1 = self.l1_cache.contains_key(id);
128
129 // Check if pending in batch queue
130 let pending = self.l2_batcher.lock().await.contains(id);
131 if pending {
132 return ItemStatus::Pending;
133 }
134
135 // Check L2 (if available) - use EXISTS, no filter
136 let in_l2 = if let Some(ref l2) = self.l2_store {
137 l2.exists(id).await.unwrap_or(false)
138 } else {
139 false
140 };
141
142 // Check L3 (if available)
143 let in_l3 = if let Some(ref l3) = self.l3_store {
144 self.l3_filter.should_check_l3(id) && l3.get(id).await.ok().flatten().is_some()
145 } else {
146 false
147 };
148
149 if in_l1 || in_l2 || in_l3 {
150 ItemStatus::Synced { in_l1, in_l2, in_l3 }
151 } else {
152 ItemStatus::Missing
153 }
154 }
155
156 /// Fetch multiple items in parallel.
157 ///
158 /// Returns a vector of `Option<SyncItem>` in the same order as input IDs.
159 /// Missing items are represented as `None`.
160 ///
161 /// # Performance
162 ///
163 /// This method fetches from L1 synchronously, then batches L2/L3 lookups
164 /// for items not in L1. Much faster than sequential `get()` calls.
165 ///
166 /// # Example
167 ///
168 /// ```rust,no_run
169 /// # use sync_engine::SyncEngine;
170 /// # async fn example(engine: &SyncEngine) {
171 /// let ids = vec!["user.1", "user.2", "user.3"];
172 /// let items = engine.get_many(&ids).await;
173 /// for (id, item) in ids.iter().zip(items.iter()) {
174 /// match item {
175 /// Some(item) => println!("{}: found", id),
176 /// None => println!("{}: missing", id),
177 /// }
178 /// }
179 /// # }
180 /// ```
181 pub async fn get_many(&self, ids: &[&str]) -> Vec<Option<SyncItem>> {
182 let mut results: Vec<Option<SyncItem>> = vec![None; ids.len()];
183 let mut missing_indices: Vec<usize> = Vec::new();
184
185 // Phase 1: Check L1 (synchronous, fast)
186 for (i, id) in ids.iter().enumerate() {
187 if let Some(item) = self.l1_cache.get(*id) {
188 results[i] = Some(item.clone());
189 } else {
190 missing_indices.push(i);
191 }
192 }
193
194 // Phase 2: Fetch missing items from L2/L3 in parallel
195 if !missing_indices.is_empty() {
196 let mut join_set: JoinSet<(usize, Option<SyncItem>)> = JoinSet::new();
197
198 for &i in &missing_indices {
199 let id = ids[i].to_string();
200 let l2_store = self.l2_store.clone();
201 let l3_store = self.l3_store.clone();
202 let l3_filter = self.l3_filter.clone();
203
204 join_set.spawn(async move {
205 // Try L2 first (no filter, just try Redis)
206 if let Some(ref l2) = l2_store {
207 if let Ok(Some(item)) = l2.get(&id).await {
208 return (i, Some(item));
209 }
210 }
211
212 // Fall back to L3 (use Cuckoo filter if trusted)
213 if let Some(ref l3) = l3_store {
214 if !l3_filter.is_trusted() || l3_filter.should_check_l3(&id) {
215 if let Ok(Some(item)) = l3.get(&id).await {
216 return (i, Some(item));
217 }
218 }
219 }
220
221 (i, None)
222 });
223 }
224
225 // Collect results
226 while let Some(result) = join_set.join_next().await {
227 if let Ok((i, item)) = result {
228 results[i] = item;
229 }
230 }
231 }
232
233 results
234 }
235
236 /// Submit multiple items for sync atomically.
237 ///
238 /// All items are added to L1 and queued for batch persistence.
239 /// Returns a `BatchResult` with success/failure counts.
240 ///
241 /// # Example
242 ///
243 /// ```rust,no_run
244 /// # use sync_engine::{SyncEngine, SyncItem};
245 /// # use serde_json::json;
246 /// # async fn example(engine: &SyncEngine) {
247 /// let items = vec![
248 /// SyncItem::from_json("user.1".into(), json!({"name": "Alice"})),
249 /// SyncItem::from_json("user.2".into(), json!({"name": "Bob"})),
250 /// ];
251 /// let result = engine.submit_many(items).await.unwrap();
252 /// println!("Submitted: {}, Failed: {}", result.succeeded, result.failed);
253 /// # }
254 /// ```
255 pub async fn submit_many(&self, items: Vec<SyncItem>) -> Result<BatchResult, StorageError> {
256 if !self.should_accept_writes() {
257 return Err(StorageError::Backend(format!(
258 "Rejecting batch write: engine state={}, pressure={}",
259 self.state(),
260 self.pressure()
261 )));
262 }
263
264 let total = items.len();
265 let mut succeeded = 0;
266
267 for item in items {
268 self.insert_l1(item.clone());
269 self.l2_batcher.lock().await.add(item);
270 succeeded += 1;
271 }
272
273 debug!(total, succeeded, "Batch submitted to L1 and queue");
274
275 Ok(BatchResult {
276 total,
277 succeeded,
278 failed: total - succeeded,
279 })
280 }
281
282 /// Delete multiple items atomically.
283 ///
284 /// Removes items from all tiers (L1, L2, L3) and updates filters.
285 /// Returns a `BatchResult` with counts.
286 ///
287 /// # Example
288 ///
289 /// ```rust,no_run
290 /// # use sync_engine::SyncEngine;
291 /// # async fn example(engine: &SyncEngine) {
292 /// let ids = vec!["user.1", "user.2", "user.3"];
293 /// let result = engine.delete_many(&ids).await.unwrap();
294 /// println!("Deleted: {}", result.succeeded);
295 /// # }
296 /// ```
297 pub async fn delete_many(&self, ids: &[&str]) -> Result<BatchResult, StorageError> {
298 if !self.should_accept_writes() {
299 return Err(StorageError::Backend(format!(
300 "Rejecting batch delete: engine state={}, pressure={}",
301 self.state(),
302 self.pressure()
303 )));
304 }
305
306 let total = ids.len();
307 let mut succeeded = 0;
308
309 // Build merkle batch for all deletions
310 let mut merkle_batch = MerkleBatch::new();
311
312 for id in ids {
313 // Remove from L1
314 if let Some((_, item)) = self.l1_cache.remove(*id) {
315 let size = Self::item_size(&item);
316 self.l1_size_bytes.fetch_sub(size, Ordering::Release);
317 }
318
319 // Remove from L3 filter (no L2 filter with TTL support)
320 self.l3_filter.remove(id);
321
322 // Queue merkle deletion
323 merkle_batch.delete(id.to_string());
324
325 succeeded += 1;
326 }
327
328 // Batch delete from L2
329 if let Some(ref l2) = self.l2_store {
330 for id in ids {
331 if let Err(e) = l2.delete(id).await {
332 warn!(id, error = %e, "Failed to delete from L2");
333 }
334 }
335 }
336
337 // Batch delete from L3
338 if let Some(ref l3) = self.l3_store {
339 for id in ids {
340 if let Err(e) = l3.delete(id).await {
341 warn!(id, error = %e, "Failed to delete from L3");
342 }
343 }
344 }
345
346 // Update merkle trees
347 if let Some(ref sql_merkle) = self.sql_merkle {
348 if let Err(e) = sql_merkle.apply_batch(&merkle_batch).await {
349 error!(error = %e, "Failed to update SQL Merkle tree for batch deletion");
350 }
351 }
352
353 if let Some(ref redis_merkle) = self.redis_merkle {
354 if let Err(e) = redis_merkle.apply_batch(&merkle_batch).await {
355 warn!(error = %e, "Failed to update Redis Merkle tree for batch deletion");
356 }
357 }
358
359 info!(total, succeeded, "Batch delete completed");
360
361 Ok(BatchResult {
362 total,
363 succeeded,
364 failed: total - succeeded,
365 })
366 }
367
368 /// Get an item, or compute and insert it if missing.
369 ///
370 /// This is the classic "get or insert" pattern, useful for cache-aside:
371 /// 1. Check cache (L1 → L2 → L3)
372 /// 2. If missing, call the async factory function
373 /// 3. Insert the result and return it
374 ///
375 /// The factory is only called if the item is not found.
376 ///
377 /// # Example
378 ///
379 /// ```rust,no_run
380 /// # use sync_engine::{SyncEngine, SyncItem};
381 /// # use serde_json::json;
382 /// # async fn example(engine: &SyncEngine) {
383 /// let item = engine.get_or_insert_with("user.123", || async {
384 /// // Expensive operation - only runs if not cached
385 /// SyncItem::from_json("user.123".into(), json!({"name": "Fetched from DB"}))
386 /// }).await.unwrap();
387 /// # }
388 /// ```
389 pub async fn get_or_insert_with<F, Fut>(
390 &self,
391 id: &str,
392 factory: F,
393 ) -> Result<SyncItem, StorageError>
394 where
395 F: FnOnce() -> Fut,
396 Fut: std::future::Future<Output = SyncItem>,
397 {
398 // Try to get existing
399 if let Some(item) = self.get(id).await? {
400 return Ok(item);
401 }
402
403 // Not found - compute new value
404 let item = factory().await;
405
406 // Insert and return
407 self.submit(item.clone()).await?;
408
409 Ok(item)
410 }
411
412 // ═══════════════════════════════════════════════════════════════════════════
413 // State-based queries: Fast indexed access by caller-defined state tag
414 // ═══════════════════════════════════════════════════════════════════════════
415
416 /// Get items by state from SQL (L3 ground truth).
417 ///
418 /// Uses indexed query for fast retrieval.
419 ///
420 /// # Example
421 ///
422 /// ```rust,no_run
423 /// # use sync_engine::{SyncEngine, StorageError};
424 /// # async fn example(engine: &SyncEngine) -> Result<(), StorageError> {
425 /// // Get all delta items for CRDT merging
426 /// let deltas = engine.get_by_state("delta", 1000).await?;
427 /// for item in deltas {
428 /// println!("Delta: {}", item.object_id);
429 /// }
430 /// # Ok(())
431 /// # }
432 /// ```
433 pub async fn get_by_state(&self, state: &str, limit: usize) -> Result<Vec<SyncItem>, StorageError> {
434 if let Some(ref sql) = self.sql_store {
435 sql.get_by_state(state, limit).await
436 } else {
437 Ok(Vec::new())
438 }
439 }
440
441 /// Count items in a given state (SQL ground truth).
442 ///
443 /// # Example
444 ///
445 /// ```rust,no_run
446 /// # use sync_engine::{SyncEngine, StorageError};
447 /// # async fn example(engine: &SyncEngine) -> Result<(), StorageError> {
448 /// let pending_count = engine.count_by_state("pending").await?;
449 /// println!("{} items pending", pending_count);
450 /// # Ok(())
451 /// # }
452 /// ```
453 pub async fn count_by_state(&self, state: &str) -> Result<u64, StorageError> {
454 if let Some(ref sql) = self.sql_store {
455 sql.count_by_state(state).await
456 } else {
457 Ok(0)
458 }
459 }
460
461 /// Get just the IDs of items in a given state (lightweight query).
462 ///
463 /// Returns IDs from SQL. For Redis state SET, use `list_state_ids_redis()`.
464 pub async fn list_state_ids(&self, state: &str, limit: usize) -> Result<Vec<String>, StorageError> {
465 if let Some(ref sql) = self.sql_store {
466 sql.list_state_ids(state, limit).await
467 } else {
468 Ok(Vec::new())
469 }
470 }
471
472 /// Update the state of an item by ID.
473 ///
474 /// Updates both SQL (ground truth) and Redis state SETs.
475 /// L1 cache is NOT updated - caller should re-fetch if needed.
476 ///
477 /// Returns true if the item was found and updated.
478 pub async fn set_state(&self, id: &str, new_state: &str) -> Result<bool, StorageError> {
479 let mut updated = false;
480
481 // Update SQL (ground truth)
482 if let Some(ref sql) = self.sql_store {
483 updated = sql.set_state(id, new_state).await?;
484 }
485
486 // Note: Redis state SETs are not updated here because we'd need to know
487 // the old state to do SREM. For full Redis state management, the item
488 // should be re-submitted with the new state via submit_with().
489
490 Ok(updated)
491 }
492
493 /// Delete all items in a given state from SQL.
494 ///
495 /// Also removes from L1 cache and Redis state SET.
496 /// Returns the number of deleted items.
497 ///
498 /// # Example
499 ///
500 /// ```rust,no_run
501 /// # use sync_engine::{SyncEngine, StorageError};
502 /// # async fn example(engine: &SyncEngine) -> Result<(), StorageError> {
503 /// // Clean up all processed deltas
504 /// let deleted = engine.delete_by_state("delta").await?;
505 /// println!("Deleted {} delta items", deleted);
506 /// # Ok(())
507 /// # }
508 /// ```
509 pub async fn delete_by_state(&self, state: &str) -> Result<u64, StorageError> {
510 let mut deleted = 0u64;
511
512 // Get IDs first (for L1 cleanup)
513 let ids = if let Some(ref sql) = self.sql_store {
514 sql.list_state_ids(state, 100_000).await?
515 } else {
516 Vec::new()
517 };
518
519 // Remove from L1 cache
520 for id in &ids {
521 self.l1_cache.remove(id);
522 }
523
524 // Delete from SQL
525 if let Some(ref sql) = self.sql_store {
526 deleted = sql.delete_by_state(state).await?;
527 }
528
529 // Note: Redis items with TTL will expire naturally.
530 // For immediate Redis cleanup, call delete_by_state on RedisStore directly.
531
532 info!(state = %state, deleted = deleted, "Deleted items by state");
533
534 Ok(deleted)
535 }
536
537 // =========================================================================
538 // Prefix Scan Operations
539 // =========================================================================
540
541 /// Scan items by ID prefix.
542 ///
543 /// Retrieves all items whose ID starts with the given prefix.
544 /// Queries SQL (ground truth) directly - does NOT check L1 cache.
545 ///
546 /// Useful for CRDT delta-first architecture where deltas are stored as:
547 /// `delta:{object_id}:{op_id}` and you need to fetch all deltas for an object.
548 ///
549 /// # Example
550 ///
551 /// ```rust,no_run
552 /// # use sync_engine::{SyncEngine, StorageError};
553 /// # async fn example(engine: &SyncEngine) -> Result<(), StorageError> {
554 /// // Get base state
555 /// let base = engine.get("base:user.123").await?;
556 ///
557 /// // Get all pending deltas for this object
558 /// let deltas = engine.scan_prefix("delta:user.123:", 1000).await?;
559 ///
560 /// // Merge on-the-fly for read-repair
561 /// for delta in deltas {
562 /// println!("Delta: {} -> {:?}", delta.object_id, delta.content_as_json());
563 /// }
564 /// # Ok(())
565 /// # }
566 /// ```
567 pub async fn scan_prefix(&self, prefix: &str, limit: usize) -> Result<Vec<SyncItem>, StorageError> {
568 if let Some(ref sql) = self.sql_store {
569 sql.scan_prefix(prefix, limit).await
570 } else {
571 Ok(Vec::new())
572 }
573 }
574
575 /// Count items matching an ID prefix (SQL ground truth).
576 pub async fn count_prefix(&self, prefix: &str) -> Result<u64, StorageError> {
577 if let Some(ref sql) = self.sql_store {
578 sql.count_prefix(prefix).await
579 } else {
580 Ok(0)
581 }
582 }
583
584 /// Delete all items matching an ID prefix.
585 ///
586 /// Removes from L1 cache, SQL, and Redis.
587 /// Returns the number of deleted items.
588 ///
589 /// # Example
590 ///
591 /// ```rust,no_run
592 /// # use sync_engine::{SyncEngine, StorageError};
593 /// # async fn example(engine: &SyncEngine) -> Result<(), StorageError> {
594 /// // After merging deltas into base, clean them up
595 /// let deleted = engine.delete_prefix("delta:user.123:").await?;
596 /// println!("Cleaned up {} deltas", deleted);
597 /// # Ok(())
598 /// # }
599 /// ```
600 pub async fn delete_prefix(&self, prefix: &str) -> Result<u64, StorageError> {
601 let mut deleted = 0u64;
602
603 // Get IDs first (for L1/L2 cleanup)
604 let items = if let Some(ref sql) = self.sql_store {
605 sql.scan_prefix(prefix, 100_000).await?
606 } else {
607 Vec::new()
608 };
609
610 // Remove from L1 cache
611 for item in &items {
612 self.l1_cache.remove(&item.object_id);
613 }
614
615 // Remove from L2 (Redis) one-by-one via CacheStore trait
616 if let Some(ref l2) = self.l2_store {
617 for item in &items {
618 let _ = l2.delete(&item.object_id).await;
619 }
620 }
621
622 // Delete from SQL
623 if let Some(ref sql) = self.sql_store {
624 deleted = sql.delete_prefix(prefix).await?;
625 }
626
627 info!(prefix = %prefix, deleted = deleted, "Deleted items by prefix");
628
629 Ok(deleted)
630 }
631}
632
633#[cfg(test)]
634mod tests {
635 use super::*;
636 use crate::config::SyncEngineConfig;
637 use serde_json::json;
638 use tokio::sync::watch;
639
640 fn test_config() -> SyncEngineConfig {
641 SyncEngineConfig {
642 redis_url: None,
643 sql_url: None,
644 wal_path: None,
645 l1_max_bytes: 1024 * 1024,
646 ..Default::default()
647 }
648 }
649
650 fn test_item(id: &str) -> SyncItem {
651 SyncItem::from_json(id.to_string(), json!({"test": "data", "id": id}))
652 }
653
654 #[tokio::test]
655 async fn test_contains_l1_hit() {
656 let config = test_config();
657 let (_tx, rx) = watch::channel(config.clone());
658 let engine = SyncEngine::new(config, rx);
659
660 engine.l1_cache.insert("test.exists".into(), test_item("test.exists"));
661
662 assert!(engine.contains("test.exists").await);
663 }
664
665 #[tokio::test]
666 async fn test_contains_with_trusted_filter() {
667 let config = test_config();
668 let (_tx, rx) = watch::channel(config.clone());
669 let engine = SyncEngine::new(config, rx);
670
671 engine.l3_filter.mark_trusted();
672
673 engine.l1_cache.insert("test.exists".into(), test_item("test.exists"));
674 engine.l3_filter.insert("test.exists");
675
676 assert!(engine.contains("test.exists").await);
677 assert!(!engine.contains("test.missing").await);
678 }
679
680 #[test]
681 fn test_len_and_is_empty() {
682 let config = test_config();
683 let (_tx, rx) = watch::channel(config.clone());
684 let engine = SyncEngine::new(config, rx);
685
686 assert!(engine.is_empty());
687 assert_eq!(engine.len(), 0);
688
689 engine.l1_cache.insert("a".into(), test_item("a"));
690 assert!(!engine.is_empty());
691 assert_eq!(engine.len(), 1);
692
693 engine.l1_cache.insert("b".into(), test_item("b"));
694 assert_eq!(engine.len(), 2);
695 }
696
697 #[tokio::test]
698 async fn test_status_synced_in_l1() {
699 use super::super::EngineState;
700
701 let config = test_config();
702 let (_tx, rx) = watch::channel(config.clone());
703 let engine = SyncEngine::new(config, rx);
704 let _ = engine.state.send(EngineState::Ready);
705
706 engine.submit(test_item("test.item")).await.expect("Submit failed");
707 let _ = engine.l2_batcher.lock().await.force_flush();
708
709 let status = engine.status("test.item").await;
710 assert!(matches!(status, ItemStatus::Synced { in_l1: true, .. }));
711 }
712
713 #[tokio::test]
714 async fn test_status_pending() {
715 use super::super::EngineState;
716
717 let config = test_config();
718 let (_tx, rx) = watch::channel(config.clone());
719 let engine = SyncEngine::new(config, rx);
720 let _ = engine.state.send(EngineState::Ready);
721
722 engine.submit(test_item("test.pending")).await.expect("Submit failed");
723
724 let status = engine.status("test.pending").await;
725 assert_eq!(status, ItemStatus::Pending);
726 }
727
728 #[tokio::test]
729 async fn test_status_missing() {
730 let config = test_config();
731 let (_tx, rx) = watch::channel(config.clone());
732 let engine = SyncEngine::new(config, rx);
733
734 let status = engine.status("test.nonexistent").await;
735 assert_eq!(status, ItemStatus::Missing);
736 }
737
738 #[tokio::test]
739 async fn test_get_many_from_l1() {
740 use super::super::EngineState;
741
742 let config = test_config();
743 let (_tx, rx) = watch::channel(config.clone());
744 let engine = SyncEngine::new(config, rx);
745 let _ = engine.state.send(EngineState::Ready);
746
747 engine.l1_cache.insert("a".into(), test_item("a"));
748 engine.l1_cache.insert("b".into(), test_item("b"));
749 engine.l1_cache.insert("c".into(), test_item("c"));
750
751 let results = engine.get_many(&["a", "b", "missing", "c"]).await;
752
753 assert_eq!(results.len(), 4);
754 assert!(results[0].is_some());
755 assert!(results[1].is_some());
756 assert!(results[2].is_none());
757 assert!(results[3].is_some());
758
759 assert_eq!(results[0].as_ref().unwrap().object_id, "a");
760 assert_eq!(results[1].as_ref().unwrap().object_id, "b");
761 assert_eq!(results[3].as_ref().unwrap().object_id, "c");
762 }
763
764 #[tokio::test]
765 async fn test_submit_many() {
766 use super::super::EngineState;
767
768 let config = test_config();
769 let (_tx, rx) = watch::channel(config.clone());
770 let engine = SyncEngine::new(config, rx);
771 let _ = engine.state.send(EngineState::Ready);
772
773 let items = vec![
774 test_item("batch.1"),
775 test_item("batch.2"),
776 test_item("batch.3"),
777 ];
778
779 let result = engine.submit_many(items).await.expect("Batch submit failed");
780
781 assert_eq!(result.total, 3);
782 assert_eq!(result.succeeded, 3);
783 assert_eq!(result.failed, 0);
784 assert!(result.is_success());
785
786 assert_eq!(engine.len(), 3);
787 assert!(engine.contains("batch.1").await);
788 assert!(engine.contains("batch.2").await);
789 assert!(engine.contains("batch.3").await);
790 }
791
792 #[tokio::test]
793 async fn test_delete_many() {
794 use super::super::EngineState;
795
796 let config = test_config();
797 let (_tx, rx) = watch::channel(config.clone());
798 let engine = SyncEngine::new(config, rx);
799 let _ = engine.state.send(EngineState::Ready);
800
801 engine.l1_cache.insert("del.1".into(), test_item("del.1"));
802 engine.l1_cache.insert("del.2".into(), test_item("del.2"));
803 engine.l1_cache.insert("keep".into(), test_item("keep"));
804
805 let result = engine.delete_many(&["del.1", "del.2"]).await.expect("Batch delete failed");
806
807 assert_eq!(result.total, 2);
808 assert_eq!(result.succeeded, 2);
809 assert!(result.is_success());
810
811 assert!(!engine.l1_cache.contains_key("del.1"));
812 assert!(!engine.l1_cache.contains_key("del.2"));
813 assert!(engine.l1_cache.contains_key("keep"));
814 }
815
816 #[tokio::test]
817 async fn test_get_or_insert_with_existing() {
818 use super::super::EngineState;
819
820 let config = test_config();
821 let (_tx, rx) = watch::channel(config.clone());
822 let engine = SyncEngine::new(config, rx);
823 let _ = engine.state.send(EngineState::Ready);
824
825 let existing = test_item("existing");
826 engine.l1_cache.insert("existing".into(), existing.clone());
827
828 let factory_called = std::sync::atomic::AtomicBool::new(false);
829 let result = engine.get_or_insert_with("existing", || {
830 factory_called.store(true, std::sync::atomic::Ordering::SeqCst);
831 async { test_item("should_not_be_used") }
832 }).await.expect("get_or_insert_with failed");
833
834 assert!(!factory_called.load(std::sync::atomic::Ordering::SeqCst));
835 assert_eq!(result.object_id, "existing");
836 }
837
838 #[tokio::test]
839 async fn test_get_or_insert_with_missing() {
840 use super::super::EngineState;
841
842 let config = test_config();
843 let (_tx, rx) = watch::channel(config.clone());
844 let engine = SyncEngine::new(config, rx);
845 let _ = engine.state.send(EngineState::Ready);
846
847 let result = engine.get_or_insert_with("new_item", || async {
848 SyncItem::from_json("new_item".into(), json!({"created": "by factory"}))
849 }).await.expect("get_or_insert_with failed");
850
851 assert_eq!(result.object_id, "new_item");
852 assert!(engine.contains("new_item").await);
853 }
854}