razeri_plugin_api/

event.rs

// Copyright (C) 2026 Gabriel Shea
// Licensed under either of the Apache License, Version 2.0 or the MIT
// license, at your option.

#![allow(clippy::doc_markdown)]

//! Structured event sink trait and fan-out registry.
//!
//! [`EventSink`] is the extension point for custom observability
//! backends (Kafka, Datadog, Prometheus, etc.).
//! [`EventRegistry`] dispatches events to all registered sinks,
//! with error resilience (one failing sink does not halt dispatch).

use crate::error::EventSinkError;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::sync::{Arc, RwLock};
use thiserror::Error;

// ── Event types ────────────────────────────────────────────────────────

/// Stable monitoring event envelope emitted by Razeri operations.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[non_exhaustive]
pub struct RazeriEvent {
    event_id: String,
    timestamp: DateTime<Utc>,
    category: EventCategory,
    severity: EventSeverity,
    workflow: Option<WorkflowContext>,
    subject: EventSubject,
    payload: EventPayload,
}

/// High-level event category used for filtering and aggregation.
#[derive(Debug, Clone, Copy, Hash, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum EventCategory {
    /// Data profiling events.
    Profiling,
    /// SQL materialization events.
    Materialization,
    /// Design loading and validation events.
    Design,
    /// Plugin loading and registration events.
    Plugin,
    /// Storage I/O events.
    Storage,
    /// AI and LLM request events.
    Ai,
}

/// Operator-facing severity level for a monitoring event.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum EventSeverity {
    /// Informational event — no operator action required.
    Info,
    /// Something unexpected occurred but processing continued.
    Warning,
    /// An operation failed; operator attention is needed.
    Error,
    /// A severe failure requiring immediate attention.
    Critical,
}

/// Correlation context for workflow-scoped operations.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[non_exhaustive]
pub struct WorkflowContext {
    /// Stable workflow identifier shared across related events.
    pub workflow_id: String,
    /// Optional request identifier from an API or RPC boundary.
    pub request_id: Option<String>,
    /// Optional actor name such as a user, service account, or agent.
    pub actor: Option<String>,
    /// Optional originating surface such as `cli`, `api`, `web`, or `mcp`.
    pub surface: Option<String>,
}

impl WorkflowContext {
    /// Create a workflow context with the required workflow identifier.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn new(workflow_id: impl Into<String>) -> Result<Self, EventValidationError> {
        let workflow_id = workflow_id.into();
        if workflow_id.trim().is_empty() {
            return Err(EventValidationError::WorkflowIdRequired);
        }

        Ok(Self { workflow_id, request_id: None, actor: None, surface: None })
    }

    /// Attach a request identifier.
    #[must_use]
    pub fn with_request_id(mut self, request_id: impl Into<String>) -> Self {
        self.request_id = Some(request_id.into());
        self
    }

    /// Attach an actor label.
    #[must_use]
    pub fn with_actor(mut self, actor: impl Into<String>) -> Self {
        self.actor = Some(actor.into());
        self
    }

    /// Attach an originating surface label.
    #[must_use]
    pub fn with_surface(mut self, surface: impl Into<String>) -> Self {
        self.surface = Some(surface.into());
        self
    }
}

/// Primary subject for filtering and grouping events.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[non_exhaustive]
pub struct EventSubject {
    /// Discriminator classifying the type of subject.
    pub kind: EventSubjectKind,
    /// Subject name (source name, entity name, plugin name, etc.).
    pub name: String,
}

impl EventSubject {
    /// Create a new event subject.
    pub fn new(kind: EventSubjectKind, name: impl Into<String>) -> Self {
        Self { kind, name: name.into() }
    }

    /// Subject representing a source dataset or file.
    pub fn source(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::Source, name)
    }

    /// Subject representing an entity.
    pub fn entity(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::Entity, name)
    }

    /// Subject representing a design.
    pub fn design(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::Design, name)
    }

    /// Subject representing a plugin.
    pub fn plugin(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::Plugin, name)
    }

    /// Subject representing a storage path or backend object.
    pub fn storage(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::StoragePath, name)
    }

    /// Subject representing an LLM request target.
    pub fn llm_model(name: impl Into<String>) -> Self {
        Self::new(EventSubjectKind::LlmModel, name)
    }
}

/// Subject kind used for filtering and aggregation.
#[derive(Debug, Clone, Copy, Hash, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum EventSubjectKind {
    /// A raw data source.
    Source,
    /// A design entity.
    Entity,
    /// A design document.
    Design,
    /// A plugin.
    Plugin,
    /// A storage path or backend object.
    StoragePath,
    /// An LLM model target.
    LlmModel,
}

/// Structured payload variants carried by the event envelope.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum EventPayload {
    /// Data profiling started on a source.
    ProfilingStarted {
        /// Name of the data source being profiled.
        source: String,
    },
    /// Data profiling completed.
    ProfilingCompleted {
        /// Name of the data source that was profiled.
        source: String,
        /// Number of fields discovered.
        fields: usize,
        /// Number of rows sampled.
        rows: usize,
        /// Duration of the profiling run in milliseconds.
        duration_ms: u64,
    },
    /// SQL materialization started for an entity.
    MaterializationStarted {
        /// Name of the entity being materialized.
        entity: String,
        /// Target SQL platform name.
        platform: String,
    },
    /// SQL materialization completed.
    MaterializationCompleted {
        /// Name of the entity that was materialized.
        entity: String,
        /// Target SQL platform name.
        platform: String,
        /// Number of SQL artifacts produced.
        artifacts: usize,
        /// Duration of the materialization run in milliseconds.
        duration_ms: u64,
    },
    /// SQL materialization failed.
    MaterializationFailed {
        /// Name of the design that failed.
        design: String,
        /// Target SQL platform name.
        platform: String,
        /// Human-readable failure description.
        message: String,
    },
    /// Design validated with warning and error counts.
    DesignValidated {
        /// Name of the validated design.
        design: String,
        /// Number of validation errors.
        errors: usize,
        /// Number of validation warnings.
        warnings: usize,
    },
    /// Design document loaded.
    DesignLoaded {
        /// File path or URI from which the design was loaded.
        path: String,
        /// Number of entities in the design.
        entities: usize,
    },
    /// A plugin was loaded.
    PluginLoaded {
        /// Plugin name.
        name: String,
        /// Plugin extension point type (e.g. `"dialect"`).
        plugin_type: String,
        /// Source description (e.g. `"static"` or a library path).
        source: String,
    },
    /// Storage I/O operation completed.
    StorageOperationCompleted {
        /// Operation name (e.g. `"put"`, `"get"`, `"delete"`).
        operation: String,
        /// Storage path of the object.
        path: String,
        /// Number of bytes transferred.
        bytes: u64,
        /// Duration of the operation in milliseconds.
        duration_ms: u64,
    },
    /// Materializer detected a storage feature that the target platform cannot honor.
    /// Emitted instead of panicking (TP-010).
    StorageWarning {
        /// Name of the entity that triggered the warning.
        entity: String,
        /// Storage feature that is unsupported.
        feature: String,
        /// Target SQL platform name.
        platform: String,
        /// Suggested alternative or workaround.
        recommendation: String,
    },
    /// LLM request completed.
    LlmRequestCompleted {
        /// LLM provider name.
        provider: String,
        /// Model identifier used for the request.
        model: String,
        /// Total tokens consumed.
        tokens: u64,
        /// Duration of the request in milliseconds.
        duration_ms: u64,
    },
    /// A file the agent found was moved to where a routing rule sends it.
    FileRouted {
        /// Where the file was found.
        path: String,
        /// Where it was moved to.
        destination: String,
        /// The routing rule that sent it there.
        rule: String,
        /// Its size in bytes.
        bytes: u64,
    },
    /// A file the agent found was set aside: no rule routes it, or it failed
    /// a check. A warning, so a person looks at it.
    FileSetAside {
        /// Where the file was found.
        path: String,
        /// Where it was set aside.
        destination: String,
        /// Why, in a reader's words.
        reason: String,
    },
    /// A file the agent found could not be moved at all, and stays where it
    /// was. An error.
    FileNotMoved {
        /// Where the file is.
        path: String,
        /// What stopped the move, in a reader's words.
        reason: String,
    },
    /// The agent finished one pass over a landing location.
    ScanCompleted {
        /// The location scanned.
        location: String,
        /// Files found.
        files: usize,
        /// Of those, routed.
        routed: usize,
        /// Of those, set aside.
        set_aside: usize,
        /// Of those, not moved.
        not_moved: usize,
        /// How long the pass took, in milliseconds.
        duration_ms: u64,
    },
}

