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, pub runtime_state: RuntimeState, pub payload_json: Value, pub previous_hash: Option, pub hash: String, } #[derive(Clone, Debug, PartialEq)] pub struct PendingEventRecord { pub event_type: EventType, pub runtime_state: RuntimeState, pub commitment_id: Option, pub payload_json: Value, } #[derive(Debug)] pub struct EventLog { path: PathBuf, next_sequence: u64, previous_hash: Option, needs_parent_sync_after_append: bool, } impl EventLog { pub fn open(path: impl AsRef) -> Result { 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, payload_json: Value, ) -> Result { 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, ) -> Result> { let events: Vec = 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> { 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)> { 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> { 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 { 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(); } }