1use serde::{Deserialize, Serialize};
13use wm_core::{CoreError, Galaxy, Result};
14
15use crate::memory::Memory;
16use crate::search::{SearchEngine, printable_ratio, sanitize_content_for_index};
17use crate::store::MemoryStore;
18
19pub const REINDEX_BATCH: usize = 2_000;
22
23#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
25pub struct GalaxyRebuildStats {
26 pub galaxy: String,
28 pub scanned: usize,
30 pub indexed: usize,
32 pub skipped: usize,
34 pub deleted: usize,
38}
39
40#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
42pub struct IndexRebuildReport {
43 pub scanned: usize,
45 pub indexed: usize,
47 pub skipped: usize,
49 pub deleted: usize,
51 pub galaxies: Vec<GalaxyRebuildStats>,
53}
54
55#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
57pub struct GalaxyConsistency {
58 pub galaxy: String,
60 pub lmdb_count: usize,
62 pub tantivy_count: usize,
64 pub drift: bool,
66}
67
68#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
70pub struct ConsistencyReport {
71 pub galaxies: Vec<GalaxyConsistency>,
73 pub total_lmdb: usize,
75 pub total_tantivy: usize,
77 pub has_drift: bool,
79}
80
81#[must_use]
93pub fn check_consistency(store: &MemoryStore, search: &SearchEngine) -> ConsistencyReport {
94 let mut report = ConsistencyReport::default();
95 for galaxy in Galaxy::memory_galaxies() {
96 let lmdb_count = store.count(galaxy).unwrap_or(0);
97 let tantivy_count = search.count_docs_in_galaxy(galaxy.db_name()).unwrap_or(0);
98 let drift = lmdb_count != tantivy_count;
99 report.total_lmdb += lmdb_count;
100 report.total_tantivy += tantivy_count;
101 if drift {
102 report.has_drift = true;
103 }
104 report.galaxies.push(GalaxyConsistency {
105 galaxy: galaxy.db_name().to_string(),
106 lmdb_count,
107 tantivy_count,
108 drift,
109 });
110 }
111 report
112}
113
114#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
124pub struct GalaxyDriftClass {
125 pub galaxy: String,
127 pub lmdb_count: usize,
129 pub tantivy_count: usize,
131 pub skip_reserve: usize,
135 pub healable_gap: i64,
139}
140
141#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
143pub struct DriftClassification {
144 pub galaxies: Vec<GalaxyDriftClass>,
146 pub skip_reserve_total: usize,
148 pub healable_total: usize,
150}
151
152#[must_use]
160pub fn classify_drift(store: &MemoryStore, search: &SearchEngine) -> DriftClassification {
161 let mut out = DriftClassification::default();
162 for galaxy in Galaxy::memory_galaxies() {
163 let lmdb_count = store.count(galaxy).unwrap_or(0);
164 let tantivy_count = search.count_docs_in_galaxy(galaxy.db_name()).unwrap_or(0);
165 let mut skip_reserve = 0usize;
166 if lmdb_count > tantivy_count {
167 for mem in store.scan(galaxy, lmdb_count).unwrap_or_default() {
168 if sanitize_content_for_index(&mem.content).is_none() {
169 skip_reserve += 1;
170 }
171 }
172 }
173 let indexable = usize::try_into(lmdb_count - skip_reserve).unwrap_or(i64::MAX);
174 let indexed = usize::try_into(tantivy_count).unwrap_or(i64::MAX);
175 let healable_gap = indexable - indexed;
176 out.skip_reserve_total += skip_reserve;
177 out.healable_total += healable_gap.unsigned_abs() as usize;
178 out.galaxies.push(GalaxyDriftClass {
179 galaxy: galaxy.db_name().to_string(),
180 lmdb_count,
181 tantivy_count,
182 skip_reserve,
183 healable_gap,
184 });
185 }
186 out
187}
188
189pub fn rebuild_index(
200 store: &MemoryStore,
201 search: &SearchEngine,
202 galaxy_filter: &[String],
203) -> Result<IndexRebuildReport> {
204 let mut snapshots = Vec::new();
208 for galaxy in Galaxy::memory_galaxies() {
209 if galaxy_filter.is_empty() || galaxy_filter.iter().any(|g| g == galaxy.db_name()) {
210 snapshots.push((galaxy, store.scan_all_strict(galaxy)?));
211 }
212 }
213 if galaxy_filter.iter().any(|name| {
214 !Galaxy::memory_galaxies()
215 .iter()
216 .any(|g| g.db_name() == name)
217 }) {
218 return Err(CoreError::Memory(
219 "reindex filter must name a memory galaxy".into(),
220 ));
221 }
222 let mut report = IndexRebuildReport::default();
223 {
224 let mut writer = search.writer()?;
225 if galaxy_filter.is_empty() {
226 writer
227 .as_mut()
228 .ok_or_else(|| {
229 CoreError::Memory("Tantivy writer unavailable: index opened read-only".into())
230 })?
231 .delete_all_documents()
232 .map_err(|e| CoreError::Memory(format!("Tantivy delete_all_documents: {e}")))?;
233 } else {
234 for galaxy in Galaxy::all() {
238 if galaxy_filter.iter().any(|g| g == galaxy.db_name()) {
239 search.delete_by_galaxy(&mut writer, galaxy.db_name())?;
240 }
241 }
242 }
243
244 for (galaxy, memories) in snapshots {
245 let mut stats = GalaxyRebuildStats {
246 galaxy: galaxy.db_name().to_string(),
247 ..GalaxyRebuildStats::default()
248 };
249 for mem in &memories {
250 stats.scanned += 1;
251 if index_memory(search, &mut writer, galaxy, mem)?.is_some() {
252 stats.indexed += 1;
253 } else {
254 stats.skipped += 1;
255 }
256 }
257 report.scanned += stats.scanned;
258 report.indexed += stats.indexed;
259 report.skipped += stats.skipped;
260 report.galaxies.push(stats);
261 }
262
263 search.commit(&mut writer)?;
264 drop(writer);
265 }
266 Ok(report)
267}
268
269fn index_memory(
272 search: &SearchEngine,
273 writer: &mut Option<tantivy::IndexWriter>,
274 galaxy: Galaxy,
275 mem: &Memory,
276) -> Result<Option<()>> {
277 let Some(content) = sanitize_content_for_index(&mem.content) else {
278 return Ok(None);
279 };
280 let timestamp = mem.metadata.created_at.timestamp();
281 let id = mem.metadata.id.to_string();
282 search.add_document(
283 writer,
284 &id,
285 galaxy.db_name(),
286 &content,
287 &mem.metadata.tags,
288 timestamp,
289 )?;
290 Ok(Some(()))
291}
292
293pub fn heal_index_drift(
319 store: &MemoryStore,
320 search: &SearchEngine,
321) -> Result<Option<IndexRebuildReport>> {
322 let class = classify_drift(store, search);
327 let drifted: Vec<String> = class
328 .galaxies
329 .iter()
330 .filter(|g| g.healable_gap != 0)
331 .map(|g| g.galaxy.clone())
332 .collect();
333 if drifted.is_empty() {
334 return Ok(None);
335 }
336
337 let mut report = IndexRebuildReport::default();
338 let mut writer = search.writer()?;
339 for name in &drifted {
340 let Some(galaxy) = Galaxy::from_db_name(name) else {
341 return Err(CoreError::Memory(format!(
342 "drift classification named a non-memory galaxy: {name}"
343 )));
344 };
345 let mut stats = GalaxyRebuildStats {
346 galaxy: name.clone(),
347 ..GalaxyRebuildStats::default()
348 };
349
350 let indexed_ids = search.indexed_ids_in_galaxy(name)?;
351 let mut lmdb_ids: std::collections::HashSet<String> =
352 std::collections::HashSet::with_capacity(indexed_ids.len());
353 for mem in store.scan_all(galaxy)? {
354 let id = mem.metadata.id.to_string();
355 lmdb_ids.insert(id.clone());
356 if !indexed_ids.contains(&id) {
357 stats.scanned += 1;
358 if index_memory(search, &mut writer, galaxy, &mem)?.is_some() {
359 stats.indexed += 1;
360 } else {
361 stats.skipped += 1;
366 }
367 }
368 }
369 for id in &indexed_ids {
372 if !lmdb_ids.contains(id) {
373 search.delete_document(&mut writer, id)?;
374 stats.deleted += 1;
375 }
376 }
377
378 report.scanned += stats.scanned;
379 report.indexed += stats.indexed;
380 report.skipped += stats.skipped;
381 report.deleted += stats.deleted;
382 report.galaxies.push(stats);
383 }
384
385 search.commit(&mut writer)?;
386 drop(writer);
387 Ok(Some(report))
388}
389
390#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
394pub struct GalaxyContentRepairStats {
395 pub galaxy: String,
397 pub scanned: usize,
399 pub repaired: usize,
401 pub unrepairable: usize,
404 pub already_clean: usize,
406}
407
408#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
410pub struct ContentRepairReport {
411 pub scanned: usize,
413 pub repaired: usize,
415 pub unrepairable: usize,
417 pub already_clean: usize,
419 pub galaxies: Vec<GalaxyContentRepairStats>,
421}
422
423fn clean_for_repair(content: &str) -> String {
427 content
428 .chars()
429 .map(|c| {
430 if c.is_control() && c != '\n' && c != '\t' && c != '\r' {
431 ' '
432 } else {
433 c
434 }
435 })
436 .collect()
437}
438
439pub fn repair_content(
463 store: &MemoryStore,
464 search: &SearchEngine,
465 galaxies: &[Galaxy],
466) -> Result<ContentRepairReport> {
467 let mut report = ContentRepairReport::default();
468 let mut writer = search.writer()?;
469 for galaxy in galaxies {
470 let mut stats = GalaxyContentRepairStats {
471 galaxy: galaxy.db_name().to_string(),
472 ..Default::default()
473 };
474 for mem in store.scan_all(*galaxy)? {
475 stats.scanned += 1;
476 if sanitize_content_for_index(&mem.content).is_some() {
477 stats.already_clean += 1;
478 continue;
479 }
480 let total = mem.content.chars().count();
484 let cleaned = clean_for_repair(&mem.content);
485 if total == 0
486 || printable_ratio(&mem.content) < 0.5
487 || sanitize_content_for_index(&cleaned).is_none()
488 {
489 stats.unrepairable += 1;
490 continue;
491 }
492 let mut repaired_mem = mem;
493 let old_hash = repaired_mem.metadata.content_hash.clone();
494 repaired_mem.content = cleaned;
495 repaired_mem.metadata.content_hash = crate::content_hash(&repaired_mem.content);
496 repaired_mem.metadata.revision_count =
497 repaired_mem.metadata.revision_count.saturating_add(1);
498 store.put(*galaxy, &repaired_mem)?;
499 store.record_revision(
503 *galaxy,
504 repaired_mem.metadata.id,
505 &old_hash,
506 &repaired_mem.metadata.content_hash,
507 crate::revision::RevisionActor {
508 session: None,
509 user: Some("wm-repair-content".to_string()),
510 compartment: None,
511 },
512 )?;
513 let id_str = repaired_mem.metadata.id.to_string();
514 search.delete_document(&mut writer, &id_str)?;
517 search.add_document(
518 &mut writer,
519 &id_str,
520 galaxy.db_name(),
521 &repaired_mem.content,
522 &repaired_mem.metadata.tags,
523 repaired_mem.metadata.created_at.timestamp(),
524 )?;
525 stats.repaired += 1;
526 }
527 report.scanned += stats.scanned;
528 report.repaired += stats.repaired;
529 report.unrepairable += stats.unrepairable;
530 report.already_clean += stats.already_clean;
531 report.galaxies.push(stats);
532 }
533 search.commit(&mut writer)?;
534 Ok(report)
535}
536
537#[must_use]
540pub fn tantivy_path_for(store_path: &std::path::Path) -> std::path::PathBuf {
541 store_path.join("tantivy")
542}
543
544#[must_use]
546pub fn missing_index_error(store_path: &std::path::Path) -> CoreError {
547 CoreError::Memory(format!(
548 "Tantivy index not found at {} — run 'wm serve' once to create it",
549 tantivy_path_for(store_path).display()
550 ))
551}
552
553pub fn quarantine_and_recreate(tantivy_path: &std::path::Path) -> Result<std::path::PathBuf> {
564 let quarantine = quarantine_index(tantivy_path)?;
565 std::fs::create_dir_all(tantivy_path).map_err(|e| {
566 CoreError::Memory(format!(
567 "Index quarantine — recreate {}: {e} (the old index is at {})",
568 tantivy_path.display(),
569 quarantine.display()
570 ))
571 })?;
572 Ok(quarantine)
573}
574
575pub fn quarantine_index(tantivy_path: &std::path::Path) -> Result<std::path::PathBuf> {
581 let ts = std::time::SystemTime::now()
582 .duration_since(std::time::UNIX_EPOCH)
583 .map_or(0, |d| d.as_millis());
584 let file_name = tantivy_path
585 .file_name()
586 .and_then(|n| n.to_str())
587 .unwrap_or("tantivy");
588 let quarantine = tantivy_path.with_file_name(format!("{file_name}.corrupt.{ts}"));
589 std::fs::rename(tantivy_path, &quarantine).map_err(|e| {
590 CoreError::Memory(format!(
591 "Index quarantine — rename {} to {}: {e}",
592 tantivy_path.display(),
593 quarantine.display()
594 ))
595 })?;
596 Ok(quarantine)
597}
598
599pub fn open_or_quarantine(
613 tantivy_path: &std::path::Path,
614) -> Result<(SearchEngine, Option<std::path::PathBuf>)> {
615 match SearchEngine::open(tantivy_path) {
616 Ok(engine) => Ok((engine, None)),
617 Err(error) => {
618 if error.to_string().contains("LockBusy") {
619 return Err(error);
620 }
621 let quarantine = quarantine_and_recreate(tantivy_path)?;
622 let engine = SearchEngine::open(tantivy_path).map_err(|e| {
623 CoreError::Memory(format!(
624 "Index at {} could not be opened ({error}); the old index was moved to {} \
625 but creating a fresh index failed: {e}",
626 tantivy_path.display(),
627 quarantine.display()
628 ))
629 })?;
630 Ok((engine, Some(quarantine)))
631 }
632 }
633}
634
635#[derive(Debug, Default, Clone, Serialize, Deserialize)]
637pub struct DrainReport {
638 pub pending: usize,
640 pub drained: usize,
642 pub missing: usize,
644 pub unknown_galaxy: usize,
646 pub failed: usize,
648}
649
650#[allow(clippy::significant_drop_tightening)] pub fn drain_index_pending(store: &MemoryStore, search: &SearchEngine) -> Result<DrainReport> {
664 let entries = store.index_pending_entries()?;
665 let mut report = DrainReport {
666 pending: entries.len(),
667 ..Default::default()
668 };
669 if entries.is_empty() {
670 return Ok(report);
671 }
672 let mut writer = search.writer()?;
673 let mut cleared: Vec<String> = Vec::new();
674 for (id, galaxy_name, _) in &entries {
675 let Some(galaxy) = Galaxy::from_db_name(galaxy_name) else {
676 report.unknown_galaxy += 1;
677 cleared.push(id.clone());
678 continue;
679 };
680 let Ok(parsed) = uuid::Uuid::parse_str(id) else {
681 report.unknown_galaxy += 1;
682 cleared.push(id.clone());
683 continue;
684 };
685 let Some(memory) = store.get(galaxy, parsed)? else {
686 report.missing += 1;
687 cleared.push(id.clone());
688 continue;
689 };
690 let indexed = search.delete_document(&mut writer, id).and_then(|()| {
691 search.add_document(
692 &mut writer,
693 id,
694 galaxy.db_name(),
695 &memory.content,
696 &memory.metadata.tags,
697 memory.metadata.created_at.timestamp(),
698 )
699 });
700 match indexed {
701 Ok(()) => {
702 report.drained += 1;
703 cleared.push(id.clone());
704 }
705 Err(e) => {
706 tracing::warn!(
707 memory_id = %id,
708 error = %e,
709 "pending-index drain could not re-index a memory"
710 );
711 report.failed += 1;
712 }
713 }
714 }
715 search.commit(&mut writer)?;
718 store.clear_index_pending(&cleared)?;
719 Ok(report)
720}
721
722#[cfg(test)]
723mod tests {
724 use super::*;
725 use crate::Memory;
726 use tempfile::tempdir;
727
728 fn setup() -> (tempfile::TempDir, MemoryStore, SearchEngine) {
729 let tmp = tempdir().unwrap();
730 let store = MemoryStore::open_default(tmp.path()).unwrap();
731 let tantivy_dir = tmp.path().join("tantivy");
732 std::fs::create_dir_all(&tantivy_dir).unwrap();
733 let search = SearchEngine::open(&tantivy_dir).unwrap();
734 (tmp, store, search)
735 }
736
737 fn put_and_index(store: &MemoryStore, search: &SearchEngine, galaxy: Galaxy, content: &str) {
738 let mem = Memory::new(galaxy, content.to_string());
739 let id = mem.metadata.id;
740 store.put(galaxy, &mem).unwrap();
741 let mut writer = search.writer().unwrap();
742 search
743 .add_document(
744 &mut writer,
745 &id.to_string(),
746 galaxy.db_name(),
747 content,
748 &mem.metadata.tags,
749 mem.metadata.created_at.timestamp(),
750 )
751 .unwrap();
752 search.commit(&mut writer).unwrap();
753 }
754
755 #[test]
760 fn pending_index_ledger_drains_and_clears() {
761 let tmp = tempdir().unwrap();
762 let store = MemoryStore::open_default(tmp.path()).unwrap();
763 let tantivy_dir = tmp.path().join("tantivy");
764 std::fs::create_dir_all(&tantivy_dir).unwrap();
765 let search = SearchEngine::open(&tantivy_dir).unwrap();
766
767 let mem = Memory::new(Galaxy::Codex, "lock-lost write".into());
768 let id = mem.metadata.id;
769 store.put(Galaxy::Codex, &mem).unwrap();
770 store
771 .mark_index_pending("codex", &id.to_string(), 1)
772 .unwrap();
773 let ghost = uuid::Uuid::new_v4();
775 store
776 .mark_index_pending("codex", &ghost.to_string(), 2)
777 .unwrap();
778 assert_eq!(store.count_index_pending().unwrap(), 2);
779
780 let report = drain_index_pending(&store, &search).unwrap();
781 assert_eq!(report.pending, 2);
782 assert_eq!(report.drained, 1);
783 assert_eq!(report.missing, 1);
784 assert_eq!(report.failed, 0);
785 assert_eq!(store.count_index_pending().unwrap(), 0);
786 assert!(
787 !search.search("lock-lost write", 5).unwrap().is_empty(),
788 "drained memory must be searchable"
789 );
790
791 assert_eq!(drain_index_pending(&store, &search).unwrap().pending, 0);
793
794 let survivor = uuid::Uuid::new_v4();
796 store
797 .mark_index_pending("codex", &survivor.to_string(), 3)
798 .unwrap();
799 let readonly = SearchEngine::open_readonly(&tantivy_dir).unwrap();
800 assert!(
801 drain_index_pending(&store, &readonly).is_err(),
802 "drain must fail when the writer lock is unavailable"
803 );
804 assert_eq!(
805 store.count_index_pending().unwrap(),
806 1,
807 "entries must survive a lock-loss drain"
808 );
809 }
810
811 #[test]
812 fn rebuild_repopulates_index_from_lmdb() {
813 let (_tmp, store, search) = setup();
814 put_and_index(&store, &search, Galaxy::Codex, "rust memory one");
815 put_and_index(&store, &search, Galaxy::Codex, "python memory two");
816 put_and_index(&store, &search, Galaxy::Research, "research notes");
817
818 {
821 let mut writer = search.writer().unwrap();
822 search
823 .add_document(
824 &mut writer,
825 "99999999-9999-9999-9999-999999999999",
826 "codex",
827 "stale ghost document",
828 &[],
829 1000,
830 )
831 .unwrap();
832 search.commit(&mut writer).unwrap();
833 }
834 let ghost = search.search("ghost", 10).unwrap();
835 assert_eq!(ghost.len(), 1);
836
837 let report = rebuild_index(&store, &search, &[]).unwrap();
838 assert_eq!(report.indexed, 3);
839 assert_eq!(report.scanned, 3);
840 assert_eq!(report.galaxies.len(), Galaxy::memory_galaxies().len());
841
842 let ghost = search.search("ghost", 10).unwrap();
843 assert!(ghost.is_empty(), "stale index entry must be purged");
844
845 let rust = search.search("rust memory one", 10).unwrap();
846 assert_eq!(rust.len(), 1);
847 assert_eq!(rust[0].content, "rust memory one");
848 }
849
850 #[test]
851 fn rebuild_refuses_undecodable_sources_before_touching_index() {
852 let (_tmp, store, search) = setup();
853 put_and_index(
854 &store,
855 &search,
856 Galaxy::Codex,
857 "preserved searchable evidence",
858 );
859 let id = uuid::Uuid::new_v4();
860 store
861 .put_raw(Galaxy::Sessions, id.as_bytes(), b"invalid messagepack")
862 .unwrap();
863 let before = search.count_docs_in_galaxy("codex").unwrap();
864 assert!(rebuild_index(&store, &search, &[]).is_err());
865 let mut writer = search.writer().unwrap();
867 search.commit(&mut writer).unwrap();
868 assert_eq!(search.count_docs_in_galaxy("codex").unwrap(), before);
869 assert_eq!(search.search("preserved", 10).unwrap().len(), 1);
870 }
871
872 #[test]
873 fn rebuild_skips_binary_garbage() {
874 let (_tmp, store, search) = setup();
875 put_and_index(&store, &search, Galaxy::Codex, "clean text entry");
876 let mem = Memory::new(Galaxy::Codex, "\u{00}\u{01}\u{02}raw bytes".to_string());
877 store.put(Galaxy::Codex, &mem).unwrap();
878
879 let report = rebuild_index(&store, &search, &[]).unwrap();
880 assert_eq!(report.indexed, 1, "garbage content must be skipped");
881 assert_eq!(report.skipped, 1);
882
883 let results = search.search("raw", 10).unwrap();
884 assert!(results.is_empty());
885 }
886
887 #[test]
888 fn rebuild_respects_galaxy_filter() {
889 let (_tmp, store, search) = setup();
890 put_and_index(&store, &search, Galaxy::Codex, "codex memory");
891 put_and_index(&store, &search, Galaxy::Research, "research memory");
892
893 let report = rebuild_index(&store, &search, &["codex".to_string()]).unwrap();
894 assert_eq!(report.indexed, 1);
895 assert_eq!(report.galaxies.len(), 1);
896 assert_eq!(report.galaxies[0].galaxy, "codex");
897
898 let codex = search
903 .search_in_galaxy("codex memory", Some(Galaxy::Codex), 10)
904 .unwrap();
905 assert_eq!(codex.len(), 1);
906 let research = search
907 .search_in_galaxy("research memory", Some(Galaxy::Research), 10)
908 .unwrap();
909 assert_eq!(
910 research.len(),
911 1,
912 "filtered rebuild must preserve documents in unselected galaxies"
913 );
914 }
915
916 #[test]
917 fn consistency_check_no_drift_when_indexed() {
918 let (_tmp, store, search) = setup();
919 put_and_index(&store, &search, Galaxy::Codex, "hello world");
920 put_and_index(&store, &search, Galaxy::Codex, "another memory");
921
922 let report = check_consistency(&store, &search);
923 assert!(!report.has_drift, "no drift expected when all indexed");
924 let codex = report
925 .galaxies
926 .iter()
927 .find(|g| g.galaxy == "codex")
928 .unwrap();
929 assert_eq!(codex.lmdb_count, 2);
930 assert_eq!(codex.tantivy_count, 2);
931 }
932
933 #[test]
934 fn consistency_check_detects_drift() {
935 let (_tmp, store, search) = setup();
936 let mem = Memory::new(Galaxy::Codex, "unindexed".to_string());
938 store.put(Galaxy::Codex, &mem).unwrap();
939
940 let report = check_consistency(&store, &search);
941 assert!(
942 report.has_drift,
943 "drift expected when LMDB has unindexed memory"
944 );
945 let codex = report
946 .galaxies
947 .iter()
948 .find(|g| g.galaxy == "codex")
949 .unwrap();
950 assert_eq!(codex.lmdb_count, 1);
951 assert_eq!(codex.tantivy_count, 0);
952 }
953
954 #[test]
955 fn heal_repairs_only_drifted_galaxies() {
956 let (_tmp, store, search) = setup();
957
958 store
960 .put(
961 Galaxy::Sessions,
962 &Memory::new(Galaxy::Sessions, "session needle".into()),
963 )
964 .unwrap();
965 store
966 .put(
967 Galaxy::Research,
968 &Memory::new(Galaxy::Research, "research needle".into()),
969 )
970 .unwrap();
971 put_and_index(&store, &search, Galaxy::Codex, "healthy codex entry");
973
974 let report = heal_index_drift(&store, &search)
975 .unwrap()
976 .expect("drift expected before heal");
977 let healed: Vec<_> = report.galaxies.iter().map(|g| g.galaxy.as_str()).collect();
978 assert!(healed.contains(&"sessions"));
979 assert!(healed.contains(&"research"));
980 assert!(
981 !healed.contains(&"codex"),
982 "healthy galaxy must be untouched"
983 );
984 assert_eq!(report.indexed, 2);
985
986 assert!(
987 heal_index_drift(&store, &search).unwrap().is_none(),
988 "second heal must be a no-op once consistent"
989 );
990 assert_eq!(
991 search
992 .search_in_galaxy("session needle", Some(Galaxy::Sessions), 10)
993 .unwrap()
994 .len(),
995 1
996 );
997 assert_eq!(
998 search
999 .search_in_galaxy("healthy codex", Some(Galaxy::Codex), 10)
1000 .unwrap()
1001 .len(),
1002 1
1003 );
1004 }
1005
1006 #[test]
1007 fn heal_noop_when_consistent() {
1008 let (_tmp, store, search) = setup();
1009 put_and_index(&store, &search, Galaxy::Codex, "indexed entry");
1010 put_and_index(&store, &search, Galaxy::Dreams, "dream entry");
1011
1012 assert!(heal_index_drift(&store, &search).unwrap().is_none());
1013 }
1014
1015 #[test]
1016 fn index_health_tracks_successes() {
1017 let (_tmp, store, search) = setup();
1018 put_and_index(&store, &search, Galaxy::Codex, "test content");
1019
1020 let health = search.health().snapshot();
1021 let successes = health
1022 .get("successes")
1023 .and_then(serde_json::Value::as_u64)
1024 .unwrap_or(0);
1025 assert!(successes > 0, "expected at least one success");
1026 let failures = health
1027 .get("failures")
1028 .and_then(serde_json::Value::as_u64)
1029 .unwrap_or(0);
1030 assert_eq!(failures, 0);
1031 assert_eq!(
1032 health.get("degraded").and_then(serde_json::Value::as_bool),
1033 Some(false)
1034 );
1035 }
1036
1037 #[test]
1038 fn consistency_check_ignores_non_memory_galaxies() {
1039 let (_tmp, store, search) = setup();
1040 put_and_index(&store, &search, Galaxy::Codex, "indexed memory");
1041
1042 store.put_raw(Galaxy::Karma, b"key1", b"value1").unwrap();
1045
1046 let report = check_consistency(&store, &search);
1047 assert!(
1048 !report.has_drift,
1049 "karma entries should not cause drift — non-memory galaxies are excluded"
1050 );
1051 let galaxy_names: Vec<_> = report.galaxies.iter().map(|g| g.galaxy.as_str()).collect();
1053 assert!(
1054 !galaxy_names.contains(&"karma"),
1055 "karma should not appear in consistency report"
1056 );
1057 assert!(
1058 !galaxy_names.contains(&"dharma"),
1059 "dharma should not appear in consistency report"
1060 );
1061 }
1062 #[test]
1065 fn classify_separates_skip_reserve_from_healable_drift() {
1066 let (tmp, store, search) = setup();
1067 put_and_index(&store, &search, Galaxy::Codex, "clean doc one");
1069 put_and_index(&store, &search, Galaxy::Codex, "clean doc two");
1070 store
1071 .put(
1072 Galaxy::Codex,
1073 &Memory::new(Galaxy::Codex, "bad \u{1}\u{2} doc".into()),
1074 )
1075 .unwrap();
1076 search.commit(&mut search.writer().unwrap()).unwrap();
1077
1078 let class = classify_drift(&store, &search);
1079 let codex = class.galaxies.iter().find(|g| g.galaxy == "codex").unwrap();
1080 assert_eq!(codex.lmdb_count, 3);
1081 assert_eq!(codex.tantivy_count, 2);
1082 assert_eq!(codex.skip_reserve, 1);
1083 assert_eq!(codex.healable_gap, 0, "skip reserve fully explains the gap");
1084 assert_eq!(class.healable_total, 0);
1085 drop(tmp);
1086 }
1087
1088 #[test]
1089 fn classify_flags_missing_indexable_docs_as_healable() {
1090 let (tmp, store, search) = setup();
1091 put_and_index(&store, &search, Galaxy::Codex, "indexed doc");
1092 store
1094 .put(
1095 Galaxy::Codex,
1096 &Memory::new(Galaxy::Codex, "unindexed clean doc".into()),
1097 )
1098 .unwrap();
1099 search.commit(&mut search.writer().unwrap()).unwrap();
1100
1101 let class = classify_drift(&store, &search);
1102 let codex = class.galaxies.iter().find(|g| g.galaxy == "codex").unwrap();
1103 assert_eq!(codex.skip_reserve, 0);
1104 assert_eq!(codex.healable_gap, 1);
1105 assert_eq!(class.healable_total, 1);
1106 drop(tmp);
1107 }
1108
1109 #[test]
1110 fn heal_ignores_pure_skip_reserve_and_heals_real_gaps() {
1111 let (tmp, store, search) = setup();
1112 put_and_index(&store, &search, Galaxy::Codex, "clean doc");
1113 store
1116 .put(
1117 Galaxy::Codex,
1118 &Memory::new(Galaxy::Codex, "gate\u{0} fails".into()),
1119 )
1120 .unwrap();
1121 search.commit(&mut search.writer().unwrap()).unwrap();
1122 let healed = heal_index_drift(&store, &search).unwrap();
1123 assert!(
1124 healed.is_none(),
1125 "skip-reserve-only drift must not trigger a rebuild"
1126 );
1127
1128 store
1130 .put(
1131 Galaxy::Codex,
1132 &Memory::new(Galaxy::Codex, "genuinely missing doc".into()),
1133 )
1134 .unwrap();
1135 let healed = heal_index_drift(&store, &search).unwrap();
1136 assert!(healed.is_some(), "healable drift must trigger a heal");
1137 let healed = healed.unwrap();
1142 assert_eq!(healed.indexed, 1);
1143 assert_eq!(healed.skipped, 1);
1144 assert_eq!(healed.deleted, 0);
1145 drop(tmp);
1146 }
1147
1148 #[test]
1149 fn heal_indexes_only_the_missing_delta() {
1150 let (_tmp, store, search) = setup();
1151 let a = Memory::new(Galaxy::Sessions, "alpha missing doc".into());
1153 let b = Memory::new(Galaxy::Sessions, "beta missing doc".into());
1154 store.put(Galaxy::Sessions, &a).unwrap();
1155 store.put(Galaxy::Sessions, &b).unwrap();
1156
1157 let report = heal_index_drift(&store, &search).unwrap().unwrap();
1159 assert_eq!(report.indexed, 2);
1160 assert_eq!(report.deleted, 0);
1161 assert_eq!(
1162 search
1163 .search_in_galaxy("alpha", Some(Galaxy::Sessions), 10)
1164 .unwrap()
1165 .len(),
1166 1
1167 );
1168
1169 store
1171 .put(
1172 Galaxy::Sessions,
1173 &Memory::new(Galaxy::Sessions, "gamma missing doc".into()),
1174 )
1175 .unwrap();
1176 let report = heal_index_drift(&store, &search).unwrap().unwrap();
1177 assert_eq!(report.indexed, 1, "only the delta is indexed");
1178 assert_eq!(
1179 search
1180 .search_in_galaxy("gamma", Some(Galaxy::Sessions), 10)
1181 .unwrap()
1182 .len(),
1183 1
1184 );
1185 assert_eq!(
1187 search
1188 .search_in_galaxy("alpha", Some(Galaxy::Sessions), 10)
1189 .unwrap()
1190 .len(),
1191 1
1192 );
1193 }
1194
1195 #[test]
1196 fn heal_deletes_orphan_index_docs() {
1197 let (_tmp, store, search) = setup();
1198 put_and_index(&store, &search, Galaxy::Codex, "legit doc");
1199 let orphan = Memory::new(Galaxy::Codex, "orphan doc".into());
1205 let orphan_id = orphan.metadata.id;
1206 store.put(Galaxy::Codex, &orphan).unwrap();
1207 {
1208 let mut writer = search.writer().unwrap();
1209 search
1210 .add_document(
1211 &mut writer,
1212 &orphan_id.to_string(),
1213 "codex",
1214 "orphan doc",
1215 &orphan.metadata.tags,
1216 orphan.metadata.created_at.timestamp(),
1217 )
1218 .unwrap();
1219 search.commit(&mut writer).unwrap();
1220 }
1221 store
1222 .delete_raw(Galaxy::Codex, orphan_id.as_bytes())
1223 .unwrap();
1224 assert_eq!(search.count_docs_in_galaxy("codex").unwrap(), 2);
1225
1226 let report = heal_index_drift(&store, &search).unwrap().unwrap();
1227 assert_eq!(report.deleted, 1, "the orphan must be deleted");
1228 assert_eq!(report.indexed, 0);
1229 assert_eq!(search.count_docs_in_galaxy("codex").unwrap(), 1);
1230 let mut gone = false;
1234 for _ in 0..40 {
1235 if search.search("orphan", 10).unwrap().is_empty() {
1238 gone = true;
1239 break;
1240 }
1241 std::thread::sleep(std::time::Duration::from_millis(50));
1242 }
1243 assert!(gone, "orphan doc must be gone from search");
1244 assert!(heal_index_drift(&store, &search).unwrap().is_none());
1245 }
1246
1247 #[test]
1248 fn repair_rewrites_in_place_and_indexes_clean_content() {
1249 let (tmp, store, search) = setup();
1250 let mut repairable = Memory::new(
1253 Galaxy::Codex,
1254 "kumquat\u{0} ratchet \u{1} repair end".into(),
1255 );
1256 let mut binary = Memory::new(Galaxy::Codex, "\u{1}\u{2}\u{3}\u{4}\u{5}\u{6}".into());
1258 let mut clean = Memory::new(Galaxy::Codex, "perfectly fine prose".into());
1260 let (id_r, id_b, id_c) = (
1261 repairable.metadata.id,
1262 binary.metadata.id,
1263 clean.metadata.id,
1264 );
1265 for m in [&mut repairable, &mut binary, &mut clean] {
1266 store.put(Galaxy::Codex, m).unwrap();
1267 }
1268
1269 let report = repair_content(&store, &search, &[Galaxy::Codex]).unwrap();
1270 assert_eq!(report.scanned, 3);
1271 assert_eq!(report.repaired, 1, "{report:?}");
1272 assert_eq!(report.unrepairable, 1, "{report:?}");
1273 assert_eq!(report.already_clean, 1);
1274
1275 let row = store.get(Galaxy::Codex, id_r).unwrap().unwrap();
1278 assert_eq!(row.content, "kumquat ratchet repair end");
1279 assert_eq!(row.metadata.content_hash, crate::content_hash(&row.content));
1280 assert!(sanitize_content_for_index(&row.content).is_some());
1281 assert_eq!(row.metadata.revision_count, 1);
1282
1283 let chain = store.revisions(Galaxy::Codex, id_r).unwrap();
1286 assert_eq!(chain.len(), 1);
1287 assert_eq!(
1288 chain[0].old_hash,
1289 crate::content_hash("kumquat\u{0} ratchet \u{1} repair end")
1290 );
1291 assert_eq!(chain[0].new_hash, row.metadata.content_hash);
1292 assert_eq!(chain[0].actor_user.as_deref(), Some("wm-repair-content"));
1293 assert_eq!(chain[0].actor_session, None);
1294 let verdict = store
1295 .verify_revision_chain(Galaxy::Codex, id_r, &row.metadata.content_hash)
1296 .unwrap();
1297 assert!(verdict.valid, "{:?}", verdict.breaks);
1298
1299 assert!(store.revisions(Galaxy::Codex, id_b).unwrap().is_empty());
1301 assert!(store.revisions(Galaxy::Codex, id_c).unwrap().is_empty());
1302
1303 let untouched = store.get(Galaxy::Codex, id_b).unwrap().unwrap();
1305 assert_eq!(untouched.content, "\u{1}\u{2}\u{3}\u{4}\u{5}\u{6}");
1306
1307 let kept = store.get(Galaxy::Codex, id_c).unwrap().unwrap();
1309 assert_eq!(kept.content, "perfectly fine prose");
1310
1311 let hits = search.search("kumquat ratchet repair", 10).unwrap();
1313 assert!(
1314 hits.iter().any(|h| h.memory_id == id_r.to_string()),
1315 "repaired doc must be indexed: {hits:?}"
1316 );
1317
1318 let again = repair_content(&store, &search, &[Galaxy::Codex]).unwrap();
1320 assert_eq!(again.repaired, 0);
1321 assert_eq!(again.already_clean, 2);
1322 assert_eq!(
1323 store.revisions(Galaxy::Codex, id_r).unwrap().len(),
1324 1,
1325 "idempotent re-run must not append"
1326 );
1327 drop(tmp);
1328 }
1329
1330 #[test]
1331 fn open_or_quarantine_recovers_an_unopenable_index() {
1332 let tmp = tempdir().unwrap();
1333 let index_dir = tmp.path().join("tantivy");
1334 std::fs::create_dir_all(&index_dir).unwrap();
1335 drop(SearchEngine::open(&index_dir).unwrap());
1336
1337 std::fs::write(index_dir.join("meta.json"), b"{ not json").unwrap();
1339
1340 let (engine, quarantine) = open_or_quarantine(&index_dir).unwrap();
1341 let quarantine = quarantine.expect("an unopenable index must be quarantined");
1342 assert!(quarantine.join("meta.json").exists(), "{quarantine:?}");
1343 assert!(
1344 quarantine
1345 .file_name()
1346 .unwrap()
1347 .to_string_lossy()
1348 .contains(".corrupt."),
1349 "quarantine keeps the old index beside the fresh one: {quarantine:?}"
1350 );
1351 assert_eq!(engine.count_docs_in_galaxy("codex").unwrap(), 0);
1353 drop(engine);
1354 assert!(index_dir.exists(), "a fresh index directory is created");
1355 }
1356}