impl EventPayload {
    /// Category derived from the payload variant.
    #[must_use]
    pub fn category(&self) -> EventCategory {
        match self {
            Self::ProfilingStarted { .. } | Self::ProfilingCompleted { .. } => {
                EventCategory::Profiling
            }
            Self::MaterializationStarted { .. } | Self::MaterializationCompleted { .. } => {
                EventCategory::Materialization
            }
            Self::MaterializationFailed { .. } => EventCategory::Materialization,
            Self::DesignValidated { .. } | Self::DesignLoaded { .. } => EventCategory::Design,
            Self::PluginLoaded { .. } => EventCategory::Plugin,
            Self::StorageOperationCompleted { .. } | Self::StorageWarning { .. } => {
                EventCategory::Storage
            }
            Self::LlmRequestCompleted { .. } => EventCategory::Ai,
            Self::FileRouted { .. }
            | Self::FileSetAside { .. }
            | Self::FileNotMoved { .. }
            | Self::ScanCompleted { .. } => EventCategory::Storage,
        }
    }

    /// Severity derived from the payload variant.
    #[must_use]
    pub fn severity(&self) -> EventSeverity {
        match self {
            Self::MaterializationFailed { .. } | Self::FileNotMoved { .. } => EventSeverity::Error,
            Self::DesignValidated { errors, .. } if *errors > 0 => EventSeverity::Error,
            Self::DesignValidated { warnings, .. } if *warnings > 0 => EventSeverity::Warning,
            Self::StorageWarning { .. } | Self::FileSetAside { .. } => EventSeverity::Warning,
            _ => EventSeverity::Info,
        }
    }

    /// Whether this payload must be correlated to a workflow.
    #[must_use]
    pub fn requires_workflow_context(&self) -> bool {
        match self {
            Self::ProfilingStarted { .. }
            | Self::ProfilingCompleted { .. }
            | Self::MaterializationStarted { .. }
            | Self::MaterializationCompleted { .. }
            | Self::MaterializationFailed { .. }
            | Self::DesignValidated { .. }
            | Self::StorageOperationCompleted { .. }
            | Self::LlmRequestCompleted { .. } => true,
            Self::DesignLoaded { .. }
            | Self::PluginLoaded { .. }
            | Self::StorageWarning { .. }
            | Self::FileRouted { .. }
            | Self::FileSetAside { .. }
            | Self::FileNotMoved { .. }
            | Self::ScanCompleted { .. } => false,
        }
    }

    pub(crate) fn name(&self) -> &'static str {
        match self {
            Self::ProfilingStarted { .. } => "profiling_started",
            Self::ProfilingCompleted { .. } => "profiling_completed",
            Self::MaterializationStarted { .. } => "materialization_started",
            Self::MaterializationCompleted { .. } => "materialization_completed",
            Self::MaterializationFailed { .. } => "materialization_failed",
            Self::DesignValidated { .. } => "design_validated",
            Self::DesignLoaded { .. } => "design_loaded",
            Self::PluginLoaded { .. } => "plugin_loaded",
            Self::StorageOperationCompleted { .. } => "storage_operation_completed",
            Self::StorageWarning { .. } => "storage_warning",
            Self::LlmRequestCompleted { .. } => "llm_request_completed",
            Self::FileRouted { .. } => "file_routed",
            Self::FileSetAside { .. } => "file_set_aside",
            Self::FileNotMoved { .. } => "file_not_moved",
            Self::ScanCompleted { .. } => "scan_completed",
        }
    }

    fn summary(&self) -> EventSummary {
        match self {
            Self::ProfilingStarted { source } => profiling_started_summary(source),
            Self::ProfilingCompleted { source, fields, rows, duration_ms } => {
                profiling_completed_summary(source, *fields, *rows, *duration_ms)
            }
            Self::MaterializationStarted { entity, platform } => {
                materialization_started_summary(entity, platform)
            }
            Self::MaterializationCompleted { entity, platform, artifacts, duration_ms } => {
                materialization_completed_summary(entity, platform, *artifacts, *duration_ms)
            }
            Self::MaterializationFailed { design, platform, message } => {
                materialization_failed_summary(design, platform, message)
            }
            Self::DesignValidated { design, errors, warnings } => {
                design_validated_summary(design, *errors, *warnings)
            }
            Self::DesignLoaded { path, entities } => design_loaded_summary(path, *entities),
            Self::PluginLoaded { name, plugin_type, source } => {
                plugin_loaded_summary(name, plugin_type, source)
            }
            Self::StorageOperationCompleted { operation, path, bytes, duration_ms } => {
                storage_operation_completed_summary(operation, path, *bytes, *duration_ms)
            }
            Self::StorageWarning { entity, feature, platform, recommendation } => {
                storage_warning_summary(entity, feature, platform, recommendation)
            }
            Self::LlmRequestCompleted { provider, model, tokens, duration_ms } => {
                llm_request_completed_summary(provider, model, *tokens, *duration_ms)
            }
            Self::FileRouted { path, destination, rule, bytes } => EventSummary {
                title: "File routed".to_string(),
                detail: format!(
                    "'{path}' moved to '{destination}' by rule '{rule}', {}",
                    counted(*bytes, "byte", "bytes")
                ),
            },
            Self::FileSetAside { path, destination, reason } => EventSummary {
                title: "File set aside".to_string(),
                detail: format!("'{path}' set aside in '{destination}': {reason}"),
            },
            Self::FileNotMoved { path, reason } => EventSummary {
                title: "File not moved".to_string(),
                detail: format!("'{path}' was not moved: {reason}"),
            },
            Self::ScanCompleted { location, files, routed, set_aside, not_moved, duration_ms } => {
                EventSummary {
                    title: "Scan completed".to_string(),
                    detail: format!(
                        "'{location}' held {}: {routed} routed, {set_aside} set aside, \
                         {not_moved} not moved, in {duration_ms} ms",
                        counted(*files as u64, "file", "files")
                    ),
                }
            }
        }
    }
}

