antidrift/legacy/src/event_log.rs

use crate::domain::{unix_secs_now, RuntimeState, POLICY_SCHEMA_VERSION};
use anyhow::{anyhow, Context, Result};
use fs2::FileExt;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventType {
    CommitmentCreated,
    RuntimeTransition,
    CommitmentTransition,
    PolicyApplied,
    EvidenceObserved,
    ViolationObserved,
    TransitionStarted,
    ReviewCompleted,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EventRecord {
    pub schema_version: u16,
    pub sequence: u64,
    pub timestamp_unix_secs: u64,
    pub event_type: EventType,
    pub commitment_id: Option<String>,
    pub runtime_state: RuntimeState,
    pub payload_json: Value,
    pub previous_hash: Option<String>,
    pub hash: String,
}

#[derive(Clone, Debug, PartialEq)]
pub struct PendingEventRecord {
    pub event_type: EventType,
    pub runtime_state: RuntimeState,
    pub commitment_id: Option<String>,
    pub payload_json: Value,
}

#[derive(Debug)]
pub struct EventLog {
    path: PathBuf,
    next_sequence: u64,
    previous_hash: Option<String>,
    needs_parent_sync_after_append: bool,
}

impl EventLog {
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref().to_path_buf();
        create_parent_dirs(&path)?;

        let (mut file, created) = open_log_file(&path)?;
        file.lock_shared()
            .with_context(|| format!("lock event log for read {}", path.display()))?;
        let (next_sequence, previous_hash) = load_tail_locked(&mut file, &path)?;

        Ok(Self {
            path,
            next_sequence,
            previous_hash,
            needs_parent_sync_after_append: created,
        })
    }

    pub fn append(
        &mut self,
        event_type: EventType,
        runtime_state: RuntimeState,
        commitment_id: Option<String>,
        payload_json: Value,
    ) -> Result<EventRecord> {
        let mut records = self.append_batch([PendingEventRecord {
            event_type,
            runtime_state,
            commitment_id,
            payload_json,
        }])?;
        records
            .pop()
            .ok_or_else(|| anyhow!("single event append produced no record"))
    }

    pub fn append_batch(
        &mut self,
        events: impl IntoIterator<Item = PendingEventRecord>,
    ) -> Result<Vec<EventRecord>> {
        let events: Vec<PendingEventRecord> = events.into_iter().collect();
        if events.is_empty() {
            return Ok(Vec::new());
        }

        create_parent_dirs(&self.path)?;

        let existed = self
            .path
            .try_exists()
            .with_context(|| format!("check whether event log exists {}", self.path.display()))?;
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .append(true)
            .open(&self.path)
            .with_context(|| format!("open event log for append {}", self.path.display()))?;
        file.lock_exclusive()
            .with_context(|| format!("lock event log {}", self.path.display()))?;

        let (next_sequence, previous_hash) = load_tail_locked(&mut file, &self.path)?;

        let mut records = Vec::with_capacity(events.len());
        let mut sequence = next_sequence;
        let mut record_previous_hash = previous_hash;
        for event in events {
            let mut record = EventRecord {
                schema_version: POLICY_SCHEMA_VERSION,
                sequence,
                timestamp_unix_secs: unix_secs_now(),
                event_type: event.event_type,
                commitment_id: event.commitment_id,
                runtime_state: event.runtime_state,
                payload_json: event.payload_json,
                previous_hash: record_previous_hash,
                hash: String::new(),
            };
            record.hash = record_hash(&record)?;
            record_previous_hash = Some(record.hash.clone());
            sequence += 1;
            records.push(record);
        }

        let mut bytes = Vec::new();
        for record in &records {
            serde_json::to_writer(&mut bytes, record).with_context(|| {
                format!("serialize event log record to {}", self.path.display())
            })?;
            bytes.push(b'\n');
        }

        file.write_all(&bytes).with_context(|| {
            format!(
                "write {} event log records to {}",
                records.len(),
                self.path.display()
            )
        })?;
        file.flush()
            .with_context(|| format!("flush event log {}", self.path.display()))?;
        file.sync_data()
            .with_context(|| format!("sync event log {}", self.path.display()))?;
        if !existed || self.needs_parent_sync_after_append {
            sync_parent_dir(&self.path)?;
            self.needs_parent_sync_after_append = false;
        }

        self.next_sequence = sequence;
        self.previous_hash = record_previous_hash;

        Ok(records)
    }

    pub fn records(&self) -> Result<Vec<EventRecord>> {
        let mut file = OpenOptions::new()
            .read(true)
            .open(&self.path)
            .with_context(|| format!("open event log for read {}", self.path.display()))?;
        file.lock_shared()
            .with_context(|| format!("lock event log for read {}", self.path.display()))?;
        load_records_locked(&mut file, &self.path)
    }
}

fn create_parent_dirs(path: &Path) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).with_context(|| {
            format!("create parent directories for event log {}", path.display())
        })?;
    }
    Ok(())
}

