nordisk/holger

server/lib/src/audit/export.rs

//! **Hand the audit records out as Arrow** (lane HOL-AUDIT, 2026-10-30): the
//! records an [`AuditFilter`] selects, written as ONE Arrow IPC file in the index
//! schema ([`super::arrow_ipc::index_schema`]), so a lakehouse or any Arrow tool
//! reads them offline and answers in its own engine what holger's queries do not.
//!
//! The export is read-only and holds no writer; every row keeps its `seq` and
//! chain `hash`, so the reader can check it against `holger audit verify`.
//! gRPC `AuditService.ExportAudit` (capped at [`MAX_EXPORT_ROWS`] per call, page with
//! `since_seq`), `holger audit export --out <file>` (offline, every row).

use super::arrow_ipc::{arrow_file_bytes, schema_string, LAYOUT};
use super::query::AuditFilter;
use super::{AuditError, QuerySource};

/// The most rows one gRPC export carries.
pub const MAX_EXPORT_ROWS: usize = 50_000;

/// One export.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ExportQuery {
    pub filter: AuditFilter,
    /// Only rows with `seq >= since_seq` (the next page: the last `seq` + 1).
    pub since_seq: u64,
    /// At most this many rows, the OLDEST that match; 0 = no cap.
    pub limit: usize,
}

/// The export.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Exported {
    /// One Arrow IPC file.
    pub arrow_ipc: Vec<u8>,
    pub rows: u64,
    /// Matching rows beyond `limit` exist; ask again from `next_seq`.
    pub truncated: bool,
    pub next_seq: u64,
    /// [`LAYOUT`] and the schema in one line.
    pub layout: &'static str,
    pub schema: String,
}

/// **Export the records `q` selects**, oldest first.
pub fn export(source: &QuerySource, q: &ExportQuery) -> Result<Exported, AuditError> {
    q.filter.check()?;
    let table = source.table()?;
    let mut matched = 0usize;
    let cap = if q.limit == 0 { usize::MAX } else { q.limit };
    let records = table.select(|r| {
        if r.seq < q.since_seq || !q.filter.matches(r) {
            return false;
        }
        matched += 1;
        matched <= cap
    })?;
    let next_seq = records.last().map(|r| r.seq + 1).unwrap_or(q.since_seq);
    Ok(Exported {
        arrow_ipc: arrow_file_bytes(&records)?,
        rows: records.len() as u64,
        truncated: matched > records.len(),
        next_seq,
        layout: LAYOUT,
        schema: schema_string(),
    })
}

#[cfg(test)]
mod tests {
    use super::super::arrow_ipc::read_arrow_file_bytes;
    use super::super::{AuditAction, AuditEvent, AuditLog, FileAuditLog};
    use super::*;

    /// ★ The export reads back as the records it selected, with their chain place.
    #[test]
    fn an_export_reads_back_as_the_records_it_selected() {
        let dir = std::env::temp_dir().join(format!(
            "holger-export-{}-{}",
            std::process::id(),
            super::super::now_nanos()
        ));
        let log = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(0);
        for i in 0..5u64 {
            log.record(AuditEvent::new(
                if i % 2 == 0 { "c" } else { "d" },
                AuditAction::Download,
                "r",
                format!("a@{i}"),
                "10.0.0.1:1",
                200,
                i,
            ))
            .unwrap();
        }
        let src = log.query_source().unwrap();
        let f = AuditFilter {
            ident: Some("c".into()),
            ..Default::default()
        };
        let all = export(
            &src,
            &ExportQuery {
                filter: f.clone(),
                ..Default::default()
            },
        )
        .unwrap();
        let back = read_arrow_file_bytes(&all.arrow_ipc).unwrap();
        assert_eq!(
            back.iter()
                .map(|r| r.event.artifact.as_str())
                .collect::<Vec<_>>(),
            ["a@0", "a@2", "a@4"]
        );
        assert_eq!(back.iter().map(|r| r.seq).collect::<Vec<_>>(), [0, 2, 4]);
        assert!(back.iter().all(|r| r.hash.len() == 64));
        assert_eq!((all.rows, all.truncated, all.layout), (3, false, LAYOUT));

        let first = export(
            &src,
            &ExportQuery {
                filter: f.clone(),
                since_seq: 0,
                limit: 2,
            },
        )
        .unwrap();
        assert_eq!((first.rows, first.truncated, first.next_seq), (2, true, 3));
        let rest = export(
            &src,
            &ExportQuery {
                filter: f,
                since_seq: first.next_seq,
                limit: 2,
            },
        )
        .unwrap();
        assert_eq!((rest.rows, rest.truncated), (1, false));
        let _ = std::fs::remove_dir_all(&dir);
    }
}