/// **`n` and its noun**, the noun said once for one: `1 byte`, `2 bytes`.
fn counted(n: u64, one: &str, many: &str) -> String {
    format!("{n} {}", if n == 1 { one } else { many })
}

/// Build summary text for a profiling-started payload.
/// Parameters:
/// - source: Source name being profiled.
fn profiling_started_summary(source: &str) -> EventSummary {
    EventSummary {
        title: "Profiling started".to_string(),
        detail: format!("Profiling started for source '{source}'"),
    }
}

/// Build summary text for a profiling-completed payload.
/// Parameters:
/// - source: Source name that was profiled.
/// - fields: Number of discovered fields.
/// - rows: Number of sampled rows.
/// - duration_ms: Profiling runtime in milliseconds.
fn profiling_completed_summary(
    source: &str,
    fields: usize,
    rows: usize,
    duration_ms: u64,
) -> EventSummary {
    EventSummary {
        title: "Profiling completed".to_string(),
        detail: format!(
            "Source '{source}' profiled with {}, {} in {duration_ms} ms",
            counted(fields as u64, "field", "fields"),
            counted(rows as u64, "row", "rows"),
        ),
    }
}

/// Build summary text for a materialization-started payload.
/// Parameters:
/// - entity: Entity being materialized.
/// - platform: Target SQL platform.
fn materialization_started_summary(entity: &str, platform: &str) -> EventSummary {
    EventSummary {
        title: "Materialization started".to_string(),
        detail: format!("Entity '{entity}' started materialization for {platform}"),
    }
}

/// Build summary text for a materialization-completed payload.
/// Parameters:
/// - entity: Entity that finished materialization.
/// - platform: Target SQL platform.
/// - artifacts: Number of generated artifacts.
/// - duration_ms: Materialization runtime in milliseconds.
fn materialization_completed_summary(
    entity: &str,
    platform: &str,
    artifacts: usize,
    duration_ms: u64,
) -> EventSummary {
    EventSummary {
        title: "Materialization completed".to_string(),
        detail: format!(
            "Entity '{entity}' produced {} for {platform} in {duration_ms} ms",
            counted(artifacts as u64, "artifact", "artifacts"),
        ),
    }
}

/// Build summary text for a materialization-failed payload.
/// Parameters:
/// - design: Design that failed.
/// - platform: Target SQL platform.
/// - message: Failure reason.
fn materialization_failed_summary(design: &str, platform: &str, message: &str) -> EventSummary {
    EventSummary {
        title: "Materialization failed".to_string(),
        detail: format!("Design '{design}' failed materialization for {platform}: {message}"),
    }
}

/// Build summary text for a design-validated payload.
/// Parameters:
/// - design: Design name.
/// - errors: Validation error count.
/// - warnings: Validation warning count.
fn design_validated_summary(design: &str, errors: usize, warnings: usize) -> EventSummary {
    EventSummary {
        title: "Design validated".to_string(),
        detail: format!(
            "Design '{design}' validated with {} and {}",
            counted(errors as u64, "error", "errors"),
            counted(warnings as u64, "warning", "warnings"),
        ),
    }
}

/// Build summary text for a design-loaded payload.
/// Parameters:
/// - path: Loaded path or URI.
/// - entities: Number of entities in the loaded design.
fn design_loaded_summary(path: &str, entities: usize) -> EventSummary {
    EventSummary {
        title: "Design loaded".to_string(),
        detail: format!(
            "Loaded design from '{path}' with {}",
            counted(entities as u64, "entity", "entities")
        ),
    }
}

/// Build summary text for a plugin-loaded payload.
/// Parameters:
/// - name: Plugin name.
/// - plugin_type: Plugin extension-point type.
/// - source: Plugin load source descriptor.
fn plugin_loaded_summary(name: &str, plugin_type: &str, source: &str) -> EventSummary {
    EventSummary {
        title: "Plugin loaded".to_string(),
        detail: format!("Loaded {plugin_type} plugin '{name}' from {source}"),
    }
}

/// Build summary text for a storage-operation-completed payload.
/// Parameters:
/// - operation: Storage operation name.
/// - path: Object path processed by the operation.
/// - bytes: Number of processed bytes.
/// - duration_ms: Operation runtime in milliseconds.
fn storage_operation_completed_summary(
    operation: &str,
    path: &str,
    bytes: u64,
    duration_ms: u64,
) -> EventSummary {
    EventSummary {
        title: "Storage operation completed".to_string(),
        detail: format!(
            "Operation '{operation}' processed {} at '{path}' in {duration_ms} ms",
            counted(bytes, "byte", "bytes"),
        ),
    }
}

/// Build summary text for a storage-warning payload.
/// Parameters:
/// - entity: Entity associated with the warning.
/// - feature: Unsupported storage feature.
/// - platform: Target SQL platform.
/// - recommendation: Suggested workaround.
fn storage_warning_summary(
    entity: &str,
    feature: &str,
    platform: &str,
    recommendation: &str,
) -> EventSummary {
    EventSummary {
        title: "Storage warning".to_string(),
        detail: format!(
            "Entity '{entity}': feature '{feature}' unavailable on '{platform}'; {recommendation}"
        ),
    }
}

