reddb_server/telemetry/

operator_event.rs

//! Operator-grade event bus for high-severity system conditions.
//!
//! # Operator / developer split
//!
//! RedDB telemetry has two audiences:
//!
//! - **Developer signal** (`tracing` spans at `DEBUG` / `INFO`): ephemeral,
//!   high-volume, lives in `red.log` or stdout. Helps engineers trace request
//!   flows and understand runtime behaviour during development.
//!
//! - **Operator-grade events** (this module): low-volume, high-severity
//!   conditions that a production operator *must* see and act on.
//!   Persisted to the tamper-evident audit log first so they survive process
//!   crashes; a `tracing::warn!` breadcrumb lands in the normal log channel
//!   as a secondary copy; `eprintln!` fallback ensures the event is never
//!   silently swallowed even if both sinks fail.
//!
//! `OperatorEvent::emit` always runs synchronously — it is intentionally
//! *not* async so callers in crash paths, signal handlers, and `Drop` impls
//! can call it without an async runtime.

use std::sync::{Arc, OnceLock};

use crate::runtime::audit_log::{
    AuditAuthSource, AuditEvent, AuditField, AuditFieldEscaper, AuditLogger, Outcome,
};

// ---------------------------------------------------------------------------
// Process-wide sink
// ---------------------------------------------------------------------------
//
// The OperatorEvent enum is defined in `telemetry/` but the deepest
// emit sites (storage layer, replication apply loop, signal handlers,
// drop impls) cannot thread an `&AuditLogger` reference through their
// call stacks without a sweeping refactor. To stay surgical (#205) we
// expose a process-wide sink that the runtime registers at startup and
// every emit site consults via `OperatorEvent::emit_global`.
//
// Trade-off: the sink is a `OnceLock<Arc<AuditLogger>>`, which means
// emits that fire *before* the runtime registers the logger fall back
// to `tracing::warn!` + `eprintln!` only — the audit copy is lost. The
// tamper-evident audit copy is the primary record; the breadcrumb /
// stderr fallbacks are the safety net the original `emit(&AuditLogger)`
// shape already accepted, so the degradation is the same one already
// documented in the module rustdoc.

static GLOBAL_SINK: OnceLock<Arc<AuditLogger>> = OnceLock::new();

/// Install the process-wide [`AuditLogger`] used by
/// [`OperatorEvent::emit_global`]. Called once at runtime startup; a
/// second call is a no-op (the first registration wins) so test
/// harnesses that build multiple in-memory runtimes cannot stomp on
/// each other's loggers — they fall back to tracing+eprintln.
pub fn install_global_audit_sink(logger: Arc<AuditLogger>) {
    let _ = GLOBAL_SINK.set(logger);
}

// ---------------------------------------------------------------------------
// OperatorEvent
// ---------------------------------------------------------------------------