fn open_log_file(path: &Path) -> Result<(File, bool)> {
    let existed = path
        .try_exists()
        .with_context(|| format!("check whether event log exists {}", path.display()))?;
    let file = OpenOptions::new()
        .create(true)
        .read(true)
        .append(true)
        .open(path)
        .with_context(|| format!("open event log {}", path.display()))?;
    Ok((file, !existed))
}

fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option<String>)> {
    let records = load_records_locked(file, path)?;
    let next_sequence = records.last().map_or(1, |record| record.sequence + 1);
    let previous_hash = records.last().map(|record| record.hash.clone());
    Ok((next_sequence, previous_hash))
}

fn load_records_locked(file: &mut File, path: &Path) -> Result<Vec<EventRecord>> {
    let mut next_sequence = 1;
    let mut previous_hash = None;
    let mut records = Vec::new();

    file.seek(SeekFrom::Start(0))
        .with_context(|| format!("seek event log {}", path.display()))?;
    for (line_index, line) in BufReader::new(file).lines().enumerate() {
        let line_number = line_index + 1;
        let line = line.with_context(|| {
            format!("read event log line {line_number} from {}", path.display())
        })?;
        if line.trim().is_empty() {
            continue;
        }

        let record: EventRecord = serde_json::from_str(&line).with_context(|| {
            format!("parse event log line {line_number} from {}", path.display())
        })?;
        validate_record(&record, next_sequence, previous_hash.as_deref())
            .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?;

        next_sequence += 1;
        previous_hash = Some(record.hash.clone());
        records.push(record);
    }

    Ok(records)
}