/// Build summary text for an LLM-request-completed payload.
/// Parameters:
/// - provider: LLM provider name.
/// - model: Model identifier.
/// - tokens: Total consumed tokens.
/// - duration_ms: Request runtime in milliseconds.
fn llm_request_completed_summary(
    provider: &str,
    model: &str,
    tokens: u64,
    duration_ms: u64,
) -> EventSummary {
    EventSummary {
        title: "LLM request completed".to_string(),
        detail: format!(
            "Provider '{provider}' model '{model}' used {} in {duration_ms} ms",
            counted(tokens, "token", "tokens"),
        ),
    }
}

/// Filter keys that can be derived from a canonical event alone.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[non_exhaustive]
pub struct EventFilterKeys {
    /// Event category derived from the payload.
    pub category: EventCategory,
    /// Event severity derived from the payload.
    pub severity: EventSeverity,
    /// Workflow identifier if the event was workflow-scoped.
    pub workflow_id: Option<String>,
    /// Kind of the primary event subject.
    pub subject_kind: EventSubjectKind,
    /// Name of the primary event subject.
    pub subject_name: String,
    /// Actor label from the workflow context.
    pub actor: Option<String>,
    /// Originating surface label from the workflow context.
    pub surface: Option<String>,
}

/// Human-readable summary derived from a canonical event.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[non_exhaustive]
pub struct EventSummary {
    /// Short summary title.
    pub title: String,
    /// Longer human-readable detail string.
    pub detail: String,
}

/// Validation error for canonical monitoring events.
#[derive(Debug, Error, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum EventValidationError {
    /// A workflow-scoped payload was submitted without a workflow context.
    #[error("workflow-scoped event payload '{payload}' requires workflow context")]
    WorkflowContextRequired {
        /// Name of the payload type that requires workflow context.
        payload: String,
    },
    /// The workflow context was supplied with an empty `workflow_id`.
    #[error("workflow context requires a non-empty workflow_id")]
    WorkflowIdRequired,
}

impl RazeriEvent {
    /// Create a new non-workflow-scoped event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn new(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        subject: EventSubject,
        payload: EventPayload,
    ) -> Result<Self, EventValidationError> {
        Self::build(event_id.into(), timestamp, None, subject, payload)
    }

    /// Create a new workflow-scoped event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn new_with_workflow(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        subject: EventSubject,
        payload: EventPayload,
    ) -> Result<Self, EventValidationError> {
        Self::build(event_id.into(), timestamp, Some(workflow), subject, payload)
    }

    fn build(
        event_id: String,
        timestamp: DateTime<Utc>,
        workflow: Option<WorkflowContext>,
        subject: EventSubject,
        payload: EventPayload,
    ) -> Result<Self, EventValidationError> {
        let event = Self {
            event_id,
            timestamp,
            category: payload.category(),
            severity: payload.severity(),
            workflow,
            subject,
            payload,
        };

        event.validate()?;
        Ok(event)
    }

    /// Validate that the canonical envelope is internally consistent.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn validate(&self) -> Result<(), EventValidationError> {
        if self.payload.requires_workflow_context() && self.workflow.is_none() {
            return Err(EventValidationError::WorkflowContextRequired {
                payload: self.payload.name().to_string(),
            });
        }

        if self.workflow.as_ref().is_some_and(|workflow| workflow.workflow_id.trim().is_empty()) {
            return Err(EventValidationError::WorkflowIdRequired);
        }

        Ok(())
    }

    /// Stable event identifier.
    #[must_use]
    pub fn event_id(&self) -> &str {
        &self.event_id
    }

    /// Event timestamp.
    #[must_use]
    pub fn timestamp(&self) -> DateTime<Utc> {
        self.timestamp
    }

    /// Derived event category.
    #[must_use]
    pub fn category(&self) -> EventCategory {
        self.category
    }

    /// Derived event severity.
    #[must_use]
    pub fn severity(&self) -> EventSeverity {
        self.severity
    }

    /// Optional workflow correlation context.
    #[must_use]
    pub fn workflow(&self) -> Option<&WorkflowContext> {
        self.workflow.as_ref()
    }

    /// Primary event subject.
    #[must_use]
    pub fn subject(&self) -> &EventSubject {
        &self.subject
    }

    /// Structured event payload.
    #[must_use]
    pub fn payload(&self) -> &EventPayload {
        &self.payload
    }

    /// Derived filter keys suitable for summaries and operator UIs.
    #[must_use]
    pub fn filter_keys(&self) -> EventFilterKeys {
        EventFilterKeys {
            category: self.category,
            severity: self.severity,
            workflow_id: self.workflow.as_ref().map(|workflow| workflow.workflow_id.clone()),
            subject_kind: self.subject.kind,
            subject_name: self.subject.name.clone(),
            actor: self.workflow.as_ref().and_then(|workflow| workflow.actor.clone()),
            surface: self.workflow.as_ref().and_then(|workflow| workflow.surface.clone()),
        }
    }

    /// Human-readable summary derived from the canonical event.
    #[must_use]
    pub fn summary(&self) -> EventSummary {
        self.payload.summary()
    }

    /// Convenience constructor for a design validation event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn design_validated(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        design: impl Into<String>,
        errors: usize,
        warnings: usize,
    ) -> Result<Self, EventValidationError> {
        let design = design.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::design(design.clone()),
            EventPayload::DesignValidated { design, errors, warnings },
        )
    }

    /// Convenience constructor for a materialization-started event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn materialization_started(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        entity: impl Into<String>,
        platform: impl Into<String>,
    ) -> Result<Self, EventValidationError> {
        let entity = entity.into();
        let platform = platform.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::entity(entity.clone()),
            EventPayload::MaterializationStarted { entity, platform },
        )
    }

    /// Convenience constructor for a materialization-completed event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn materialization_completed(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        entity: impl Into<String>,
        platform: impl Into<String>,
        artifacts: usize,
        duration_ms: u64,
    ) -> Result<Self, EventValidationError> {
        let entity = entity.into();
        let platform = platform.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::entity(entity.clone()),
            EventPayload::MaterializationCompleted { entity, platform, artifacts, duration_ms },
        )
    }

    /// Convenience constructor for a materialization-failed event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn materialization_failed(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        design: impl Into<String>,
        platform: impl Into<String>,
        message: impl Into<String>,
    ) -> Result<Self, EventValidationError> {
        let design = design.into();
        let platform = platform.into();
        let message = message.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::design(design.clone()),
            EventPayload::MaterializationFailed { design, platform, message },
        )
    }

    /// Convenience constructor for a profiling-started event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn profiling_started(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        source: impl Into<String>,
    ) -> Result<Self, EventValidationError> {
        let source = source.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::source(source.clone()),
            EventPayload::ProfilingStarted { source },
        )
    }

    /// Convenience constructor for a profiling-completed event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn profiling_completed(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        source: impl Into<String>,
        fields: usize,
        rows: usize,
        duration_ms: u64,
    ) -> Result<Self, EventValidationError> {
        let source = source.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::source(source.clone()),
            EventPayload::ProfilingCompleted { source, fields, rows, duration_ms },
        )
    }

    /// Convenience constructor for a design-loaded event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn design_loaded(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        path: impl Into<String>,
        entities: usize,
    ) -> Result<Self, EventValidationError> {
        let path = path.into();
        Self::new(
            event_id,
            timestamp,
            EventSubject::design(path.clone()),
            EventPayload::DesignLoaded { path, entities },
        )
    }

    /// Convenience constructor for a plugin-loaded event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn plugin_loaded(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        name: impl Into<String>,
        plugin_type: impl Into<String>,
        source: impl Into<String>,
    ) -> Result<Self, EventValidationError> {
        let name = name.into();
        let plugin_type = plugin_type.into();
        let source = source.into();
        Self::new(
            event_id,
            timestamp,
            EventSubject::plugin(name.clone()),
            EventPayload::PluginLoaded { name, plugin_type, source },
        )
    }

    /// Convenience constructor for a storage-operation-completed event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn storage_operation_completed(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        operation: impl Into<String>,
        path: impl Into<String>,
        bytes: u64,
        duration_ms: u64,
    ) -> Result<Self, EventValidationError> {
        let operation = operation.into();
        let path = path.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::storage(path.clone()),
            EventPayload::StorageOperationCompleted { operation, path, bytes, duration_ms },
        )
    }

    /// Convenience constructor for an LLM-request-completed event.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn llm_request_completed(
        event_id: impl Into<String>,
        timestamp: DateTime<Utc>,
        workflow: WorkflowContext,
        provider: impl Into<String>,
        model: impl Into<String>,
        tokens: u64,
        duration_ms: u64,
    ) -> Result<Self, EventValidationError> {
        let provider = provider.into();
        let model = model.into();
        Self::new_with_workflow(
            event_id,
            timestamp,
            workflow,
            EventSubject::llm_model(model.clone()),
            EventPayload::LlmRequestCompleted { provider, model, tokens, duration_ms },
        )
    }
}