/// High-severity system conditions that require operator attention.
///
/// Every variant carries typed [`crate::runtime::audit_log::AuditValue`]
/// fields so adversarial bytes (CRLF, NUL, quote, non-UTF-8) are
/// escape-safe at the audit boundary (ADR 0010).
#[derive(Debug)]
pub enum OperatorEvent {
    /// A replication stream to a follower/replica broke unexpectedly.
    ReplicationBroken { peer: String, reason: String },
    /// Replication state diverged: the follower's committed LSN or
    /// checksum disagrees with the leader.
    Divergence {
        peer: String,
        leader_lsn: u64,
        follower_lsn: u64,
    },
    /// The WAL fsync call failed. Data may be at risk on the current host.
    WalFsyncFailed { path: String, error: String },
    /// Available disk space fell below the configured critical threshold.
    DiskSpaceCritical {
        path: String,
        available_bytes: u64,
        threshold_bytes: u64,
    },
    /// An authentication bypass was detected (e.g. auth gate returned
    /// `allow` for a request that should have been rejected).
    AuthBypass {
        principal: String,
        resource: String,
        detail: String,
    },
    /// An admin capability was granted to a principal at runtime.
    AdminCapabilityGranted {
        granted_to: String,
        capability: String,
        granted_by: String,
    },
    /// Secret rotation failed; the current secret may be stale.
    SecretRotationFailed { secret_ref: String, error: String },
    /// A runtime configuration change was applied to a live instance.
    ConfigChanged {
        key: String,
        old_value: String,
        new_value: String,
        changed_by: String,
    },
    /// The server process failed to start cleanly.
    StartupFailed { phase: String, error: String },
    /// The server process was forced to shut down (e.g. OOM killer,
    /// SIGKILL, unrecoverable error).
    ShutdownForced { reason: String },
    /// On-disk schema metadata is corrupt or inconsistent.
    SchemaCorruption { collection: String, detail: String },
    /// A scheduled or triggered checkpoint failed to complete.
    CheckpointFailed { lsn: u64, error: String },
    /// An admin intent that was started but never reached a terminal phase
    /// (completed or aborted). Emitted by [`super::admin_intent_log::AdminIntentLog::scan_and_report`]
    /// at startup so operators can investigate interrupted operations.
    DanglingAdminIntent {
        id: crate::crypto::uuid::Uuid,
        op: crate::telemetry::admin_intent_log::IntentOp,
        /// Unix milliseconds when the intent was started.
        started_at_ms: u64,
        last_phase: crate::telemetry::admin_intent_log::IntentPhase,
    },
    /// A config-file change was detected but one or more changed fields
    /// require a full server restart to take effect. The change was NOT
    /// applied; the operator must restart to pick it up.
    ConfigChangeRequiresRestart { fields_changed: String },
    /// An ALTER TABLE on a collection with active event subscriptions
    /// added or removed columns. Downstream consumers may see a different
    /// payload shape starting at `lsn`.
    SubscriptionSchemaChange {
        collection: String,
        /// Comma-separated subscription names affected.
        subscription_names: String,
        /// Comma-separated columns added (empty string if none).
        fields_added: String,
        /// Comma-separated columns removed (empty string if none).
        fields_removed: String,
        lsn: u64,
    },
    /// An event could not be delivered to its target queue after all
    /// retry attempts, and was routed to the dead-letter queue instead.
    OutboxDlqActivated {
        queue: String,
        dlq: String,
        reason: String,
    },
    /// A queue message exhausted `max_attempts` and was promoted to its
    /// DLQ target. Forensic: data has left the normal flow and operators
    /// must be able to trace it later. Slice 10 of issue #527.
    QueueDlqPromoted {
        queue: String,
        group: String,
        dlq: String,
        message_id: u64,
        attempts: u32,
        reason: String,
    },
    /// A deposed primary auto-rolled-back its divergent tail to the common
    /// point on rejoin (issue #840, ADR 0030). The tail is, by definition,
    /// non-committed (above the commit watermark), so removing it from the
    /// live timeline is correct — but the discarded writes are **always**
    /// persisted to a rollback file so they remain auditable and
    /// reconcilable. Rollback is never silent. The `common_point_lsn` is the
    /// recover-to-LSN target; everything in `(common_point_lsn,
    /// tail_to_lsn]` was removed from the live timeline and saved to
    /// `rollback_file`. `commit_watermark` is the durable floor that was
    /// never crossed.
    DeposedPrimaryRollback {
        /// LSN the deposed primary recovered to — the common point with the
        /// new primary, the recover-to-LSN target.
        common_point_lsn: u64,
        /// Highest LSN in the discarded divergent tail (inclusive).
        tail_to_lsn: u64,
        /// Number of LSNs removed from the live timeline.
        tail_lsns: u64,
        /// The commit watermark — the durable floor. The invariant holds:
        /// `common_point_lsn >= commit_watermark`, so nothing at or below
        /// the watermark was rolled back.
        commit_watermark: u64,
        /// Path/handle of the rollback file the discarded tail was saved to.
        rollback_file: String,
        /// The new primary the node rejoins as a replica of.
        new_primary_addr: String,
        /// The term the node now follows.
        new_term: u64,
    },
    /// A durable write was rejected by the ownership admission gate below
    /// routing (ADR 0037). This is auditable fencing evidence for stale owners.
    OwnershipFenced {
        reason: String,
        ownership_epoch: u64,
        range_identity: String,
    },
    /// Replica apply rejected a stamped record because its range authority
    /// term/ownership epoch conflicts with the current ownership state.
    ReplicaAuthorityDivergence {
        range_identity: String,
        expected_term: u64,
        expected_ownership_epoch: u64,
        observed_term: u64,
        observed_ownership_epoch: u64,
        lsn: u64,
    },
}