#[cfg(unix)]
fn sync_parent_dir(path: &Path) -> Result<()> {
    if let Some(parent) = path.parent() {
        let dir = File::open(parent)
            .with_context(|| format!("open parent directory for event log {}", path.display()))?;
        dir.sync_all().with_context(|| {
            format!(
                "sync parent directory {} for event log {}",
                parent.display(),
                path.display()
            )
        })?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn sync_parent_dir(_path: &Path) -> Result<()> {
    // Directory fsync is not exposed portably by std on all supported platforms.
    // Linux and other Unix targets take the durable path above.
    Ok(())
}

fn validate_record(
    record: &EventRecord,
    expected_sequence: u64,
    expected_previous_hash: Option<&str>,
) -> Result<()> {
    if record.schema_version != POLICY_SCHEMA_VERSION {
        return Err(anyhow!(
            "unsupported schema version at sequence {}: expected {}, found {}",
            record.sequence,
            POLICY_SCHEMA_VERSION,
            record.schema_version
        ));
    }
    if record.sequence != expected_sequence {
        return Err(anyhow!(
            "sequence mismatch: expected {}, found {}",
            expected_sequence,
            record.sequence
        ));
    }
    if record.previous_hash.as_deref() != expected_previous_hash {
        return Err(anyhow!(
            "previous hash mismatch at sequence {}",
            record.sequence
        ));
    }

    let expected_hash = record_hash(record)?;
    if record.hash != expected_hash {
        return Err(anyhow!("hash mismatch at sequence {}", record.sequence));
    }

    Ok(())
}

fn record_hash(record: &EventRecord) -> Result<String> {
    let mut hashable = record.clone();
    hashable.hash.clear();
    let bytes = serde_json::to_vec(&hashable).context("serialize event log record for hash")?;
    Ok(hex::encode(Sha256::digest(bytes)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::RuntimeState;
    use serde_json::json;
    use std::fs;
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn temp_log_path(name: &str) -> PathBuf {
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir().join(format!(
            "antidrift-event-log-{name}-{}-{nonce}.jsonl",
            std::process::id()
        ))
    }

    #[test]
    fn appends_hash_chained_jsonl_records() {
        let path = temp_log_path("append");
        let mut log = EventLog::open(&path).unwrap();

        let first = log
            .append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                Some("commitment-1".to_string()),
                json!({"from": "active", "to": "transition"}),
            )
            .unwrap();
        let second = log
            .append(
                EventType::PolicyApplied,
                RuntimeState::Active,
                Some("commitment-1".to_string()),
                json!({"policy_id": "policy-1"}),
            )
            .unwrap();

        assert_eq!(first.sequence, 1);
        assert_eq!(second.sequence, 2);
        assert_eq!(second.previous_hash.as_deref(), Some(first.hash.as_str()));
        assert_ne!(first.hash, second.hash);

        let contents = fs::read_to_string(&path).unwrap();
        assert_eq!(contents.lines().count(), 2);

        fs::remove_file(path).unwrap();
    }

    #[test]
    fn batch_appends_hash_chained_adjacent_records() {
        let path = temp_log_path("batch");
        let mut log = EventLog::open(&path).unwrap();

        let records = log
            .append_batch([
                PendingEventRecord {
                    event_type: EventType::CommitmentCreated,
                    runtime_state: RuntimeState::Active,
                    commitment_id: Some("commitment-1".to_string()),
                    payload_json: json!({"commitment_id": "commitment-1"}),
                },
                PendingEventRecord {
                    event_type: EventType::PolicyApplied,
                    runtime_state: RuntimeState::Active,
                    commitment_id: Some("commitment-1".to_string()),
                    payload_json: json!({"policy_id": "policy-1"}),
                },
            ])
            .unwrap();

        assert_eq!(records.len(), 2);
        assert_eq!(records[0].sequence, 1);
        assert_eq!(records[1].sequence, 2);
        assert_eq!(
            records[1].previous_hash.as_deref(),
            Some(records[0].hash.as_str())
        );

        let contents = fs::read_to_string(&path).unwrap();
        assert_eq!(contents.lines().count(), 2);
        EventLog::open(&path).unwrap();

        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_existing_log_validates_chain_and_continues_sequence() {
        let path = temp_log_path("reopen");
        {
            let mut log = EventLog::open(&path).unwrap();
            log.append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                None,
                json!({"from": "active"}),
            )
            .unwrap();
        }

        let mut reopened = EventLog::open(&path).unwrap();
        let next = reopened
            .append(
                EventType::PolicyApplied,
                RuntimeState::Active,
                None,
                json!({"policy_id": "policy-1"}),
            )
            .unwrap();

        assert_eq!(next.sequence, 2);

        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_missing_log_creates_reopenable_file() {
        let path = temp_log_path("open-missing");

        EventLog::open(&path).unwrap();

        assert!(path.exists());
        EventLog::open(&path).unwrap();
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_malformed_json_log_fails() {
        let path = temp_log_path("malformed");
        fs::write(&path, "{not json}\n").unwrap();

        let err = EventLog::open(&path).expect_err("malformed JSONL must fail");

        assert!(err.to_string().contains("parse event log"));
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_tampered_log_chain_fails() {
        let path = temp_log_path("tampered");
        {
            let mut log = EventLog::open(&path).unwrap();
            log.append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                None,
                json!({"from": "active"}),
            )
            .unwrap();
            log.append(
                EventType::PolicyApplied,
                RuntimeState::Active,
                None,
                json!({"policy_id": "policy-1"}),
            )
            .unwrap();
        }
        let contents = fs::read_to_string(&path).unwrap();
        fs::write(&path, contents.replace("\"sequence\":2", "\"sequence\":3")).unwrap();

        let err = EventLog::open(&path).expect_err("tampered chain must fail");

        assert!(err.to_string().contains("sequence mismatch"));
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_log_with_unknown_record_field_fails() {
        let path = temp_log_path("unknown-field");
        {
            let mut log = EventLog::open(&path).unwrap();
            log.append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                None,
                json!({"from": "active"}),
            )
            .unwrap();
        }
        let contents = fs::read_to_string(&path).unwrap();
        let mut record_json: Value =
            serde_json::from_str(contents.lines().next().unwrap()).unwrap();
        record_json["unexpected_field"] = json!("ignored-before-hashing");
        fs::write(
            &path,
            format!("{}\n", serde_json::to_string(&record_json).unwrap()),
        )
        .unwrap();

        let err = EventLog::open(&path).expect_err("unknown record fields must fail");

        assert!(err.to_string().contains("parse event log"));
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn opening_log_with_unsupported_schema_version_fails() {
        let path = temp_log_path("unsupported-schema");
        let mut record = EventRecord {
            schema_version: 999,
            sequence: 1,
            timestamp_unix_secs: unix_secs_now(),
            event_type: EventType::RuntimeTransition,
            commitment_id: None,
            runtime_state: RuntimeState::Transition,
            payload_json: json!({"from": "active"}),
            previous_hash: None,
            hash: String::new(),
        };
        record.hash = record_hash(&record).unwrap();
        fs::write(
            &path,
            format!("{}\n", serde_json::to_string(&record).unwrap()),
        )
        .unwrap();

        let err = EventLog::open(&path).expect_err("unsupported schema version must fail");

        assert!(err.to_string().contains("unsupported schema version"));
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn stale_handle_reloads_tail_before_append() {
        let path = temp_log_path("stale-handle");
        let mut log1 = EventLog::open(&path).unwrap();
        let mut log2 = EventLog::open(&path).unwrap();

        let first = log1
            .append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                None,
                json!({"from": "active"}),
            )
            .unwrap();
        let second = log2
            .append(
                EventType::PolicyApplied,
                RuntimeState::Active,
                None,
                json!({"policy_id": "policy-1"}),
            )
            .unwrap();

        assert_eq!(second.sequence, 2);
        assert_eq!(second.previous_hash.as_deref(), Some(first.hash.as_str()));
        EventLog::open(&path).unwrap();
        fs::remove_file(path).unwrap();
    }

    #[test]
    fn append_creates_parent_dirs_and_writes_reopenable_file() {
        let path = temp_log_path("append-parents")
            .with_file_name("nested")
            .join("events.jsonl");
        let mut log = EventLog {
            path: path.clone(),
            next_sequence: 1,
            previous_hash: None,
            needs_parent_sync_after_append: false,
        };

        let record = log
            .append(
                EventType::RuntimeTransition,
                RuntimeState::Transition,
                None,
                json!({"from": "active"}),
            )
            .unwrap();

        assert_eq!(record.sequence, 1);
        EventLog::open(&path).unwrap();
        fs::remove_file(&path).unwrap();
        fs::remove_dir(path.parent().unwrap()).unwrap();
    }
}