/// Shared interface for components that emit canonical monitoring events.
#[async_trait]
pub trait EventEmitter: Send + Sync {
    /// Emit a canonical event into the monitoring pipeline.
    async fn emit_event(&self, event: RazeriEvent) -> Vec<(String, EventSinkError)>;
}

// ── EventSink trait ────────────────────────────────────────────────────

/// Sink that receives structured events from Razeri operations.
///
/// Implementations should be non-blocking and resilient to transient
/// failures (e.g., network issues). Razeri will continue operating
/// even if an event sink fails.
#[async_trait]
pub trait EventSink: Send + Sync {
    /// Machine-readable sink name.
    fn name(&self) -> &str;

    /// Handle an event. Should not block or panic.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    async fn on_event(&self, event: &RazeriEvent) -> Result<(), EventSinkError>;

    /// Flush any buffered events.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    async fn flush(&self) -> Result<(), EventSinkError>;
}

// ── EventRegistry ──────────────────────────────────────────────────────

/// Registry that dispatches events to all registered sinks.
///
/// [`emit`](Self::emit) hands an event to each sink **one after another, in
/// the order the sinks were registered**, and every sink has the event before
/// [`emit`](Self::emit) returns, so each sink receives every event in the
/// order it was emitted. A sink that refuses an event does not stop the sinks
/// after it: its refusal is returned with its name. A sink that is slow holds
/// up the sinks after it, so a sink that sends events over a network hands
/// them to a queue of its own and returns.
pub struct EventRegistry {
    sinks: Vec<Arc<dyn EventSink>>,
    history: RwLock<Vec<RazeriEvent>>,
}

impl EventRegistry {
    /// Create an empty registry.
    #[must_use]
    pub fn new() -> Self {
        Self { sinks: Vec::new(), history: RwLock::new(Vec::new()) }
    }

    /// Register an event sink.
    pub fn register(&mut self, sink: Arc<dyn EventSink>) {
        self.sinks.push(sink);
    }

    /// Record an event, then hand it to every sink, one after another in the
    /// order they were registered.
    ///
    /// Returns each sink's refusal with the sink's name; a refusal does not
    /// stop the sinks after it.
    pub async fn emit(&self, event: &RazeriEvent) -> Vec<(String, EventSinkError)> {
        self.record(event.clone());
        let mut errors = Vec::new();
        for sink in &self.sinks {
            if let Err(e) = sink.on_event(event).await {
                errors.push((sink.name().to_string(), e));
            }
        }
        errors
    }

    /// Record an event without dispatching it to sinks.
    pub fn record(&self, event: RazeriEvent) {
        self.history.write().unwrap_or_else(std::sync::PoisonError::into_inner).push(event);
    }

    /// Return all recorded events in chronological order.
    pub fn events(&self) -> Vec<RazeriEvent> {
        self.history.read().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
    }

    /// Return the most recent events in chronological order.
    pub fn recent_events(&self, limit: usize) -> Vec<RazeriEvent> {
        let events = self.history.read().unwrap_or_else(std::sync::PoisonError::into_inner);
        let start = events.len().saturating_sub(limit);
        events[start..].to_vec()
    }

    /// Return events associated with the provided workflow identifier.
    pub fn events_for_workflow(&self, workflow_id: &str) -> Vec<RazeriEvent> {
        self.history
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .iter()
            .filter(|event| {
                event.workflow().is_some_and(|workflow| workflow.workflow_id == workflow_id)
            })
            .cloned()
            .collect()
    }

    /// Return human-readable summaries for the most recent events.
    pub fn recent_summaries(&self, limit: usize) -> Vec<EventSummary> {
        self.recent_events(limit).into_iter().map(|event| event.summary()).collect()
    }

    /// Flush all sinks.
    pub async fn flush_all(&self) -> Vec<(String, EventSinkError)> {
        let mut errors = Vec::new();
        for sink in &self.sinks {
            if let Err(e) = sink.flush().await {
                errors.push((sink.name().to_string(), e));
            }
        }
        errors
    }

    /// The names of the sinks this registry dispatches to, in the order they
    /// were registered.
    #[must_use]
    pub fn sink_names(&self) -> Vec<String> {
        self.sinks.iter().map(|sink| sink.name().to_owned()).collect()
    }

    /// Number of registered sinks.
    pub fn len(&self) -> usize {
        self.sinks.len()
    }

    /// Whether the registry has no sinks.
    pub fn is_empty(&self) -> bool {
        self.sinks.is_empty()
    }

