Skip to main content

journal/
lib.rs

1//! Pure-Rust systemd journal reader and writer SDK.
2//!
3//! This crate provides a public Rust layer over the imported Netdata journal
4//! reader/writer crates. It intentionally keeps the low-level file parsing in
5//! the imported implementation and adds byte-safe entries, directory reading,
6//! export/JSON formatting, and a libsystemd-style facade.
7
8mod directory;
9mod explorer;
10mod export;
11mod facade;
12pub mod netdata;
13mod parse;
14mod reader_helpers;
15mod sealed_verify;
16mod verify_graph;
17
18pub use directory::DirectoryReader;
19pub use explorer::{
20    ExplorerAnchor, ExplorerComparison, ExplorerControl, ExplorerFieldMode, ExplorerFilter,
21    ExplorerFtsPattern, ExplorerHistogram, ExplorerHistogramBucket, ExplorerProgress,
22    ExplorerQuery, ExplorerResult, ExplorerRow, ExplorerSampling, ExplorerStats,
23    ExplorerStopReason, ExplorerStrategy,
24};
25pub use export::{export_entry, export_entry_bytes, format_entry_text, json_entry};
26pub use parse::{ParseError, ParsedCursor, parse_cursor, parse_match_bytes, parse_match_string};
27pub use sealed_verify::{verify_file, verify_file_with_key};
28
29use ouroboros::self_referencing;
30use std::collections::HashMap;
31use std::fmt;
32use std::num::NonZeroU64;
33use std::path::{Path, PathBuf};
34
35use directory::DirectoryEntryKey;
36#[cfg(test)]
37use directory::is_journal_file_name;
38use reader_helpers::*;
39#[cfg(test)]
40use sealed_verify::{
41    COMPACT_DATA_OBJECT_HEADER_SIZE, DATA_OBJECT_HEADER_SIZE, HEADER_MIN_SIZE,
42    INCOMPATIBLE_COMPACT, OBJECT_HEADER_SIZE, OBJECT_TYPE_DATA, OBJECT_TYPE_TAG, align8,
43};
44
45pub use facade::{
46    ERR_END_OF_ENTRIES, ERR_INVALID_CURSOR, ERR_NO_ENTRY, ERR_UNSUPPORTED, Error as FacadeError,
47    OutputMode, SdJournal, SdJournalAddConjunction, SdJournalAddDisjunction, SdJournalAddMatch,
48    SdJournalClose, SdJournalEnumerateAvailableData, SdJournalEnumerateAvailableUnique,
49    SdJournalEnumerateField, SdJournalEnumerateFields, SdJournalFlushMatches, SdJournalGetCursor,
50    SdJournalGetData, SdJournalGetEntry, SdJournalGetMonotonicUsec, SdJournalGetRealtimeUsec,
51    SdJournalGetSeqnum, SdJournalListBoots, SdJournalNext, SdJournalNextSkip, SdJournalOpen,
52    SdJournalOpenDirectory, SdJournalOpenDirectoryWithOptions, SdJournalOpenFile,
53    SdJournalOpenFileWithOptions, SdJournalOpenFiles, SdJournalOpenFilesWithOptions,
54    SdJournalPrevious, SdJournalPreviousSkip, SdJournalProcessOutput, SdJournalQueryUnique,
55    SdJournalQueryUniqueState, SdJournalRestartData, SdJournalRestartFields,
56    SdJournalRestartUnique, SdJournalSeekCursor, SdJournalSeekHead, SdJournalSeekRealtimeUsec,
57    SdJournalSeekTail, SdJournalSetOutputMode, SdJournalTestCursor, SdJournalVisitUniqueValues,
58};
59pub use journal_core::error::JournalError;
60use journal_core::file::ExperimentalMmapStrategy;
61pub use journal_core::file::{
62    BucketUtilization, Compression, Direction, EntryItemsType, FieldNamePolicy, HashableObject,
63    JournalFile, JournalReader, Location, Mmap, WindowManagerStats,
64};
65use journal_core::file::{CurrentRowMetadata, CurrentRowView};
66pub use journal_log_writer::{
67    Config, EntryTimestamps, Log, LogLifecycleEvent, LogLifecycleObserver, RetentionPolicy,
68    RotationPolicy, WriterError,
69};
70pub use journal_registry::{Origin, Source};
71
72pub type Result<T> = std::result::Result<T, SdkError>;
73
74#[derive(Debug)]
75pub enum SdkError {
76    Journal(JournalError),
77    InvalidPath(String),
78    InvalidCursor(String),
79    NoEntry,
80    DecompressionFailed(String),
81    Unsupported(&'static str),
82    VerificationError(String),
83}
84
85impl fmt::Display for SdkError {
86    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
87        match self {
88            Self::Journal(err) => write!(f, "{err}"),
89            Self::InvalidPath(path) => write!(f, "invalid path: {path}"),
90            Self::InvalidCursor(cursor) => write!(f, "invalid cursor: {cursor}"),
91            Self::NoEntry => write!(f, "no entry at current position"),
92            Self::DecompressionFailed(err) => write!(f, "decompression failed: {err}"),
93            Self::Unsupported(op) => write!(f, "unsupported operation: {op}"),
94            Self::VerificationError(msg) => {
95                write!(f, "journal verification failed: corrupt file: {msg}")
96            }
97        }
98    }
99}
100
101impl std::error::Error for SdkError {}
102
103impl From<JournalError> for SdkError {
104    fn from(err: JournalError) -> Self {
105        Self::Journal(err)
106    }
107}
108
109impl From<std::io::Error> for SdkError {
110    fn from(err: std::io::Error) -> Self {
111        Self::Journal(JournalError::Io(err))
112    }
113}
114
115#[derive(Debug, Clone)]
116pub struct Field {
117    pub name: String,
118    pub value: Vec<u8>,
119}
120
121impl Field {
122    pub fn new(name: &str, value: &str) -> Self {
123        Self {
124            name: name.to_string(),
125            value: value.as_bytes().to_vec(),
126        }
127    }
128
129    pub fn with_bytes(name: &str, value: Vec<u8>) -> Self {
130        Self {
131            name: name.to_string(),
132            value,
133        }
134    }
135
136    pub fn payload(&self) -> Vec<u8> {
137        let mut payload = Vec::with_capacity(self.name.len() + 1 + self.value.len());
138        payload.extend_from_slice(self.name.as_bytes());
139        payload.push(b'=');
140        payload.extend_from_slice(&self.value);
141        payload
142    }
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
146pub enum ReaderBounds {
147    /// Systemd-style mutable reader bounds.
148    ///
149    /// The reader keeps a cached file size and refreshes it only when a read
150    /// would go beyond the cached end of file, matching libsystemd's active
151    /// journal behavior without a metadata syscall on every object read.
152    Live,
153    /// Immutable reader bounds.
154    ///
155    /// The reader fixes the file size at open time, like
156    /// `SD_JOURNAL_ASSUME_IMMUTABLE`, for polling/query consumers that do not
157    /// need to observe appends during the current scan.
158    Snapshot,
159}
160
161pub const DEFAULT_READER_WINDOW_SIZE: u64 = 32 * 1024 * 1024;
162
163impl Default for ReaderBounds {
164    fn default() -> Self {
165        Self::Live
166    }
167}
168
169#[derive(Debug, Clone, Copy, PartialEq, Eq)]
170pub struct ReaderOptions {
171    pub window_size: u64,
172    pub bounds: ReaderBounds,
173    mmap_strategy: ExperimentalMmapStrategy,
174}
175
176impl Default for ReaderOptions {
177    fn default() -> Self {
178        Self {
179            window_size: DEFAULT_READER_WINDOW_SIZE,
180            bounds: ReaderBounds::Live,
181            mmap_strategy: ExperimentalMmapStrategy::Windowed,
182        }
183    }
184}
185
186impl ReaderOptions {
187    pub fn live() -> Self {
188        Self::default()
189    }
190
191    pub fn snapshot() -> Self {
192        Self {
193            bounds: ReaderBounds::Snapshot,
194            ..Self::default()
195        }
196    }
197
198    pub fn with_window_size(mut self, window_size: u64) -> Self {
199        self.window_size = window_size;
200        self
201    }
202
203    pub fn with_bounds(mut self, bounds: ReaderBounds) -> Self {
204        self.bounds = bounds;
205        self
206    }
207
208    #[doc(hidden)]
209    pub fn with_experimental_mmap_strategy(mut self, strategy: ExperimentalMmapStrategy) -> Self {
210        self.mmap_strategy = strategy;
211        self
212    }
213}
214
215#[derive(Debug, Clone, Copy, PartialEq, Eq)]
216pub struct RawField<'a> {
217    pub name: &'a [u8],
218    pub value: &'a [u8],
219}
220
221impl RawField<'_> {
222    pub fn payload(&self) -> Vec<u8> {
223        let mut payload = Vec::with_capacity(self.name.len() + 1 + self.value.len());
224        payload.extend_from_slice(self.name);
225        payload.push(b'=');
226        payload.extend_from_slice(self.value);
227        payload
228    }
229
230    pub fn name_str(&self) -> Option<&str> {
231        std::str::from_utf8(self.name).ok()
232    }
233}
234
235#[derive(Debug, Clone)]
236pub struct Entry {
237    /// Convenience map for UTF-8 field names. RAW-mode files may contain field
238    /// names that are not valid UTF-8; use `raw_fields()` or `get_raw_values()`
239    /// when byte-identical field-name identity matters.
240    pub fields: HashMap<String, Vec<u8>>,
241    /// Convenience repeated-value map for UTF-8 field names.
242    pub field_values: HashMap<String, Vec<Vec<u8>>>,
243    /// Full on-disk DATA payloads as `FIELD=value` bytes.
244    pub payloads: Vec<Vec<u8>>,
245    pub seqnum: u64,
246    pub realtime: u64,
247    pub monotonic: u64,
248    pub boot_id: [u8; 16],
249    pub cursor: String,
250}
251
252impl Entry {
253    pub fn get(&self, key: &str) -> Option<&[u8]> {
254        self.fields.get(key).map(Vec::as_slice)
255    }
256
257    pub fn get_str(&self, key: &str) -> Option<&str> {
258        self.get(key)
259            .and_then(|value| std::str::from_utf8(value).ok())
260    }
261
262    pub fn raw_fields(&self) -> impl Iterator<Item = RawField<'_>> {
263        self.payloads
264            .iter()
265            .filter_map(|payload| split_raw_payload(payload))
266    }
267
268    pub fn get_raw(&self, key: &[u8]) -> Option<&[u8]> {
269        self.raw_fields()
270            .find(|field| field.name == key)
271            .map(|field| field.value)
272    }
273
274    pub fn get_raw_values(&self, key: &[u8]) -> Vec<&[u8]> {
275        self.raw_fields()
276            .filter_map(|field| (field.name == key).then_some(field.value))
277            .collect()
278    }
279}
280
281fn split_raw_payload(payload: &[u8]) -> Option<RawField<'_>> {
282    let eq = payload.iter().position(|byte| *byte == b'=')?;
283    Some(RawField {
284        name: &payload[..eq],
285        value: &payload[eq + 1..],
286    })
287}
288
289#[derive(Debug, Clone)]
290pub struct BootInfo {
291    pub index: i64,
292    pub boot_id: String,
293    pub first_entry: i64,
294    pub last_entry: i64,
295}
296
297#[derive(Debug, Clone, Copy)]
298pub struct FileHeader {
299    pub signature: [u8; 8],
300    pub compatible_flags: u32,
301    pub incompatible_flags: u32,
302    pub state: u8,
303    pub file_id: [u8; 16],
304    pub machine_id: [u8; 16],
305    pub header_size: u64,
306    pub arena_size: u64,
307    pub data_hash_table_size: u64,
308    pub field_hash_table_size: u64,
309    pub n_objects: u64,
310    pub n_entries: u64,
311    pub head_entry_realtime: u64,
312    pub tail_entry_realtime: u64,
313    pub tail_entry_monotonic: u64,
314    pub head_entry_seqnum: u64,
315    pub tail_entry_seqnum: u64,
316    pub tail_entry_boot_id: [u8; 16],
317    pub seqnum_id: [u8; 16],
318    pub n_data: u64,
319    pub n_fields: u64,
320    pub n_tags: u64,
321    pub n_entry_arrays: u64,
322    pub data_hash_chain_depth: u64,
323    pub field_hash_chain_depth: u64,
324}
325
326#[derive(Debug, Clone, Copy)]
327pub(crate) struct FileHeaderSnapshot {
328    pub(crate) header: FileHeader,
329}
330
331impl FileHeaderSnapshot {
332    fn from_file(file: &JournalFile<Mmap>) -> Self {
333        let header = file.journal_header_ref();
334        Self {
335            header: FileHeader {
336                signature: header.signature,
337                compatible_flags: header.compatible_flags,
338                incompatible_flags: header.incompatible_flags,
339                state: header.state,
340                file_id: header.file_id,
341                machine_id: header.machine_id,
342                header_size: header.header_size,
343                arena_size: header.arena_size,
344                data_hash_table_size: header
345                    .data_hash_table_size
346                    .map(|value| value.get())
347                    .unwrap_or(0),
348                field_hash_table_size: header
349                    .field_hash_table_size
350                    .map(|value| value.get())
351                    .unwrap_or(0),
352                n_objects: header.n_objects,
353                n_entries: header.n_entries,
354                head_entry_realtime: header.head_entry_realtime,
355                tail_entry_realtime: header.tail_entry_realtime,
356                tail_entry_monotonic: header.tail_entry_monotonic,
357                head_entry_seqnum: header.head_entry_seqnum,
358                tail_entry_seqnum: header.tail_entry_seqnum,
359                tail_entry_boot_id: header.tail_entry_boot_id,
360                seqnum_id: header.seqnum_id,
361                n_data: header.n_data,
362                n_fields: header.n_fields,
363                n_tags: header.n_tags,
364                n_entry_arrays: header.n_entry_arrays,
365                data_hash_chain_depth: header.data_hash_chain_depth,
366                field_hash_chain_depth: header.field_hash_chain_depth,
367            },
368        }
369    }
370}
371
372#[self_referencing]
373struct ReaderCell {
374    file: JournalFile<Mmap>,
375    #[borrows(file)]
376    #[not_covariant]
377    reader: JournalReader<'this, Mmap>,
378}
379
380pub struct FileReader {
381    inner: ReaderCell,
382    temp_path: Option<PathBuf>,
383    row: CurrentRowView,
384    header_snapshot: FileHeaderSnapshot,
385    bounds: ReaderBounds,
386}
387
388fn key_from_metadata(metadata: CurrentRowMetadata) -> DirectoryEntryKey {
389    DirectoryEntryKey {
390        seqnum_id: metadata.seqnum_id,
391        seqnum: metadata.seqnum,
392        boot_id: metadata.boot_id,
393        monotonic: metadata.monotonic,
394        realtime: metadata.realtime,
395        xor_hash: metadata.xor_hash,
396    }
397}
398
399enum StepStatus {
400    Valid,
401    Skip,
402    End,
403}
404
405impl Drop for FileReader {
406    fn drop(&mut self) {
407        self.inner
408            .with_file(|file| self.row.clear_current_best_effort(file));
409        if let Some(path) = &self.temp_path {
410            let _ = std::fs::remove_file(path);
411        }
412    }
413}
414
415impl FileReader {
416    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
417        Self::open_with_options(path, ReaderOptions::default())
418    }
419
420    pub fn open_with_options(path: impl AsRef<Path>, options: ReaderOptions) -> Result<Self> {
421        let path = path.as_ref();
422        if is_zst_file(path) {
423            return Self::open_zst(path, options);
424        }
425
426        let file = open_journal_file(path, options)?;
427        let header_snapshot = FileHeaderSnapshot::from_file(&file);
428        Ok(Self {
429            inner: ReaderCellBuilder {
430                file,
431                reader_builder: |_file| JournalReader::default(),
432            }
433            .build(),
434            temp_path: None,
435            row: CurrentRowView::default(),
436            header_snapshot,
437            bounds: options.bounds,
438        })
439    }
440
441    fn open_zst(path: &Path, options: ReaderOptions) -> Result<Self> {
442        let temp_path = decompress_zst_to_temp(path, "rust-sdk-journal")?;
443        let file = match open_journal_file(&temp_path, options) {
444            Ok(file) => file,
445            Err(err) => {
446                let _ = std::fs::remove_file(&temp_path);
447                return Err(err);
448            }
449        };
450        let header_snapshot = FileHeaderSnapshot::from_file(&file);
451        Ok(Self {
452            inner: ReaderCellBuilder {
453                file,
454                reader_builder: |_file| JournalReader::default(),
455            }
456            .build(),
457            temp_path: Some(temp_path),
458            row: CurrentRowView::default(),
459            header_snapshot,
460            bounds: options.bounds,
461        })
462    }
463
464    pub fn header(&self) -> FileHeader {
465        if self.bounds == ReaderBounds::Snapshot {
466            return self.header_snapshot.header;
467        }
468        self.live_header()
469    }
470
471    pub(crate) fn cached_header(&self) -> FileHeaderSnapshot {
472        self.header_snapshot
473    }
474
475    fn live_header(&self) -> FileHeader {
476        self.inner
477            .with_file(|file| FileHeaderSnapshot::from_file(file).header)
478    }
479
480    pub fn bucket_utilization(&self) -> Option<BucketUtilization> {
481        self.inner.with_file(JournalFile::bucket_utilization)
482    }
483
484    #[doc(hidden)]
485    pub fn mmap_stats(&self) -> Result<WindowManagerStats> {
486        self.inner
487            .with_file(|file| file.mmap_stats())
488            .map_err(Into::into)
489    }
490
491    pub fn seek_head(&mut self) {
492        self.inner
493            .with_file(|file| self.row.clear_current_best_effort(file));
494        self.inner.with_reader_mut(|reader| {
495            reader.set_location(Location::Head);
496        });
497    }
498
499    pub fn seek_tail(&mut self) {
500        self.inner
501            .with_file(|file| self.row.clear_current_best_effort(file));
502        self.inner.with_reader_mut(|reader| {
503            reader.set_location(Location::Tail);
504        });
505    }
506
507    pub fn seek_realtime(&mut self, usec: u64) {
508        self.inner
509            .with_file(|file| self.row.clear_current_best_effort(file));
510        self.inner.with_reader_mut(|reader| {
511            reader.set_location(Location::Realtime(usec));
512        });
513    }
514
515    pub fn seek_cursor(&mut self, cursor: &str) -> Result<()> {
516        let want = parse::parse_cursor_location(cursor, true)
517            .map_err(|err| SdkError::InvalidCursor(err.to_string()))?;
518        if want.realtime_set {
519            self.seek_realtime(want.realtime);
520        } else {
521            self.seek_head();
522        }
523        while self.next()? {
524            let current_cursor = self.get_cursor()?;
525            let got = parse::parse_cursor_location(&current_cursor, false)
526                .map_err(|err| SdkError::InvalidCursor(err.to_string()))?;
527            if parse::cursor_location_at_or_after(&got, &want) {
528                return Ok(());
529            }
530        }
531        self.seek_tail();
532        Ok(())
533    }
534
535    pub fn next(&mut self) -> Result<bool> {
536        self.step_valid(Direction::Forward)
537    }
538
539    pub fn previous(&mut self) -> Result<bool> {
540        self.step_valid(Direction::Backward)
541    }
542
543    fn step_valid(&mut self, direction: Direction) -> Result<bool> {
544        self.inner
545            .with_file(|file| self.row.clear_current(file))
546            .map_err(SdkError::from)?;
547        loop {
548            let row = &mut self.row;
549            let status = self.inner.with_mut(|fields| {
550                if !fields.reader.step(fields.file, direction)? {
551                    return Ok(StepStatus::End);
552                }
553
554                match fields
555                    .reader
556                    .get_entry_offset()
557                    .and_then(|offset| row.load_entry(fields.file, offset))
558                {
559                    Ok(_) => Ok(StepStatus::Valid),
560                    Err(err) if recoverable_entry_error(&err) => Ok(StepStatus::Skip),
561                    Err(err) => Err(err),
562                }
563            })?;
564
565            match status {
566                StepStatus::Valid => {
567                    return Ok(true);
568                }
569                StepStatus::Skip => continue,
570                StepStatus::End => {
571                    self.inner
572                        .with_file(|file| self.row.clear_current(file))
573                        .map_err(SdkError::from)?;
574                    return Ok(false);
575                }
576            }
577        }
578    }
579
580    pub fn get_entry(&mut self) -> Result<Entry> {
581        self.invalidate_entry_data_state();
582        let inner = &mut self.inner;
583        let row = &mut self.row;
584        inner.with_mut(|fields| {
585            if row.entry_offset().is_none() {
586                let offset = fields.reader.get_entry_offset()?;
587                row.load_entry(fields.file, offset)?;
588            }
589            read_current_row_entry(fields.file, row)
590        })
591    }
592
593    pub fn visit_entry_payloads<F>(&mut self, mut visitor: F) -> Result<()>
594    where
595        F: FnMut(&[u8]) -> Result<()>,
596    {
597        self.invalidate_entry_data_state();
598        let inner = &mut self.inner;
599        let row = &mut self.row;
600        inner.with_mut(|fields| {
601            fields.reader.release_object_guards();
602            if row.entry_offset().is_none() {
603                let offset = fields.reader.get_entry_offset()?;
604                row.load_entry(fields.file, offset)?;
605            }
606            row.restart_data()?;
607            loop {
608                let payload = match row.read_next_payload(fields.file) {
609                    Ok(Some(payload)) => payload,
610                    Ok(None) => break,
611                    Err(err) if recoverable_entry_data_error(&err) => continue,
612                    Err(err) => {
613                        let _ = row.reset_data_state(fields.file);
614                        return Err(err.into());
615                    }
616                };
617                let payload = row.payload_slice(payload);
618                if let Err(err) = visitor(payload) {
619                    let _ = row.reset_data_state(fields.file);
620                    return Err(err);
621                }
622            }
623            row.reset_data_state(fields.file)?;
624            Ok(())
625        })
626    }
627
628    pub fn clear_entry_data_state(&mut self) {
629        self.inner
630            .with_file(|file| self.row.reset_data_state_best_effort(file));
631        self.inner
632            .with_reader_mut(|reader| reader.entry_data_restart());
633    }
634
635    fn invalidate_entry_data_state(&mut self) {
636        if self.row.data_state_active() {
637            self.clear_entry_data_state();
638        }
639    }
640
641    pub fn entry_data_restart(&mut self) -> Result<()> {
642        self.inner
643            .with_file(|file| self.row.clear_pins(file))
644            .map_err(SdkError::from)?;
645        self.inner
646            .with_reader_mut(|reader| reader.entry_data_restart());
647        if self.row.entry_offset().is_none() {
648            let row = &mut self.row;
649            self.inner.with_mut(|fields| {
650                let offset = fields.reader.get_entry_offset()?;
651                row.load_entry(fields.file, offset).map(|_| ())
652            })?;
653        }
654        self.row.restart_data().map_err(Into::into)
655    }
656
657    pub fn enumerate_entry_payload(&mut self) -> Result<Option<&[u8]>> {
658        let row = &mut self.row;
659        let payload = self.inner.with_mut(|fields| {
660            fields.reader.release_object_guards();
661            row.read_next_payload(fields.file)
662        })?;
663        Ok(payload.map(|payload| self.row.payload_slice(payload)))
664    }
665
666    pub fn collect_entry_payloads(&mut self, payloads: &mut Vec<Vec<u8>>) -> Result<()> {
667        payloads.clear();
668        self.visit_entry_payloads(|payload| {
669            payloads.push(payload.to_vec());
670            Ok(())
671        })
672    }
673
674    pub fn get_entry_payload(&mut self, field: &[u8]) -> Result<Option<Vec<u8>>> {
675        let mut found = None;
676        self.visit_entry_payloads(|payload| {
677            if found.is_none()
678                && payload.len() > field.len()
679                && payload.starts_with(field)
680                && payload[field.len()] == b'='
681            {
682                found = Some(payload.to_vec());
683            }
684            Ok(())
685        })?;
686        Ok(found)
687    }
688
689    pub fn get_realtime_usec(&self) -> Result<u64> {
690        if let Some(metadata) = self.row.metadata() {
691            return Ok(metadata.realtime);
692        }
693        self.inner
694            .with(|fields| fields.reader.get_realtime_usec(fields.file))
695            .map_err(Into::into)
696    }
697
698    pub fn get_seqnum(&self) -> Result<(u64, [u8; 16])> {
699        let key = self.current_directory_entry_key()?;
700        Ok((key.seqnum, key.seqnum_id))
701    }
702
703    pub fn get_monotonic_usec(&self) -> Result<(u64, [u8; 16])> {
704        let key = self.current_directory_entry_key()?;
705        Ok((key.monotonic, key.boot_id))
706    }
707
708    pub fn get_cursor(&self) -> Result<String> {
709        if let Some(metadata) = self.row.metadata() {
710            return Ok(format_cursor_from_key(key_from_metadata(metadata)));
711        }
712        let seqnum_id = self.header_snapshot.header.seqnum_id;
713        self.inner
714            .with(|fields| build_cursor(fields.file, fields.reader, seqnum_id))
715    }
716
717    fn current_directory_entry_key(&self) -> Result<DirectoryEntryKey> {
718        if let Some(metadata) = self.row.metadata() {
719            return Ok(key_from_metadata(metadata));
720        }
721        self.inner.with(|fields| {
722            let offset = fields.reader.get_entry_offset()?;
723            let entry = fields.file.entry_ref(offset)?;
724            Ok(DirectoryEntryKey {
725                seqnum_id: self.header_snapshot.header.seqnum_id,
726                seqnum: entry.header.seqnum,
727                boot_id: entry.header.boot_id,
728                monotonic: entry.header.monotonic,
729                realtime: entry.header.realtime,
730                xor_hash: entry.header.xor_hash,
731            })
732        })
733    }
734
735    pub fn test_cursor(&self, cursor: &str) -> Result<bool> {
736        let current = match self.get_cursor() {
737            Ok(cursor) => cursor,
738            Err(SdkError::Journal(JournalError::UnsetCursor)) => return Err(SdkError::NoEntry),
739            Err(err) => return Err(err),
740        };
741        let current = parse::parse_cursor_location(&current, false)
742            .map_err(|err| SdkError::InvalidCursor(err.to_string()))?;
743        let Ok(want) = parse::parse_cursor_location(cursor, false) else {
744            return Ok(false);
745        };
746        Ok(parse::cursor_location_matches(&current, &want))
747    }
748
749    pub fn add_match(&mut self, data: &[u8]) {
750        self.inner.with_reader_mut(|reader| reader.add_match(data));
751    }
752
753    pub fn add_conjunction(&mut self) -> Result<()> {
754        self.inner
755            .with_mut(|fields| fields.reader.add_conjunction(fields.file))
756            .map_err(Into::into)
757    }
758
759    pub fn add_disjunction(&mut self) -> Result<()> {
760        self.inner
761            .with_mut(|fields| fields.reader.add_disjunction(fields.file))
762            .map_err(Into::into)
763    }
764
765    pub fn flush_matches(&mut self) {
766        self.inner.with_reader_mut(|reader| reader.flush_matches());
767    }
768}
769
770impl FileReader {
771    fn header_realtime_start(&self) -> u64 {
772        self.header_snapshot.header.head_entry_realtime
773    }
774
775    pub fn enumerate_fields(&mut self) -> Result<Vec<String>> {
776        self.invalidate_entry_data_state();
777        match self.enumerate_fields_indexed() {
778            Ok(fields) => Ok(fields),
779            Err(_) => enumerate_file_fields_by_scan(self),
780        }
781    }
782
783    pub(crate) fn enumerate_fields_indexed(&mut self) -> Result<Vec<String>> {
784        self.invalidate_entry_data_state();
785        self.inner.with_file(enumerate_file_fields_indexed)
786    }
787
788    pub fn query_unique(&mut self, field_name: &str) -> Result<Vec<Vec<u8>>> {
789        let mut out = Vec::new();
790        self.visit_unique_values(field_name, |value| {
791            out.push(value.to_vec());
792            Ok(())
793        })?;
794        Ok(out)
795    }
796
797    pub fn visit_unique_values<F>(&mut self, field_name: &str, visitor: F) -> Result<()>
798    where
799        F: FnMut(&[u8]) -> Result<()>,
800    {
801        self.invalidate_entry_data_state();
802        let decompressed = self.row.decompressed_mut();
803        self.inner.with_file(|file| {
804            visit_file_unique_values_indexed(file, field_name.as_bytes(), decompressed, visitor)
805        })
806    }
807
808    pub fn query_unique_state(&mut self, field_name: &str) -> Result<()> {
809        self.invalidate_entry_data_state();
810        self.inner.with_mut(|fields| {
811            fields
812                .reader
813                .field_data_query_unique(fields.file, field_name.as_bytes())
814        })?;
815        Ok(())
816    }
817
818    pub fn restart_unique_state(&mut self) {
819        self.inner
820            .with_reader_mut(|reader| reader.field_data_restart());
821    }
822
823    pub fn clear_unique_state(&mut self) {
824        self.inner
825            .with_reader_mut(|reader| reader.field_data_clear());
826    }
827
828    pub fn enumerate_unique_payload(&mut self, field_name: &str) -> Result<Option<Vec<u8>>> {
829        let field = field_name.as_bytes();
830        let decompressed = self.row.decompressed_mut();
831        self.inner.with_mut(|fields| {
832            let Some(data) = fields.reader.field_data_enumerate(fields.file)? else {
833                return Ok(None);
834            };
835            let payload = if data.is_compressed() {
836                decompressed.clear();
837                let len = data.decompress(decompressed)?;
838                &decompressed[..len]
839            } else {
840                data.raw_payload()
841            };
842            let Some(_) = payload
843                .strip_prefix(field)
844                .and_then(|rest| rest.strip_prefix(b"="))
845            else {
846                return Err(SdkError::VerificationError(
847                    "field DATA chain object does not match requested field".to_string(),
848                ));
849            };
850            Ok(Some(payload.to_vec()))
851        })
852    }
853}
854
855#[cfg(test)]
856mod tests;