1use super::*;
2use std::collections::{HashMap, HashSet};
3
4pub struct DirectoryReader {
5 files: Vec<FileReader>,
6 index: usize,
7 pending_realtime_seek: Option<u64>,
8 realtime_seek_bound: Option<(u64, Direction)>,
9 candidates: Vec<Option<DirectoryCandidate>>,
10 current_key: Option<DirectoryEntryKey>,
11 direction: Option<Direction>,
12 boot_newest: HashMap<[u8; 16], DirectoryBootNewest>,
13 pub(super) non_overlapping: bool,
14}
15
16#[derive(Debug, Clone, Copy)]
17struct DirectoryCandidate {
18 reader_index: usize,
19 key: DirectoryEntryKey,
20}
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub(super) struct DirectoryEntryKey {
24 pub(super) seqnum_id: [u8; 16],
25 pub(super) seqnum: u64,
26 pub(super) boot_id: [u8; 16],
27 pub(super) monotonic: u64,
28 pub(super) realtime: u64,
29 pub(super) xor_hash: u64,
30}
31
32#[derive(Debug, Clone, Copy)]
33struct DirectoryBootNewest {
34 machine_id: [u8; 16],
35 monotonic: u64,
36 realtime: u64,
37}
38
39impl DirectoryReader {
40 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
41 Self::open_with_options(path, ReaderOptions::default())
42 }
43
44 pub fn open_with_options(path: impl AsRef<Path>, options: ReaderOptions) -> Result<Self> {
45 let path = path.as_ref();
46 if !path.is_dir() {
47 return Err(SdkError::InvalidPath(format!(
48 "not a directory: {}",
49 path.display()
50 )));
51 }
52
53 let mut files = Vec::new();
54 for file_path in collect_journal_files(path)? {
55 if let Ok(reader) = FileReader::open_with_options(&file_path, options) {
56 files.push(reader);
57 }
58 }
59
60 Self::from_readers(files, true)
61 }
62
63 pub fn open_files<I, P>(paths: I) -> Result<Self>
64 where
65 I: IntoIterator<Item = P>,
66 P: AsRef<Path>,
67 {
68 Self::open_files_with_options(paths, ReaderOptions::default())
69 }
70
71 pub fn open_files_with_options<I, P>(paths: I, options: ReaderOptions) -> Result<Self>
72 where
73 I: IntoIterator<Item = P>,
74 P: AsRef<Path>,
75 {
76 let mut files = Vec::new();
77 for path in paths {
78 let path = path.as_ref();
79 if !path.is_file() || !is_journal_file_name(path) {
80 return Err(SdkError::InvalidPath(format!(
81 "not a journal file: {}",
82 path.display()
83 )));
84 }
85 files.push(FileReader::open_with_options(path, options)?);
86 }
87
88 Self::from_readers(files, false)
89 }
90
91 fn from_readers(mut files: Vec<FileReader>, allow_empty: bool) -> Result<Self> {
92 if files.is_empty() && !allow_empty {
93 return Err(SdkError::InvalidPath(
94 "no readable journal files".to_string(),
95 ));
96 }
97
98 files.sort_by_key(FileReader::header_realtime_start);
99 let boot_newest = build_directory_boot_newest(&files);
100 let non_overlapping = directory_files_non_overlapping(&files);
101 let candidates = vec![None; files.len()];
102 Ok(Self {
103 files,
104 index: usize::MAX,
105 pending_realtime_seek: None,
106 realtime_seek_bound: None,
107 candidates,
108 current_key: None,
109 direction: None,
110 boot_newest,
111 non_overlapping,
112 })
113 }
114
115 pub fn seek_head(&mut self) {
116 self.pending_realtime_seek = None;
117 self.realtime_seek_bound = None;
118 self.index = usize::MAX;
119 self.current_key = None;
120 self.direction = None;
121 self.reset_candidates();
122 for reader in &mut self.files {
123 reader.seek_head();
124 }
125 }
126
127 pub fn seek_tail(&mut self) {
128 self.pending_realtime_seek = None;
129 self.realtime_seek_bound = None;
130 self.index = usize::MAX;
131 self.current_key = None;
132 self.direction = None;
133 self.reset_candidates();
134 for reader in &mut self.files {
135 reader.seek_tail();
136 }
137 }
138
139 pub fn seek_realtime(&mut self, usec: u64) {
140 self.pending_realtime_seek = Some(usec);
141 self.realtime_seek_bound = None;
142 self.index = usize::MAX;
143 self.current_key = None;
144 self.direction = None;
145 self.reset_candidates();
146 }
147
148 pub fn next(&mut self) -> Result<bool> {
149 self.step_merged(Direction::Forward)
150 }
151
152 pub fn previous(&mut self) -> Result<bool> {
153 self.step_merged(Direction::Backward)
154 }
155
156 fn step_merged(&mut self, direction: Direction) -> Result<bool> {
157 if self.can_step_sequential(direction) {
158 return self.step_sequential(direction);
159 }
160
161 self.prepare_merge_direction(direction);
162
163 let mut best: Option<DirectoryCandidate> = None;
164 for idx in 0..self.files.len() {
165 self.fill_candidate(idx, direction)?;
166 let Some(candidate) = self.candidates[idx] else {
167 continue;
168 };
169 let replace = match best {
170 None => true,
171 Some(current) => {
172 let cmp = self.compare_entry_keys(candidate.key, current.key);
173 (direction == Direction::Forward && cmp < 0)
174 || (direction == Direction::Backward && cmp > 0)
175 }
176 };
177 if replace {
178 best = Some(candidate);
179 }
180 }
181
182 let Some(best) = best else {
183 self.index = usize::MAX;
184 self.realtime_seek_bound = None;
185 return Ok(false);
186 };
187
188 self.index = best.reader_index;
189 self.current_key = Some(best.key);
190 self.candidates[best.reader_index] = None;
191 self.realtime_seek_bound = None;
192 Ok(true)
193 }
194
195 fn prepare_merge_direction(&mut self, direction: Direction) {
196 if let Some(usec) = self.pending_realtime_seek.take() {
197 for reader in &mut self.files {
198 reader.seek_realtime(usec);
199 }
200 self.reset_candidates();
201 self.realtime_seek_bound = Some((usec, direction));
202 self.direction = Some(direction);
203 return;
204 }
205
206 if self.direction == Some(direction) {
207 return;
208 }
209
210 if let Some(current) = self.current_key {
211 for reader in &mut self.files {
212 reader.seek_realtime(current.realtime);
213 }
214 } else if direction == Direction::Forward {
215 for reader in &mut self.files {
216 reader.seek_head();
217 }
218 } else {
219 for reader in &mut self.files {
220 reader.seek_tail();
221 }
222 }
223
224 self.reset_candidates();
225 self.direction = Some(direction);
226 }
227
228 fn fill_candidate(&mut self, reader_index: usize, direction: Direction) -> Result<()> {
229 if self.candidates[reader_index].is_some() {
230 return Ok(());
231 }
232
233 loop {
234 if !self.advance_candidate_reader(reader_index, direction)? {
235 return Ok(());
236 }
237 let key = self.files[reader_index].current_directory_entry_key()?;
238 if !self.candidate_matches_realtime_bound(key) {
239 continue;
240 }
241 if !self.candidate_is_after_current(key, direction) {
242 continue;
243 }
244
245 self.candidates[reader_index] = Some(DirectoryCandidate { reader_index, key });
246 return Ok(());
247 }
248 }
249
250 fn advance_candidate_reader(
251 &mut self,
252 reader_index: usize,
253 direction: Direction,
254 ) -> Result<bool> {
255 match direction {
256 Direction::Forward => self.files[reader_index].next(),
257 Direction::Backward => self.files[reader_index].previous(),
258 }
259 }
260
261 fn candidate_matches_realtime_bound(&self, key: DirectoryEntryKey) -> bool {
262 let Some((usec, seek_direction)) = self.realtime_seek_bound else {
263 return true;
264 };
265 match seek_direction {
266 Direction::Forward => key.realtime >= usec,
267 Direction::Backward => key.realtime <= usec,
268 }
269 }
270
271 fn candidate_is_after_current(&self, key: DirectoryEntryKey, direction: Direction) -> bool {
272 let Some(current) = self.current_key else {
273 return true;
274 };
275 let cmp = self.compare_entry_keys(key, current);
276 match direction {
277 Direction::Forward => cmp > 0,
278 Direction::Backward => cmp < 0,
279 }
280 }
281
282 fn compare_entry_keys(&self, a: DirectoryEntryKey, b: DirectoryEntryKey) -> i8 {
283 if a == b {
284 return 0;
285 }
286
287 if a.seqnum_id == b.seqnum_id {
288 let cmp = cmp_u64(a.seqnum, b.seqnum);
289 if cmp != 0 {
290 return cmp;
291 }
292 }
293
294 if a.boot_id == b.boot_id {
295 let cmp = cmp_u64(a.monotonic, b.monotonic);
296 if cmp != 0 {
297 return cmp;
298 }
299 } else {
300 let cmp = self.compare_boot_ids(a.boot_id, b.boot_id);
301 if cmp != 0 {
302 return cmp;
303 }
304 }
305
306 let cmp = cmp_u64(a.realtime, b.realtime);
307 if cmp != 0 {
308 return cmp;
309 }
310 cmp_u64(a.xor_hash, b.xor_hash)
311 }
312
313 fn compare_boot_ids(&self, a: [u8; 16], b: [u8; 16]) -> i8 {
314 let Some(a_newest) = self.boot_newest.get(&a) else {
315 return 0;
316 };
317 let Some(b_newest) = self.boot_newest.get(&b) else {
318 return 0;
319 };
320 if a_newest.machine_id != b_newest.machine_id {
321 return 0;
322 }
323 cmp_u64(a_newest.realtime, b_newest.realtime)
324 }
325
326 fn reset_candidates(&mut self) {
327 if self.candidates.len() != self.files.len() {
328 self.candidates = vec![None; self.files.len()];
329 return;
330 }
331 for candidate in &mut self.candidates {
332 *candidate = None;
333 }
334 }
335
336 pub fn get_entry(&mut self) -> Result<Entry> {
337 if self.index >= self.files.len() {
338 return Err(SdkError::NoEntry);
339 }
340 self.files[self.index].get_entry()
341 }
342
343 pub fn visit_entry_payloads<F>(&mut self, visitor: F) -> Result<()>
344 where
345 F: FnMut(&[u8]) -> Result<()>,
346 {
347 if self.index >= self.files.len() {
348 return Err(SdkError::NoEntry);
349 }
350 self.files[self.index].visit_entry_payloads(visitor)
351 }
352
353 pub fn clear_entry_data_state(&mut self) {
354 if self.index < self.files.len() {
355 self.files[self.index].clear_entry_data_state();
356 }
357 }
358
359 pub fn entry_data_restart(&mut self) -> Result<()> {
360 if self.index >= self.files.len() {
361 return Err(SdkError::NoEntry);
362 }
363 self.files[self.index].entry_data_restart()
364 }
365
366 pub fn enumerate_entry_payload(&mut self) -> Result<Option<&[u8]>> {
367 if self.index >= self.files.len() {
368 return Err(SdkError::NoEntry);
369 }
370 self.files[self.index].enumerate_entry_payload()
371 }
372
373 pub fn collect_entry_payloads(&mut self, payloads: &mut Vec<Vec<u8>>) -> Result<()> {
374 if self.index >= self.files.len() {
375 return Err(SdkError::NoEntry);
376 }
377 self.files[self.index].collect_entry_payloads(payloads)
378 }
379
380 pub fn get_entry_payload(&mut self, field: &[u8]) -> Result<Option<Vec<u8>>> {
381 if self.index >= self.files.len() {
382 return Err(SdkError::NoEntry);
383 }
384 self.files[self.index].get_entry_payload(field)
385 }
386
387 pub fn get_realtime_usec(&self) -> Result<u64> {
388 if self.index >= self.files.len() {
389 return Err(SdkError::NoEntry);
390 }
391 self.files[self.index].get_realtime_usec()
392 }
393
394 pub fn get_seqnum(&self) -> Result<(u64, [u8; 16])> {
395 if self.index >= self.files.len() {
396 return Err(SdkError::NoEntry);
397 }
398 if let Some(key) = self.current_key {
399 return Ok((key.seqnum, key.seqnum_id));
400 }
401 self.files[self.index].get_seqnum()
402 }
403
404 pub fn get_monotonic_usec(&self) -> Result<(u64, [u8; 16])> {
405 if self.index >= self.files.len() {
406 return Err(SdkError::NoEntry);
407 }
408 if let Some(key) = self.current_key {
409 return Ok((key.monotonic, key.boot_id));
410 }
411 self.files[self.index].get_monotonic_usec()
412 }
413
414 pub fn get_cursor(&self) -> Result<String> {
415 if self.index >= self.files.len() {
416 return Err(SdkError::NoEntry);
417 }
418 self.files[self.index].get_cursor()
419 }
420
421 pub fn test_cursor(&self, cursor: &str) -> Result<bool> {
422 if self.index >= self.files.len() {
423 return Ok(false);
424 }
425 self.files[self.index].test_cursor(cursor)
426 }
427
428 pub fn seek_cursor(&mut self, cursor: &str) -> Result<()> {
429 let want = parse::parse_cursor_location(cursor, true)
430 .map_err(|err| SdkError::InvalidCursor(err.to_string()))?;
431 if want.realtime_set {
432 self.seek_realtime(want.realtime);
433 } else {
434 self.seek_head();
435 }
436 while self.next()? {
437 let current_cursor = self.get_cursor()?;
438 let got = parse::parse_cursor_location(¤t_cursor, false)
439 .map_err(|err| SdkError::InvalidCursor(err.to_string()))?;
440 if parse::cursor_location_at_or_after(&got, &want) {
441 return Ok(());
442 }
443 }
444 self.seek_tail();
445 Ok(())
446 }
447
448 pub fn enumerate_fields(&mut self) -> Result<Vec<String>> {
449 let mut fields = HashSet::new();
450 for reader in &mut self.files {
451 for field in reader.enumerate_fields()? {
452 fields.insert(field);
453 }
454 }
455 let mut out: Vec<_> = fields.into_iter().collect();
456 out.sort();
457 Ok(out)
458 }
459
460 pub fn query_unique(&mut self, field_name: &str) -> Result<Vec<Vec<u8>>> {
461 let mut out = Vec::new();
462 self.visit_unique_values(field_name, |value| {
463 out.push(value.to_vec());
464 Ok(())
465 })?;
466 Ok(out)
467 }
468
469 pub fn visit_unique_values<F>(&mut self, field_name: &str, mut visitor: F) -> Result<()>
470 where
471 F: FnMut(&[u8]) -> Result<()>,
472 {
473 if self.files.len() == 1 {
474 return self.files[0].visit_unique_values(field_name, visitor);
475 }
476
477 let mut seen = HashSet::new();
478 for reader in &mut self.files {
479 reader.visit_unique_values(field_name, |value| {
480 if seen.insert(value.to_vec()) {
481 visitor(value)?;
482 }
483 Ok(())
484 })?;
485 }
486 Ok(())
487 }
488
489 pub fn list_boots(&self) -> Vec<BootInfo> {
490 let mut boots: HashMap<String, (i64, i64)> = HashMap::new();
491 for reader in &self.files {
492 let header = reader.cached_header().header;
493 let boot_id = hex::encode(header.tail_entry_boot_id);
494 let first = header.head_entry_realtime as i64;
495 let last = header.tail_entry_realtime as i64;
496 boots
497 .entry(boot_id)
498 .and_modify(|range| {
499 range.0 = range.0.min(first);
500 range.1 = range.1.max(last);
501 })
502 .or_insert((first, last));
503 }
504
505 let mut out: Vec<_> = boots
506 .into_iter()
507 .map(|(boot_id, (first_entry, last_entry))| BootInfo {
508 index: 0,
509 boot_id,
510 first_entry,
511 last_entry,
512 })
513 .collect();
514 out.sort_by_key(|boot| boot.first_entry);
515 let base = 1 - out.len() as i64;
516 for (idx, boot) in out.iter_mut().enumerate() {
517 boot.index = base + idx as i64;
518 }
519 out
520 }
521
522 pub fn add_match(&mut self, data: &[u8]) {
523 for reader in &mut self.files {
524 reader.add_match(data);
525 }
526 self.reset_merge_state();
527 }
528
529 pub fn add_conjunction(&mut self) -> Result<()> {
530 for reader in &mut self.files {
531 reader.add_conjunction()?;
532 }
533 self.reset_merge_state();
534 Ok(())
535 }
536
537 pub fn add_disjunction(&mut self) -> Result<()> {
538 for reader in &mut self.files {
539 reader.add_disjunction()?;
540 }
541 self.reset_merge_state();
542 Ok(())
543 }
544
545 pub fn flush_matches(&mut self) {
546 for reader in &mut self.files {
547 reader.flush_matches();
548 }
549 self.reset_merge_state();
550 }
551
552 fn reset_merge_state(&mut self) {
553 self.index = usize::MAX;
554 self.current_key = None;
555 self.direction = None;
556 self.realtime_seek_bound = None;
557 self.reset_candidates();
558 }
559
560 fn can_step_sequential(&self, direction: Direction) -> bool {
561 if !self.non_overlapping || self.pending_realtime_seek.is_some() {
562 return false;
563 }
564 if self.direction.is_some_and(|current| current != direction) && self.current_key.is_some()
565 {
566 return false;
567 }
568 true
569 }
570
571 fn step_sequential(&mut self, direction: Direction) -> Result<bool> {
572 if self.files.is_empty() {
573 self.clear_current_directory_entry();
574 return Ok(false);
575 }
576
577 if self.direction != Some(direction) {
578 self.reset_sequential_direction(direction);
579 }
580
581 match direction {
582 Direction::Forward => self.step_sequential_forward(),
583 Direction::Backward => self.step_sequential_backward(),
584 }
585 }
586
587 fn reset_sequential_direction(&mut self, direction: Direction) {
588 match direction {
589 Direction::Forward => {
590 for reader in &mut self.files {
591 reader.seek_head();
592 }
593 self.index = 0;
594 }
595 Direction::Backward => {
596 for reader in &mut self.files {
597 reader.seek_tail();
598 }
599 self.index = self.files.len() - 1;
600 }
601 }
602 self.reset_candidates();
603 self.current_key = None;
604 self.realtime_seek_bound = None;
605 self.direction = Some(direction);
606 }
607
608 fn step_sequential_forward(&mut self) -> Result<bool> {
609 if self.index == usize::MAX {
610 self.index = 0;
611 }
612 while self.index < self.files.len() {
613 if self.files[self.index].next()? {
614 self.current_key = Some(self.files[self.index].current_directory_entry_key()?);
615 return Ok(true);
616 }
617 self.index += 1;
618 }
619 self.finish_sequential_end()
620 }
621
622 fn step_sequential_backward(&mut self) -> Result<bool> {
623 if self.index >= self.files.len() {
624 self.index = self.files.len() - 1;
625 }
626 loop {
627 if self.files[self.index].previous()? {
628 self.current_key = Some(self.files[self.index].current_directory_entry_key()?);
629 return Ok(true);
630 }
631 if self.index == 0 {
632 break;
633 }
634 self.index -= 1;
635 }
636 self.finish_sequential_end()
637 }
638
639 fn finish_sequential_end(&mut self) -> Result<bool> {
640 self.clear_current_directory_entry();
641 Ok(false)
642 }
643
644 fn clear_current_directory_entry(&mut self) {
645 self.index = usize::MAX;
646 self.current_key = None;
647 }
648}
649
650pub(super) fn is_journal_file_name(path: &Path) -> bool {
651 path.file_name()
652 .and_then(|name| name.to_str())
653 .is_some_and(|name| {
654 name.ends_with(".journal")
655 || name.ends_with(".journal~")
656 || name.ends_with(".journal.zst")
657 || name.ends_with(".journal~.zst")
658 })
659}
660
661fn collect_journal_files(path: &Path) -> Result<Vec<PathBuf>> {
662 let entries: Vec<_> = std::fs::read_dir(path)?.collect::<std::io::Result<Vec<_>>>()?;
663 let mut files = Vec::new();
664
665 for entry in &entries {
666 let file_path = entry.path();
667 if file_path.is_file() && is_journal_file_name(&file_path) {
668 files.push(file_path);
669 }
670 }
671
672 for entry in &entries {
673 let Some(name) = entry.file_name().to_str().map(str::to_owned) else {
674 continue;
675 };
676 if !is_journal_subdir_name(&name) {
677 continue;
678 }
679 let child_path = entry.path();
680 if !child_path.is_dir() {
681 continue;
682 }
683 let Ok(children) = std::fs::read_dir(&child_path) else {
684 continue;
685 };
686 for child in children.flatten() {
687 let file_path = child.path();
688 if file_path.is_file() && is_journal_file_name(&file_path) {
689 files.push(file_path);
690 }
691 }
692 }
693
694 files.sort();
695 Ok(files)
696}
697
698fn is_journal_subdir_name(name: &str) -> bool {
699 if name.contains('.') {
700 return false;
701 }
702 id128_string_valid(name)
703}
704
705fn id128_string_valid(s: &str) -> bool {
706 match s.len() {
707 32 => s.bytes().all(|byte| byte.is_ascii_hexdigit()),
708 36 => s.bytes().enumerate().all(|(idx, byte)| {
709 if matches!(idx, 8 | 13 | 18 | 23) {
710 byte == b'-'
711 } else {
712 byte.is_ascii_hexdigit()
713 }
714 }),
715 _ => false,
716 }
717}
718
719fn build_directory_boot_newest(files: &[FileReader]) -> HashMap<[u8; 16], DirectoryBootNewest> {
720 let mut newest: HashMap<[u8; 16], DirectoryBootNewest> = HashMap::new();
721 for reader in files {
722 let header = reader.cached_header();
723 if header.header.tail_entry_boot_id == [0; 16] {
724 continue;
725 }
726 let replace = match newest.get(&header.header.tail_entry_boot_id) {
727 None => true,
728 Some(current) => header.header.tail_entry_monotonic > current.monotonic,
729 };
730 if replace {
731 newest.insert(
732 header.header.tail_entry_boot_id,
733 DirectoryBootNewest {
734 machine_id: header.header.machine_id,
735 monotonic: header.header.tail_entry_monotonic,
736 realtime: header.header.tail_entry_realtime,
737 },
738 );
739 }
740 }
741 newest
742}
743
744fn directory_files_non_overlapping(files: &[FileReader]) -> bool {
745 if files.is_empty() {
746 return false;
747 }
748
749 for pair in files.windows(2) {
750 let previous = pair[0].cached_header().header;
751 let next = pair[1].cached_header().header;
752 if previous.seqnum_id != next.seqnum_id
753 || previous.tail_entry_seqnum == 0
754 || next.head_entry_seqnum == 0
755 || previous.tail_entry_seqnum >= next.head_entry_seqnum
756 || previous.tail_entry_realtime == 0
757 || next.head_entry_realtime == 0
758 || previous.tail_entry_realtime >= next.head_entry_realtime
759 {
760 return false;
761 }
762 }
763
764 true
765}
766
767fn cmp_u64(a: u64, b: u64) -> i8 {
768 match a.cmp(&b) {
769 std::cmp::Ordering::Less => -1,
770 std::cmp::Ordering::Equal => 0,
771 std::cmp::Ordering::Greater => 1,
772 }
773}