    /// Execute a flexible query against the recorded event history.
    pub fn query(&self, q: &EventQuery) -> Vec<RazeriEvent> {
        let history = self.history.read().unwrap_or_else(std::sync::PoisonError::into_inner);
        let iter = history.iter().filter(|e| {
            q.category.is_none_or(|c| e.category() == c)
                && q.min_severity.is_none_or(|s| e.severity() >= s)
                && q.workflow_id
                    .as_ref()
                    .is_none_or(|wf| e.workflow().is_some_and(|w| w.workflow_id == *wf))
                && q.subject_name.as_ref().is_none_or(|n| e.subject().name == *n)
                && q.subject_kind.is_none_or(|k| e.subject().kind == k)
                && q.since.is_none_or(|t| e.timestamp() >= t)
        });
        match q.limit {
            Some(n) => iter.take(n).cloned().collect(),
            None => iter.cloned().collect(),
        }
    }
}

/// Flexible event query for filtering and pagination.
///
/// All fields are optional. When `None`, no filtering is applied for that
/// dimension. Combine fields to narrow results (they are `ANDed` together).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct EventQuery {
    /// Filter by event category.
    pub category: Option<EventCategory>,
    /// Filter to events at or above this severity.
    pub min_severity: Option<EventSeverity>,
    /// Filter by workflow identifier.
    pub workflow_id: Option<String>,
    /// Filter by subject name.
    pub subject_name: Option<String>,
    /// Filter by subject kind.
    pub subject_kind: Option<EventSubjectKind>,
    /// Only include events at or after this timestamp.
    pub since: Option<DateTime<Utc>>,
    /// Maximum number of events to return.
    pub limit: Option<usize>,
}

impl Default for EventRegistry {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl EventEmitter for EventRegistry {
    async fn emit_event(&self, event: RazeriEvent) -> Vec<(String, EventSinkError)> {
        self.emit(&event).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// **A count of one says its noun once** (`fr-1019-91`): `1 byte`, never
    /// `1 bytes`, in every summary that counts.
    #[test]
    fn a_count_of_one_says_its_noun_once() {
        let one = [
            EventPayload::ProfilingCompleted {
                source: "s".into(),
                fields: 1,
                rows: 1,
                duration_ms: 1,
            },
            EventPayload::MaterializationCompleted {
                entity: "e".into(),
                platform: "p".into(),
                artifacts: 1,
                duration_ms: 1,
            },
            EventPayload::DesignValidated { design: "d".into(), errors: 1, warnings: 1 },
            EventPayload::DesignLoaded { path: "d.yaml".into(), entities: 1 },
            EventPayload::StorageOperationCompleted {
                operation: "put".into(),
                path: "p".into(),
                bytes: 1,
                duration_ms: 1,
            },
            EventPayload::LlmRequestCompleted {
                provider: "p".into(),
                model: "m".into(),
                tokens: 1,
                duration_ms: 1,
            },
        ];
        for payload in one {
            let detail = payload.summary().detail;
            for plural in
                ["1 fields", "1 rows", "1 artifacts", "1 errors", "1 warnings", "1 entities"]
            {
                assert!(!detail.contains(plural), "{detail}");
            }
            assert!(!detail.contains("1 bytes") && !detail.contains("1 tokens"), "{detail}");
        }
        let two = EventPayload::DesignLoaded { path: "d.yaml".into(), entities: 2 };
        assert!(two.summary().detail.contains("2 entities"), "{}", two.summary().detail);
    }

    /// **A file set aside is a warning, a file not moved an error**
    /// (`fr-1019-91`), so `razeri status` and an alert rule can count them;
    /// a routed file and a finished scan are information.
    #[test]
    fn a_file_set_aside_warns_and_a_file_not_moved_is_an_error() {
        let aside = EventPayload::FileSetAside {
            path: "in/x.csv".into(),
            destination: "quarantine/x.csv".into(),
            reason: "no rule routes it".into(),
        };
        let stuck = EventPayload::FileNotMoved { path: "in/y.csv".into(), reason: "denied".into() };
        let routed = EventPayload::FileRouted {
            path: "in/z.csv".into(),
            destination: "raw/z.csv".into(),
            rule: "customers".into(),
            bytes: 1,
        };
        assert_eq!(aside.severity(), EventSeverity::Warning);
        assert_eq!(stuck.severity(), EventSeverity::Error);
        assert_eq!(routed.severity(), EventSeverity::Info);
        assert_eq!(aside.category(), EventCategory::Storage);
        assert_eq!(aside.name(), "file_set_aside");
        assert!(routed.summary().detail.ends_with("1 byte"), "{}", routed.summary().detail);
    }

    fn sample_timestamp() -> DateTime<Utc> {
        DateTime::parse_from_rfc3339("2026-03-13T12:34:56Z")
            .expect("valid timestamp fixture")
            .with_timezone(&Utc)
    }

    fn sample_workflow() -> WorkflowContext {
        WorkflowContext::new("workflow-123")
            .expect("workflow id should be valid")
            .with_request_id("request-456")
            .with_actor("analyst@example.com")
            .with_surface("web")
    }

    struct CountingSink {
        event_count: AtomicUsize,
        flush_count: AtomicUsize,
    }

    impl CountingSink {
        fn new() -> Self {
            Self { event_count: AtomicUsize::new(0), flush_count: AtomicUsize::new(0) }
        }

        fn events_received(&self) -> usize {
            self.event_count.load(Ordering::SeqCst)
        }

        fn flushes(&self) -> usize {
            self.flush_count.load(Ordering::SeqCst)
        }
    }

    #[async_trait]
    impl EventSink for CountingSink {
        fn name(&self) -> &'static str {
            "counting"
        }

        async fn on_event(&self, _event: &RazeriEvent) -> Result<(), EventSinkError> {
            self.event_count.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }

        async fn flush(&self) -> Result<(), EventSinkError> {
            self.flush_count.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }

    struct FailingSink;

    #[async_trait]
    impl EventSink for FailingSink {
        fn name(&self) -> &'static str {
            "failing"
        }

        async fn on_event(&self, _event: &RazeriEvent) -> Result<(), EventSinkError> {
            Err(EventSinkError::SinkError("intentional failure".to_string()))
        }

        async fn flush(&self) -> Result<(), EventSinkError> {
            Err(EventSinkError::SinkError("intentional failure".to_string()))
        }
    }

    fn sample_event() -> RazeriEvent {
        RazeriEvent::new_with_workflow(
            "event-123",
            sample_timestamp(),
            sample_workflow(),
            EventSubject::source("test.csv"),
            EventPayload::ProfilingCompleted {
                source: "test.csv".to_string(),
                fields: 10,
                rows: 1000,
                duration_ms: 42,
            },
        )
        .expect("workflow-scoped sample event should be valid")
    }