impl OperatorEvent {
    /// All variant names as CamelCase strings, in declaration order.
    pub fn all_variant_names() -> &'static [&'static str] {
        &[
            "ReplicationBroken",
            "Divergence",
            "WalFsyncFailed",
            "DiskSpaceCritical",
            "AuthBypass",
            "AdminCapabilityGranted",
            "SecretRotationFailed",
            "ConfigChanged",
            "StartupFailed",
            "ShutdownForced",
            "SchemaCorruption",
            "CheckpointFailed",
            "DanglingAdminIntent",
            "ConfigChangeRequiresRestart",
            "SubscriptionSchemaChange",
            "OutboxDlqActivated",
            "QueueDlqPromoted",
            "DeposedPrimaryRollback",
            "OwnershipFenced",
            "ReplicaAuthorityDivergence",
        ]
    }

    /// Return the CamelCase variant name for routing table lookup.
    pub(super) fn variant_name(&self) -> &'static str {
        match self {
            Self::ReplicationBroken { .. } => "ReplicationBroken",
            Self::Divergence { .. } => "Divergence",
            Self::WalFsyncFailed { .. } => "WalFsyncFailed",
            Self::DiskSpaceCritical { .. } => "DiskSpaceCritical",
            Self::AuthBypass { .. } => "AuthBypass",
            Self::AdminCapabilityGranted { .. } => "AdminCapabilityGranted",
            Self::SecretRotationFailed { .. } => "SecretRotationFailed",
            Self::ConfigChanged { .. } => "ConfigChanged",
            Self::StartupFailed { .. } => "StartupFailed",
            Self::ShutdownForced { .. } => "ShutdownForced",
            Self::SchemaCorruption { .. } => "SchemaCorruption",
            Self::CheckpointFailed { .. } => "CheckpointFailed",
            Self::DanglingAdminIntent { .. } => "DanglingAdminIntent",
            Self::ConfigChangeRequiresRestart { .. } => "ConfigChangeRequiresRestart",
            Self::SubscriptionSchemaChange { .. } => "SubscriptionSchemaChange",
            Self::OutboxDlqActivated { .. } => "OutboxDlqActivated",
            Self::QueueDlqPromoted { .. } => "QueueDlqPromoted",
            Self::DeposedPrimaryRollback { .. } => "DeposedPrimaryRollback",
            Self::OwnershipFenced { .. } => "OwnershipFenced",
            Self::ReplicaAuthorityDivergence { .. } => "ReplicaAuthorityDivergence",
        }
    }

    /// Emit the event.
    ///
    /// Execution order (per issue #202):
    /// 1. Persist to `audit` — tamper-evident, durable.
    /// 2. `tracing::warn!` breadcrumb — lands in `red.log` / stderr.
    /// 3. `eprintln!` fallback — fires only if the audit write fails,
    ///    ensuring the event is never silently lost.
    ///
    /// Emit the event using the process-wide sink installed by the
    /// runtime at startup. When no sink is installed (early boot,
    /// tests without an audit logger), the tracing breadcrumb and
    /// eprintln fallback still fire so the event is never silently
    /// lost.
    pub fn emit_global(self) {
        match GLOBAL_SINK.get() {
            Some(logger) => self.emit(logger.as_ref()),
            None => {
                let (_, _, summary) = self.decompose();
                tracing::warn!(target: "reddb::operator", "{summary}");
                eprintln!("[reddb::operator] (no audit sink) {summary}");
            }
        }
    }

    pub fn emit(self, audit: &AuditLogger) {
        let (action, fields, summary) = self.decompose();

        let ev = AuditEvent::builder(action)
            .source(AuditAuthSource::System)
            .outcome(Outcome::Error)
            .fields(fields)
            .build();

        // 1. Audit log (primary, tamper-evident).
        let audit_ok = {
            // `record_event` is infallible from the caller's perspective
            // (it falls back to sync write internally), so we treat it as
            // always succeeding for the breadcrumb decision.
            audit.record_event(ev);
            true
        };

        // 2. tracing breadcrumb.
        tracing::warn!(target: "reddb::operator", "{summary}");

        // 3. eprintln fallback — guard against silent loss when audit
        //    is unhealthy (e.g. disk full, writer thread dead).
        if !audit_ok {
            eprintln!("[reddb::operator] {summary}");
        }
    }

    /// Decompose `self` into `(action, audit_fields, human_summary)`.
    pub(super) fn decompose(self) -> (&'static str, Vec<AuditField>, String) {
        match self {
            Self::ReplicationBroken { peer, reason } => {
                let summary = format!("replication broken: peer={peer} reason={reason}");
                let fields = vec![
                    AuditFieldEscaper::field("peer", peer),
                    AuditFieldEscaper::field("reason", reason),
                ];
                ("operator/replication_broken", fields, summary)
            }
            Self::Divergence {
                peer,
                leader_lsn,
                follower_lsn,
            } => {
                let summary = format!(
                    "replication divergence: peer={peer} leader_lsn={leader_lsn} follower_lsn={follower_lsn}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("peer", peer),
                    AuditFieldEscaper::field("leader_lsn", leader_lsn),
                    AuditFieldEscaper::field("follower_lsn", follower_lsn),
                ];
                ("operator/divergence", fields, summary)
            }
            Self::WalFsyncFailed { path, error } => {
                let summary = format!("WAL fsync failed: path={path} error={error}");
                let fields = vec![
                    AuditFieldEscaper::field("path", path),
                    AuditFieldEscaper::field("error", error),
                ];
                ("operator/wal_fsync_failed", fields, summary)
            }
            Self::DiskSpaceCritical {
                path,
                available_bytes,
                threshold_bytes,
            } => {
                let summary = format!(
                    "disk space critical: path={path} available={available_bytes} threshold={threshold_bytes}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("path", path),
                    AuditFieldEscaper::field("available_bytes", available_bytes),
                    AuditFieldEscaper::field("threshold_bytes", threshold_bytes),
                ];
                ("operator/disk_space_critical", fields, summary)
            }
            Self::AuthBypass {
                principal,
                resource,
                detail,
            } => {
                let summary =
                    format!("auth bypass detected: principal={principal} resource={resource}");
                let fields = vec![
                    AuditFieldEscaper::field("principal", principal),
                    AuditFieldEscaper::field("resource", resource),
                    AuditFieldEscaper::field("detail", detail),
                ];
                ("operator/auth_bypass", fields, summary)
            }
            Self::AdminCapabilityGranted {
                granted_to,
                capability,
                granted_by,
            } => {
                let summary = format!(
                    "admin capability granted: to={granted_to} capability={capability} by={granted_by}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("granted_to", granted_to),
                    AuditFieldEscaper::field("capability", capability),
                    AuditFieldEscaper::field("granted_by", granted_by),
                ];
                ("operator/admin_capability_granted", fields, summary)
            }
            Self::SecretRotationFailed { secret_ref, error } => {
                let summary = format!("secret rotation failed: ref={secret_ref} error={error}");
                let fields = vec![
                    AuditFieldEscaper::field("secret_ref", secret_ref),
                    AuditFieldEscaper::field("error", error),
                ];
                ("operator/secret_rotation_failed", fields, summary)
            }
            Self::ConfigChanged {
                key,
                old_value,
                new_value,
                changed_by,
            } => {
                let summary = format!(
                    "config changed: key={key} old={old_value} new={new_value} by={changed_by}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("key", key),
                    AuditFieldEscaper::field("old_value", old_value),
                    AuditFieldEscaper::field("new_value", new_value),
                    AuditFieldEscaper::field("changed_by", changed_by),
                ];
                ("operator/config_changed", fields, summary)
            }
            Self::StartupFailed { phase, error } => {
                let summary = format!("startup failed: phase={phase} error={error}");
                let fields = vec![
                    AuditFieldEscaper::field("phase", phase),
                    AuditFieldEscaper::field("error", error),
                ];
                ("operator/startup_failed", fields, summary)
            }
            Self::ShutdownForced { reason } => {
                let summary = format!("shutdown forced: reason={reason}");
                let fields = vec![AuditFieldEscaper::field("reason", reason)];
                ("operator/shutdown_forced", fields, summary)
            }
            Self::SchemaCorruption { collection, detail } => {
                let summary = format!("schema corruption: collection={collection} detail={detail}");
                let fields = vec![
                    AuditFieldEscaper::field("collection", collection),
                    AuditFieldEscaper::field("detail", detail),
                ];
                ("operator/schema_corruption", fields, summary)
            }
            Self::CheckpointFailed { lsn, error } => {
                let summary = format!("checkpoint failed: lsn={lsn} error={error}");
                let fields = vec![
                    AuditFieldEscaper::field("lsn", lsn),
                    AuditFieldEscaper::field("error", error),
                ];
                ("operator/checkpoint_failed", fields, summary)
            }
            Self::DanglingAdminIntent {
                id,
                op,
                started_at_ms,
                last_phase,
            } => {
                let summary = format!(
                    "dangling admin intent: id={id} op={op} started_at_ms={started_at_ms} last_phase={last_phase}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("id", id.to_string()),
                    AuditFieldEscaper::field("op", op.to_string()),
                    AuditFieldEscaper::field("started_at_ms", started_at_ms),
                    AuditFieldEscaper::field("last_phase", last_phase.to_string()),
                ];
                ("operator/dangling_admin_intent", fields, summary)
            }
            Self::ConfigChangeRequiresRestart { fields_changed } => {
                let summary = format!("config change requires restart: fields={fields_changed}");
                let fields = vec![AuditFieldEscaper::field("fields_changed", fields_changed)];
                ("operator/config_change_requires_restart", fields, summary)
            }
            Self::SubscriptionSchemaChange {
                collection,
                subscription_names,
                fields_added,
                fields_removed,
                lsn,
            } => {
                let summary = format!(
                    "subscription schema change: collection={collection} subscriptions=[{subscription_names}] added=[{fields_added}] removed=[{fields_removed}] lsn={lsn}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("collection", collection),
                    AuditFieldEscaper::field("subscription_names", subscription_names),
                    AuditFieldEscaper::field("fields_added", fields_added),
                    AuditFieldEscaper::field("fields_removed", fields_removed),
                    AuditFieldEscaper::field("lsn", lsn),
                ];
                ("operator/subscription_schema_change", fields, summary)
            }
            Self::OutboxDlqActivated { queue, dlq, reason } => {
                let summary =
                    format!("outbox DLQ activated: queue={queue} dlq={dlq} reason={reason}");
                let fields = vec![
                    AuditFieldEscaper::field("queue", queue),
                    AuditFieldEscaper::field("dlq", dlq),
                    AuditFieldEscaper::field("reason", reason),
                ];
                ("operator/outbox_dlq_activated", fields, summary)
            }
            Self::QueueDlqPromoted {
                queue,
                group,
                dlq,
                message_id,
                attempts,
                reason,
            } => {
                let summary = format!(
                    "queue DLQ promoted: queue={queue} group={group} dlq={dlq} message_id={message_id} attempts={attempts} reason={reason}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("queue", queue),
                    AuditFieldEscaper::field("group", group),
                    AuditFieldEscaper::field("dlq", dlq),
                    AuditFieldEscaper::field("message_id", message_id),
                    AuditFieldEscaper::field("attempts", attempts as u64),
                    AuditFieldEscaper::field("reason", reason),
                ];
                ("operator/queue_dlq_promoted", fields, summary)
            }
            Self::DeposedPrimaryRollback {
                common_point_lsn,
                tail_to_lsn,
                tail_lsns,
                commit_watermark,
                rollback_file,
                new_primary_addr,
                new_term,
            } => {
                let summary = format!(
                    "deposed primary auto-rollback: recovered to common_point_lsn={common_point_lsn} \
                     (commit_watermark={commit_watermark}); discarded divergent tail of {tail_lsns} LSN(s) \
                     up to {tail_to_lsn} saved to {rollback_file}; rejoining as replica of \
                     {new_primary_addr} under term {new_term}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("common_point_lsn", common_point_lsn),
                    AuditFieldEscaper::field("tail_to_lsn", tail_to_lsn),
                    AuditFieldEscaper::field("tail_lsns", tail_lsns),
                    AuditFieldEscaper::field("commit_watermark", commit_watermark),
                    AuditFieldEscaper::field("rollback_file", rollback_file),
                    AuditFieldEscaper::field("new_primary_addr", new_primary_addr),
                    AuditFieldEscaper::field("new_term", new_term),
                ];
                ("operator/deposed_primary_rollback", fields, summary)
            }
            Self::OwnershipFenced {
                reason,
                ownership_epoch,
                range_identity,
            } => {
                let summary = format!(
                    "ownership fenced: reason={reason} ownership_epoch={ownership_epoch} range={range_identity}"
                );
                let fields = vec![
                    AuditFieldEscaper::field("reason", reason),
                    AuditFieldEscaper::field("ownership_epoch", ownership_epoch),
                    AuditFieldEscaper::field("range_identity", range_identity),
                ];
                ("operator/ownership_fenced", fields, summary)
            }
            Self::ReplicaAuthorityDivergence {
                range_identity,
                expected_term,
                expected_ownership_epoch,
                observed_term,
                observed_ownership_epoch,
                lsn,
            } => {
                let summary = format!(
                    "replica authority divergence: range={range_identity} lsn={lsn} \
                     expected=({expected_term},{expected_ownership_epoch}) \
                     observed=({observed_term},{observed_ownership_epoch})"
                );
                let fields = vec![
                    AuditFieldEscaper::field("range_identity", range_identity),
                    AuditFieldEscaper::field("expected_term", expected_term),
                    AuditFieldEscaper::field("expected_ownership_epoch", expected_ownership_epoch),
                    AuditFieldEscaper::field("observed_term", observed_term),
                    AuditFieldEscaper::field("observed_ownership_epoch", observed_ownership_epoch),
                    AuditFieldEscaper::field("lsn", lsn),
                ];
                ("operator/replica_authority_divergence", fields, summary)
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;
    use crate::runtime::audit_log::AuditLogger;

    struct TempAuditPath(std::path::PathBuf);

    impl std::ops::Deref for TempAuditPath {
        type Target = std::path::PathBuf;

        fn deref(&self) -> &std::path::PathBuf {
            &self.0
        }
    }

    impl AsRef<std::path::Path> for TempAuditPath {
        fn as_ref(&self) -> &std::path::Path {
            &self.0
        }
    }

    impl Drop for TempAuditPath {
        fn drop(&mut self) {
            if let Some(parent) = self.0.parent() {
                let _ = std::fs::remove_dir_all(parent);
            }
        }
    }

    fn make_logger() -> (AuditLogger, TempAuditPath) {
        let dir = std::env::temp_dir().join(format!(
            "reddb-op-event-{}-{}",
            std::process::id(),
            crate::utils::now_unix_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join(".audit.log");
        let logger = AuditLogger::with_path(path.clone());
        (logger, TempAuditPath(path))
    }

    fn drain(logger: &AuditLogger) {
        assert!(
            logger.wait_idle(Duration::from_secs(2)),
            "audit logger drain timed out"
        );
    }

    fn read_last_line(path: &std::path::Path) -> crate::json::Value {
        let body = std::fs::read_to_string(path).unwrap();
        let line = body.lines().last().expect("at least one audit line");
        crate::json::from_str(line).expect("valid JSON")
    }

    // ------------------------------------------------------------------
    // One test per variant — verifies action string + a representative field
    // ------------------------------------------------------------------

    #[test]
    fn replication_broken_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::ReplicationBroken {
            peer: "replica-1".into(),
            reason: "TCP reset".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/replication_broken")
        );
        let peer = v
            .get("detail")
            .and_then(|d| d.get("peer"))
            .and_then(|x| x.as_str());
        assert_eq!(peer, Some("replica-1"));
    }

    #[test]
    fn divergence_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::Divergence {
            peer: "replica-2".into(),
            leader_lsn: 1000,
            follower_lsn: 999,
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/divergence")
        );
        let lsn = v
            .get("detail")
            .and_then(|d| d.get("leader_lsn"))
            .and_then(|x| x.as_i64());
        assert_eq!(lsn, Some(1000));
    }

    #[test]
    fn wal_fsync_failed_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::WalFsyncFailed {
            path: "/data/wal".into(),
            error: "EIO".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/wal_fsync_failed")
        );
        let err = v
            .get("detail")
            .and_then(|d| d.get("error"))
            .and_then(|x| x.as_str());
        assert_eq!(err, Some("EIO"));
    }

    #[test]
    fn disk_space_critical_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::DiskSpaceCritical {
            path: "/data".into(),
            available_bytes: 1024,
            threshold_bytes: 10240,
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/disk_space_critical")
        );
        let avail = v
            .get("detail")
            .and_then(|d| d.get("available_bytes"))
            .and_then(|x| x.as_i64());
        assert_eq!(avail, Some(1024));
    }

    #[test]
    fn auth_bypass_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::AuthBypass {
            principal: "alice".into(),
            resource: "/admin/drop".into(),
            detail: "scope check skipped".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/auth_bypass")
        );
        let res = v
            .get("detail")
            .and_then(|d| d.get("resource"))
            .and_then(|x| x.as_str());
        assert_eq!(res, Some("/admin/drop"));
    }

    #[test]
    fn admin_capability_granted_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::AdminCapabilityGranted {
            granted_to: "bob".into(),
            capability: "ADMIN_WRITE".into(),
            granted_by: "root".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/admin_capability_granted")
        );
        let cap = v
            .get("detail")
            .and_then(|d| d.get("capability"))
            .and_then(|x| x.as_str());
        assert_eq!(cap, Some("ADMIN_WRITE"));
    }

    #[test]
    fn secret_rotation_failed_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::SecretRotationFailed {
            secret_ref: "jwt-signing-key".into(),
            error: "HSM unreachable".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/secret_rotation_failed")
        );
        let r = v
            .get("detail")
            .and_then(|d| d.get("secret_ref"))
            .and_then(|x| x.as_str());
        assert_eq!(r, Some("jwt-signing-key"));
    }

    #[test]
    fn config_changed_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::ConfigChanged {
            key: "max_connections".into(),
            old_value: "100".into(),
            new_value: "200".into(),
            changed_by: "ops-bot".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/config_changed")
        );
        let nv = v
            .get("detail")
            .and_then(|d| d.get("new_value"))
            .and_then(|x| x.as_str());
        assert_eq!(nv, Some("200"));
    }

    #[test]
    fn startup_failed_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::StartupFailed {
            phase: "wal_recovery".into(),
            error: "corrupt frame".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/startup_failed")
        );
        let phase = v
            .get("detail")
            .and_then(|d| d.get("phase"))
            .and_then(|x| x.as_str());
        assert_eq!(phase, Some("wal_recovery"));
    }

    #[test]
    fn shutdown_forced_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::ShutdownForced {
            reason: "OOM".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/shutdown_forced")
        );
        let r = v
            .get("detail")
            .and_then(|d| d.get("reason"))
            .and_then(|x| x.as_str());
        assert_eq!(r, Some("OOM"));
    }

    #[test]
    fn schema_corruption_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::SchemaCorruption {
            collection: "users".into(),
            detail: "unknown type tag 0xFF".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/schema_corruption")
        );
        let coll = v
            .get("detail")
            .and_then(|d| d.get("collection"))
            .and_then(|x| x.as_str());
        assert_eq!(coll, Some("users"));
    }

    #[test]
    fn checkpoint_failed_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::CheckpointFailed {
            lsn: 42_000,
            error: "write stall".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/checkpoint_failed")
        );
        let lsn = v
            .get("detail")
            .and_then(|d| d.get("lsn"))
            .and_then(|x| x.as_i64());
        assert_eq!(lsn, Some(42_000));
    }

    #[test]
    fn deposed_primary_rollback_emits() {
        let (logger, path) = make_logger();
        OperatorEvent::DeposedPrimaryRollback {
            common_point_lsn: 200,
            tail_to_lsn: 230,
            tail_lsns: 30,
            commit_watermark: 200,
            rollback_file: "/data/rollback/term7-lsn200-230.rbk".into(),
            new_primary_addr: "http://node-b:55055".into(),
            new_term: 8,
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/deposed_primary_rollback")
        );
        let cp = v
            .get("detail")
            .and_then(|d| d.get("common_point_lsn"))
            .and_then(|x| x.as_i64());
        assert_eq!(cp, Some(200));
        let file = v
            .get("detail")
            .and_then(|d| d.get("rollback_file"))
            .and_then(|x| x.as_str());
        assert_eq!(file, Some("/data/rollback/term7-lsn200-230.rbk"));
    }

    #[test]
    fn ownership_fenced_emits_reason_epoch_and_range() {
        let (logger, path) = make_logger();
        OperatorEvent::OwnershipFenced {
            reason: "stale_ownership".into(),
            ownership_epoch: 2,
            range_identity: "system.global/0".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/ownership_fenced")
        );
        let detail = v.get("detail").expect("event detail");
        assert_eq!(
            detail.get("reason").and_then(|x| x.as_str()),
            Some("stale_ownership")
        );
        assert_eq!(
            detail.get("ownership_epoch").and_then(|x| x.as_i64()),
            Some(2)
        );
        assert_eq!(
            detail.get("range_identity").and_then(|x| x.as_str()),
            Some("system.global/0")
        );
    }

    #[test]
    fn replica_authority_divergence_emits_range_and_term_epoch_pairs() {
        let (logger, path) = make_logger();
        OperatorEvent::ReplicaAuthorityDivergence {
            range_identity: "system.global/0".into(),
            expected_term: 8,
            expected_ownership_epoch: 2,
            observed_term: 7,
            observed_ownership_epoch: 1,
            lsn: 42,
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(
            v.get("action").and_then(|x| x.as_str()),
            Some("operator/replica_authority_divergence")
        );
        let detail = v.get("detail").expect("event detail");
        assert_eq!(
            detail.get("range_identity").and_then(|x| x.as_str()),
            Some("system.global/0")
        );
        assert_eq!(
            detail.get("expected_term").and_then(|x| x.as_i64()),
            Some(8)
        );
        assert_eq!(
            detail
                .get("expected_ownership_epoch")
                .and_then(|x| x.as_i64()),
            Some(2)
        );
        assert_eq!(
            detail.get("observed_term").and_then(|x| x.as_i64()),
            Some(7)
        );
        assert_eq!(
            detail
                .get("observed_ownership_epoch")
                .and_then(|x| x.as_i64()),
            Some(1)
        );
        assert_eq!(detail.get("lsn").and_then(|x| x.as_i64()), Some(42));
    }

    // ------------------------------------------------------------------
    // Adversarial corpus: CRLF / NUL / quote / non-UTF-8-ish in fields
    // ------------------------------------------------------------------

    #[test]
    fn adversarial_fields_are_escape_safe() {
        let payloads: &[(&str, &str)] = &[
            ("crlf", "line1\r\nline2"),
            ("nul", "before\0after"),
            ("quote", r#"she said "hi""#),
            ("json_inject", r#"{"injected":true}"#),
            ("low_ctrl", "\x01\x02\x07\x1f"),
            ("backslash", "C:\\path\\file"),
            ("mixed", "name=\"x\"\n\\path\t\x01end"),
        ];

        for (label, payload) in payloads {
            let (logger, path) = make_logger();
            OperatorEvent::SchemaCorruption {
                collection: payload.to_string(),
                detail: payload.to_string(),
            }
            .emit(&logger);
            drain(&logger);

            let body = std::fs::read_to_string(&path).unwrap();
            let line = body.lines().last().unwrap_or("");

            // Single JSONL row — no embedded newline.
            assert!(
                !line.contains('\n'),
                "{label}: embedded newline in JSONL row"
            );

            let v: crate::json::Value = crate::json::from_str(line)
                .unwrap_or_else(|e| panic!("{label}: audit line not valid JSON: {e}\n{line:?}"));
            let recovered = v
                .get("detail")
                .and_then(|d| d.get("collection"))
                .and_then(|x| x.as_str())
                .unwrap_or("");
            assert_eq!(recovered, *payload, "{label}: round-trip mismatch");
        }
    }

    // ------------------------------------------------------------------
    // Outcome is always Error; source is always System
    // ------------------------------------------------------------------

    #[test]
    fn emit_sets_outcome_error_and_source_system() {
        let (logger, path) = make_logger();
        OperatorEvent::ShutdownForced {
            reason: "test".into(),
        }
        .emit(&logger);
        drain(&logger);
        let v = read_last_line(&path);
        assert_eq!(v.get("outcome").and_then(|x| x.as_str()), Some("error"));
        assert_eq!(v.get("source").and_then(|x| x.as_str()), Some("system"));
    }
}