    fn plugin_event() -> RazeriEvent {
        RazeriEvent::new(
            "event-plugin-1",
            sample_timestamp(),
            EventSubject::plugin("builtin-logger"),
            EventPayload::PluginLoaded {
                name: "builtin-logger".to_string(),
                plugin_type: "event_sink".to_string(),
                source: "builtin".to_string(),
            },
        )
        .expect("plugin events are not workflow scoped")
    }

    #[tokio::test]
    async fn event_dispatched_to_all_sinks() {
        let sink1 = Arc::new(CountingSink::new());
        let sink2 = Arc::new(CountingSink::new());

        let mut registry = EventRegistry::new();
        registry.register(sink1.clone());
        registry.register(sink2.clone());

        let errors = registry.emit(&sample_event()).await;
        assert!(errors.is_empty());
        assert_eq!(sink1.events_received(), 1);
        assert_eq!(sink2.events_received(), 1);
    }

    #[tokio::test]
    async fn sink_failure_does_not_halt_dispatch() {
        let good_sink = Arc::new(CountingSink::new());
        let bad_sink: Arc<dyn EventSink> = Arc::new(FailingSink);

        let mut registry = EventRegistry::new();
        registry.register(bad_sink);
        registry.register(good_sink.clone());

        let errors = registry.emit(&sample_event()).await;
        // Failing sink produced an error, but good sink still received the event
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0].0, "failing");
        assert_eq!(good_sink.events_received(), 1);
    }

    #[tokio::test]
    async fn flush_all_flushes_every_sink() {
        let sink1 = Arc::new(CountingSink::new());
        let sink2 = Arc::new(CountingSink::new());

        let mut registry = EventRegistry::new();
        registry.register(sink1.clone());
        registry.register(sink2.clone());

        let errors = registry.flush_all().await;
        assert!(errors.is_empty());
        assert_eq!(sink1.flushes(), 1);
        assert_eq!(sink2.flushes(), 1);
    }

    #[tokio::test]
    async fn emitted_events_are_recorded_and_queryable() {
        let registry = EventRegistry::new();
        registry.emit_event(sample_event()).await;
        registry.emit_event(plugin_event()).await;

        let recorded = registry.events();
        assert_eq!(recorded.len(), 2);

        let workflow_events = registry.events_for_workflow("workflow-123");
        assert_eq!(workflow_events.len(), 1);
        assert_eq!(workflow_events[0].category(), EventCategory::Profiling);

        let summaries = registry.recent_summaries(2);
        assert_eq!(summaries.len(), 2);
        assert_eq!(summaries[0].title, "Profiling completed");
        assert_eq!(summaries[1].title, "Plugin loaded");
    }

    #[test]
    fn materialization_failure_has_error_severity_and_summary() {
        let event = RazeriEvent::materialization_failed(
            "event-materialization-failed",
            sample_timestamp(),
            sample_workflow(),
            "sales_design",
            "duckdb",
            "dialect mismatch",
        )
        .expect("materialization failure event should be valid");

        assert_eq!(event.category(), EventCategory::Materialization);
        assert_eq!(event.severity(), EventSeverity::Error);
        assert!(event.summary().detail.contains("dialect mismatch"));
    }

    #[test]
    fn event_serializes_to_json() {
        let event = sample_event();
        let json = serde_json::to_string(&event);
        assert!(json.is_ok());
        let json_str = json.unwrap();
        assert!(json_str.contains("profiling_completed"));
        assert!(json_str.contains("test.csv"));
        assert!(json_str.contains("workflow-123"));
    }

    #[test]
    fn workflow_scoped_event_requires_workflow_context() {
        let event = RazeriEvent::new(
            "event-missing-workflow",
            sample_timestamp(),
            EventSubject::source("missing.csv"),
            EventPayload::ProfilingStarted { source: "missing.csv".to_string() },
        );

        assert_eq!(
            event,
            Err(EventValidationError::WorkflowContextRequired {
                payload: "profiling_started".to_string(),
            })
        );
    }

    #[test]
    fn event_round_trips_through_json_without_losing_meaning() {
        let event = sample_event();
        let json = serde_json::to_string(&event).expect("event should serialize");
        let round_trip: RazeriEvent =
            serde_json::from_str(&json).expect("event should deserialize");

        assert_eq!(event, round_trip);
        assert_eq!(round_trip.validate(), Ok(()));
    }

    #[test]
    fn summaries_and_filter_keys_are_derived_from_canonical_event() {
        let event = sample_event();

        let keys = event.filter_keys();
        assert_eq!(keys.category, EventCategory::Profiling);
        assert_eq!(keys.severity, EventSeverity::Info);
        assert_eq!(keys.workflow_id.as_deref(), Some("workflow-123"));
        assert_eq!(keys.subject_kind, EventSubjectKind::Source);
        assert_eq!(keys.subject_name, "test.csv");
        assert_eq!(keys.actor.as_deref(), Some("analyst@example.com"));
        assert_eq!(keys.surface.as_deref(), Some("web"));

        let summary = event.summary();
        assert_eq!(summary.title, "Profiling completed");
        assert!(summary.detail.contains("10 fields"));
        assert!(summary.detail.contains("1000 rows"));
    }

    #[test]
    fn non_workflow_scoped_events_can_be_created_without_workflow_context() {
        let event = plugin_event();

        assert_eq!(event.category(), EventCategory::Plugin);
        assert_eq!(event.severity(), EventSeverity::Info);
        assert!(event.workflow().is_none());
    }

    #[test]
    fn profiling_started_constructs_with_source_subject() {
        let event = RazeriEvent::profiling_started(
            "ps-1",
            sample_timestamp(),
            sample_workflow(),
            "customers.csv",
        )
        .unwrap();
        assert_eq!(event.category(), EventCategory::Profiling);
        assert_eq!(event.subject().kind, EventSubjectKind::Source);
        assert_eq!(event.summary().title, "Profiling started");
    }

    #[test]
    fn profiling_completed_constructs_with_stats() {
        let event = RazeriEvent::profiling_completed(
            "pc-1",
            sample_timestamp(),
            sample_workflow(),
            "customers.csv",
            10,
            5000,
            123,
        )
        .unwrap();
        assert_eq!(event.category(), EventCategory::Profiling);
        assert!(event.summary().detail.contains("10 fields"));
        assert!(event.summary().detail.contains("5000 rows"));
    }

    #[test]
    fn design_loaded_constructs_without_workflow() {
        let event =
            RazeriEvent::design_loaded("dl-1", sample_timestamp(), "designs/retail.yaml", 3)
                .unwrap();
        assert_eq!(event.category(), EventCategory::Design);
        assert!(event.workflow().is_none());
        assert!(event.summary().detail.contains("3 entities"));
    }

    #[test]
    fn plugin_loaded_constructs_without_workflow() {
        let event = RazeriEvent::plugin_loaded(
            "pl-1",
            sample_timestamp(),
            "PostgresDialect",
            "dialect",
            "inventory",
        )
        .unwrap();
        assert_eq!(event.category(), EventCategory::Plugin);
        assert_eq!(event.subject().kind, EventSubjectKind::Plugin);
        assert!(event.summary().detail.contains("dialect"));
    }

    #[test]
    fn storage_operation_completed_constructs_with_workflow() {
        let event = RazeriEvent::storage_operation_completed(
            "so-1",
            sample_timestamp(),
            sample_workflow(),
            "move",
            "/data/file.csv",
            1024,
            50,
        )
        .unwrap();
        assert_eq!(event.category(), EventCategory::Storage);
        assert_eq!(event.subject().kind, EventSubjectKind::StoragePath);
    }

    #[test]
    fn llm_request_completed_constructs_with_workflow() {
        let event = RazeriEvent::llm_request_completed(
            "lr-1",
            sample_timestamp(),
            sample_workflow(),
            "ollama",
            "llama3",
            500,
            3000,
        )
        .unwrap();
        assert_eq!(event.category(), EventCategory::Ai);
        assert_eq!(event.subject().kind, EventSubjectKind::LlmModel);
        assert!(event.summary().detail.contains("500 tokens"));
    }

    #[test]
    fn query_returns_all_events_when_empty() {
        let registry = EventRegistry::new();
        registry.record(sample_event());
        registry.record(plugin_event());

        let results = registry.query(&EventQuery::default());
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn query_filters_by_category() {
        let registry = EventRegistry::new();
        registry.record(sample_event());
        registry.record(plugin_event());

        let results = registry
            .query(&EventQuery { category: Some(EventCategory::Plugin), ..Default::default() });
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].category(), EventCategory::Plugin);
    }

    #[test]
    fn query_filters_by_min_severity() {
        let registry = EventRegistry::new();
        registry.record(sample_event()); // Info
        let error_event = RazeriEvent::materialization_failed(
            "err-1",
            sample_timestamp(),
            sample_workflow(),
            "design-a",
            "PostgreSQL",
            "connection error",
        )
        .unwrap();
        registry.record(error_event);

        let results = registry
            .query(&EventQuery { min_severity: Some(EventSeverity::Error), ..Default::default() });
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].severity(), EventSeverity::Error);
    }

    #[test]
    fn query_filters_by_workflow_id() {
        let registry = EventRegistry::new();
        registry.record(sample_event()); // workflow-123
        registry.record(plugin_event()); // no workflow

        let results = registry.query(&EventQuery {
            workflow_id: Some("workflow-123".to_string()),
            ..Default::default()
        });
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn query_respects_limit() {
        let registry = EventRegistry::new();
        registry.record(sample_event());
        registry.record(plugin_event());

        let results = registry.query(&EventQuery { limit: Some(1), ..Default::default() });
        assert_eq!(results.len(), 1);
    }

    // ── Snapshot tests ──────────────────────────────────────────────────

    #[test]
    fn snapshot_profiling_completed_json_shape() {
        let event = sample_event();
        let val: serde_json::Value = serde_json::to_value(&event).unwrap();
        let obj = val.as_object().unwrap();

        // Envelope fields
        assert!(obj.contains_key("event_id"));
        assert!(obj.contains_key("timestamp"));
        assert!(obj.contains_key("category"));
        assert!(obj.contains_key("severity"));
        assert!(obj.contains_key("workflow"));
        assert!(obj.contains_key("subject"));
        assert!(obj.contains_key("payload"));

        // Payload is internally tagged with "kind"
        let payload = obj["payload"].as_object().unwrap();
        assert_eq!(payload["kind"], "profiling_completed");
        assert!(payload.contains_key("source"));
        assert!(payload.contains_key("fields"));
        assert!(payload.contains_key("rows"));
        assert!(payload.contains_key("duration_ms"));

        // Subject has kind+name
        let subject = obj["subject"].as_object().unwrap();
        assert_eq!(subject["kind"], "source");
        assert!(subject.contains_key("name"));
    }

    #[test]
    fn snapshot_event_round_trip_preserves_identity() {
        let event = sample_event();
        let json = serde_json::to_string(&event).unwrap();
        let roundtrip: RazeriEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(roundtrip.event_id(), event.event_id());
        assert_eq!(roundtrip.category(), event.category());
        assert_eq!(roundtrip.severity(), event.severity());
        assert_eq!(roundtrip.subject().name, event.subject().name);
    }

    #[test]
    fn snapshot_materialization_failed_json_shape() {
        let event = RazeriEvent::materialization_failed(
            "snap-mat-fail",
            sample_timestamp(),
            sample_workflow(),
            "sales_design",
            "duckdb",
            "type mismatch",
        )
        .unwrap();
        let val: serde_json::Value = serde_json::to_value(&event).unwrap();
        let payload = val["payload"].as_object().unwrap();
        assert_eq!(payload["kind"], "materialization_failed");
        assert_eq!(payload["design"], "sales_design");
        assert_eq!(payload["platform"], "duckdb");
        assert_eq!(payload["message"], "type mismatch");
    }

    #[test]
    fn snapshot_plugin_loaded_json_shape() {
        let event = plugin_event();
        let val: serde_json::Value = serde_json::to_value(&event).unwrap();
        let payload = val["payload"].as_object().unwrap();
        assert_eq!(payload["kind"], "plugin_loaded");
        assert!(payload.contains_key("name"));
        assert!(payload.contains_key("plugin_type"));
        assert!(payload.contains_key("source"));
    }

    #[test]
    fn snapshot_event_summary_format_is_stable() {
        let event = sample_event();
        let summary = event.summary();
        assert_eq!(summary.title, "Profiling completed");
        assert!(summary.detail.contains("test.csv"));
        assert!(summary.detail.contains("10"));
        assert!(summary.detail.contains("100"));
    }

    #[test]
    fn snapshot_filter_keys_are_stable() {
        let event = sample_event();
        let keys = event.filter_keys();
        assert_eq!(keys.category, EventCategory::Profiling);
        assert_eq!(keys.severity, EventSeverity::Info);
        assert_eq!(keys.workflow_id.as_deref(), Some("workflow-123"));
        assert_eq!(keys.subject_name, "test.csv");
    }
}