oxirs_stream/

time_travel.rs

//! # Time-Travel Query System
//!
//! Advanced temporal query capabilities for OxiRS Stream, enabling querying
//! data at any point in time, temporal analytics, and historical state reconstruction.

use crate::event_sourcing::{EventStoreTrait, EventStream};
use crate::StreamEvent;
use anyhow::{anyhow, Result};
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// Type alias for custom filter functions to reduce complexity
pub type CustomFilterFn = Box<dyn Fn(&StreamEvent) -> bool + Send + Sync>;

/// Time-travel query configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeTravelConfig {
    /// Maximum time window for time-travel queries
    pub max_time_window_days: u32,
    /// Enable temporal indexing for faster queries
    pub enable_temporal_indexing: bool,
    /// Temporal index granularity (minutes)
    pub index_granularity_minutes: u32,
    /// Maximum concurrent time-travel queries
    pub max_concurrent_queries: usize,
    /// Query timeout in seconds
    pub query_timeout_seconds: u64,
    /// Enable result caching
    pub enable_result_caching: bool,
    /// Cache TTL in minutes
    pub cache_ttl_minutes: u32,
    /// Maximum cache size in MB
    pub max_cache_size_mb: usize,
}

impl Default for TimeTravelConfig {
    fn default() -> Self {
        Self {
            max_time_window_days: 365,
            enable_temporal_indexing: true,
            index_granularity_minutes: 60,
            max_concurrent_queries: 100,
            query_timeout_seconds: 300,
            enable_result_caching: true,
            cache_ttl_minutes: 60,
            max_cache_size_mb: 1024,
        }
    }
}

/// Time point specification for queries
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TimePoint {
    /// Specific timestamp
    Timestamp(DateTime<Utc>),
    /// Relative time from now
    RelativeTime(ChronoDuration),
    /// Event version number
    Version(u64),
    /// Event ID
    EventId(Uuid),
    /// Named snapshot
    Snapshot(String),
}

/// Time range specification for queries
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeRange {
    pub start: TimePoint,
    pub end: TimePoint,
}

/// Temporal query specification
#[derive(Debug, Clone)]
pub struct TemporalQuery {
    pub query_id: Uuid,
    pub time_point: Option<TimePoint>,
    pub time_range: Option<TimeRange>,
    pub filter: TemporalFilter,
    pub projection: TemporalProjection,
    pub ordering: TemporalOrdering,
    pub limit: Option<usize>,
}

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

impl TemporalQuery {
    /// Create a new temporal query
    pub fn new() -> Self {
        Self {
            query_id: Uuid::new_v4(),
            time_point: None,
            time_range: None,
            filter: TemporalFilter::default(),
            projection: TemporalProjection::default(),
            ordering: TemporalOrdering::default(),
            limit: None,
        }
    }

    /// Query at specific time point
    pub fn at_time(mut self, time_point: TimePoint) -> Self {
        self.time_point = Some(time_point);
        self
    }

    /// Query within time range
    pub fn in_range(mut self, time_range: TimeRange) -> Self {
        self.time_range = Some(time_range);
        self
    }

    /// Add filter
    pub fn filter(mut self, filter: TemporalFilter) -> Self {
        self.filter = filter;
        self
    }

    /// Set projection
    pub fn project(mut self, projection: TemporalProjection) -> Self {
        self.projection = projection;
        self
    }

    /// Set ordering
    pub fn order_by(mut self, ordering: TemporalOrdering) -> Self {
        self.ordering = ordering;
        self
    }

    /// Set limit
    pub fn limit(mut self, limit: usize) -> Self {
        self.limit = Some(limit);
        self
    }
}

/// Temporal filter for events
#[derive(Default)]
pub struct TemporalFilter {
    pub event_types: Option<HashSet<String>>,
    pub aggregate_ids: Option<HashSet<String>>,
    pub user_ids: Option<HashSet<String>>,
    pub sources: Option<HashSet<String>>,
    pub custom_filters: Vec<CustomFilterFn>,
}

impl std::fmt::Debug for TemporalFilter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TemporalFilter")
            .field("event_types", &self.event_types)
            .field("aggregate_ids", &self.aggregate_ids)
            .field("user_ids", &self.user_ids)
            .field("sources", &self.sources)
            .field(
                "custom_filters",
                &format!("<{} filters>", self.custom_filters.len()),
            )
            .finish()
    }
}

impl Clone for TemporalFilter {
    fn clone(&self) -> Self {
        Self {
            event_types: self.event_types.clone(),
            aggregate_ids: self.aggregate_ids.clone(),
            user_ids: self.user_ids.clone(),
            sources: self.sources.clone(),
            custom_filters: Vec::new(), // Cannot clone function pointers
        }
    }
}

/// Temporal projection specification
#[derive(Debug, Clone, Default)]
pub enum TemporalProjection {
    /// Return full events
    #[default]
    FullEvents,
    /// Return only metadata
    MetadataOnly,
    /// Return specific fields
    Fields(Vec<String>),
    /// Return aggregated data
    Aggregation(AggregationType),
}

/// Aggregation type for temporal queries
#[derive(Debug, Clone)]
pub enum AggregationType {
    Count,
    CountBy(String),
    Timeline(ChronoDuration),
    Statistics,
}

/// Temporal ordering specification
#[derive(Debug, Clone, Default)]
pub enum TemporalOrdering {
    /// Order by timestamp ascending
    TimeAscending,
    /// Order by timestamp descending
    #[default]
    TimeDescending,
    /// Order by version ascending
    VersionAscending,
    /// Order by version descending
    VersionDescending,
    /// Order by custom field
    Custom(String, bool), // field, ascending
}

/// Result of a temporal query
#[derive(Debug, Clone)]
pub struct TemporalQueryResult {
    pub query_id: Uuid,
    pub events: Vec<StreamEvent>,
    pub metadata: TemporalResultMetadata,
    pub aggregations: Option<TemporalAggregations>,
    pub execution_time: Duration,
    pub from_cache: bool,
}

/// Metadata about temporal query results
#[derive(Debug, Clone)]
pub struct TemporalResultMetadata {
    pub total_events: usize,
    pub time_range_covered: Option<(DateTime<Utc>, DateTime<Utc>)>,
    pub version_range_covered: Option<(u64, u64)>,
    pub aggregates_scanned: HashSet<String>,
    pub index_hits: usize,
    pub index_misses: usize,
}

/// Aggregated data from temporal queries
#[derive(Debug, Clone)]
pub struct TemporalAggregations {
    pub count: usize,
    pub count_by_type: HashMap<String, usize>,
    pub timeline: Vec<TimelinePoint>,
    pub statistics: TemporalStatistics,
}

/// Point in timeline aggregation
#[derive(Debug, Clone)]
pub struct TimelinePoint {
    pub timestamp: DateTime<Utc>,
    pub count: usize,
    pub event_types: HashMap<String, usize>,
}

/// Statistical data from temporal queries
#[derive(Debug, Clone)]
pub struct TemporalStatistics {
    pub events_per_second: f64,
    pub peak_throughput: f64,
    pub average_event_size: f64,
    pub unique_aggregates: usize,
    pub unique_users: usize,
    pub time_span: ChronoDuration,
}

/// Temporal index for efficient time-travel queries
#[derive(Debug)]
struct TemporalIndex {
    /// Time-based index: timestamp -> event IDs
    time_index: BTreeMap<DateTime<Utc>, Vec<Uuid>>,
    /// Version-based index: version -> event metadata
    version_index: BTreeMap<u64, EventIndexEntry>,
    /// Aggregate-based index: aggregate_id -> time-ordered events
    aggregate_index: HashMap<String, BTreeMap<DateTime<Utc>, Vec<Uuid>>>,
    /// Type-based index: event_type -> time-ordered events
    type_index: HashMap<String, BTreeMap<DateTime<Utc>, Vec<Uuid>>>,
}

#[derive(Debug, Clone)]
struct EventIndexEntry {
    pub event_id: Uuid,
    pub timestamp: DateTime<Utc>,
    pub aggregate_id: String,
    pub event_type: String,
    pub version: u64,
}

impl TemporalIndex {
    fn new() -> Self {
        Self {
            time_index: BTreeMap::new(),
            version_index: BTreeMap::new(),
            aggregate_index: HashMap::new(),
            type_index: HashMap::new(),
        }
    }

    fn add_event(&mut self, event: &StreamEvent) {
        let metadata = event.metadata();
        let timestamp = metadata.timestamp;
        let event_id = uuid::Uuid::parse_str(&metadata.event_id).unwrap_or(uuid::Uuid::new_v4());
        let aggregate_id = metadata.context.clone().unwrap_or_default();
        let event_type = format!("{event:?}");
        let version = metadata.version.parse::<u64>().unwrap_or(0);

        // Time index
        self.time_index.entry(timestamp).or_default().push(event_id);

        // Version index
        self.version_index.insert(
            version,
            EventIndexEntry {
                event_id,
                timestamp,
                aggregate_id: aggregate_id.clone(),
                event_type: event_type.clone(),
                version,
            },
        );

        // Aggregate index
        self.aggregate_index
            .entry(aggregate_id)
            .or_default()
            .entry(timestamp)
            .or_default()
            .push(event_id);

        // Type index
        self.type_index
            .entry(event_type)
            .or_default()
            .entry(timestamp)
            .or_default()
            .push(event_id);
    }

    fn find_events_by_time_range(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> Vec<Uuid> {
        let mut event_ids = Vec::new();

        for (_, ids) in self.time_index.range(start..=end) {
            event_ids.extend_from_slice(ids);
        }

        event_ids
    }

    fn find_events_by_version_range(&self, start: u64, end: u64) -> Vec<Uuid> {
        let mut event_ids = Vec::new();

        for (_, entry) in self.version_index.range(start..=end) {
            event_ids.push(entry.event_id);
        }

        event_ids
    }

    fn find_events_by_aggregate(
        &self,
        aggregate_id: &str,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Vec<Uuid> {
        if let Some(time_map) = self.aggregate_index.get(aggregate_id) {
            let mut event_ids = Vec::new();
            for (_, ids) in time_map.range(start..=end) {
                event_ids.extend_from_slice(ids);
            }
            event_ids
        } else {
            Vec::new()
        }
    }
}

/// Time-travel query engine
pub struct TimeTravelEngine {
    config: TimeTravelConfig,
    event_store: Arc<dyn EventStoreTrait>,
    event_stream: Arc<dyn EventStream>,
    temporal_index: Arc<RwLock<TemporalIndex>>,
    query_cache: Arc<RwLock<QueryCache>>,
    query_semaphore: Arc<tokio::sync::Semaphore>,
    metrics: Arc<RwLock<TimeTravelMetrics>>,
}

impl TimeTravelEngine {
    /// Create a new time-travel engine
    pub fn new(
        config: TimeTravelConfig,
        event_store: Arc<dyn EventStoreTrait>,
        event_stream: Arc<dyn EventStream>,
    ) -> Self {
        Self {
            query_semaphore: Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_queries)),
            temporal_index: Arc::new(RwLock::new(TemporalIndex::new())),
            query_cache: Arc::new(RwLock::new(QueryCache::new(config.clone()))),
            config,
            event_store,
            event_stream,
            metrics: Arc::new(RwLock::new(TimeTravelMetrics::default())),
        }
    }

    /// Start the time-travel engine
    pub async fn start(&self) -> Result<()> {
        info!("Starting time-travel engine");

        // Build initial index if enabled
        if self.config.enable_temporal_indexing {
            self.build_temporal_index().await?;
        }

        // Start index maintenance task
        let index = Arc::clone(&self.temporal_index);
        let event_stream = Arc::clone(&self.event_stream);

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(60));
            loop {
                interval.tick().await;
                if let Err(e) =
                    Self::update_index(Arc::clone(&index), Arc::clone(&event_stream)).await
                {
                    error!("Failed to update temporal index: {}", e);
                }
            }
        });

        info!("Time-travel engine started successfully");
        Ok(())
    }

    /// Execute a temporal query
    pub async fn execute_query(&self, query: TemporalQuery) -> Result<TemporalQueryResult> {
        let start_time = Instant::now();
        let query_id = query.query_id;

        debug!("Executing temporal query {}", query_id);

        // Acquire semaphore for concurrency control
        let _permit = self.query_semaphore.acquire().await?;

        // Update metrics
        {
            let mut metrics = self.metrics.write().await;
            metrics.queries_executed += 1;
            metrics.active_queries += 1;
        }

        // Check cache first
        let cache_key = self.generate_cache_key(&query);
        if self.config.enable_result_caching {
            let cache = self.query_cache.read().await;
            if let Some(cached_result) = cache.get(&cache_key) {
                let mut metrics = self.metrics.write().await;
                metrics.active_queries -= 1;
                metrics.cache_hits += 1;

                return Ok(TemporalQueryResult {
                    query_id,
                    events: cached_result.events,
                    metadata: cached_result.metadata,
                    aggregations: cached_result.aggregations,
                    execution_time: start_time.elapsed(),
                    from_cache: true,
                });
            }
        }

        let result = self.execute_query_internal(query).await;

        // Update metrics
        {
            let mut metrics = self.metrics.write().await;
            metrics.active_queries -= 1;
            match &result {
                Ok(_) => {
                    metrics.queries_succeeded += 1;
                    if !self.config.enable_result_caching {
                        metrics.cache_misses += 1;
                    }
                }
                Err(_) => metrics.queries_failed += 1,
            }
        }

        let execution_time = start_time.elapsed();
        debug!(
            "Temporal query {} executed in {:?}",
            query_id, execution_time
        );

        if let Ok(ref res) = result {
            // Cache result if applicable
            if self.config.enable_result_caching {
                let mut cache = self.query_cache.write().await;
                cache.set(cache_key, res.clone());
            }
        }

        result.map(|mut r| {
            r.execution_time = execution_time;
            r.from_cache = false;
            r
        })
    }

    /// Execute query internally
    async fn execute_query_internal(&self, query: TemporalQuery) -> Result<TemporalQueryResult> {
        let query_id = query.query_id;

        // Resolve time points to actual timestamps
        let (start_time, end_time) = self.resolve_time_range(&query).await?;

        // Find candidate events
        let candidate_event_ids = if self.config.enable_temporal_indexing {
            self.find_events_with_index(&query, start_time, end_time)
                .await?
        } else {
            self.find_events_without_index(&query, start_time, end_time)
                .await?
        };

        // Load full events
        let mut events = Vec::new();
        for event_id in candidate_event_ids {
            if let Some(event) = self.load_event(event_id).await? {
                if self.matches_filter(&event, &query.filter) {
                    events.push(event);
                }
            }
        }

        // Apply ordering
        self.apply_ordering(&mut events, &query.ordering);

        // Apply limit
        if let Some(limit) = query.limit {
            events.truncate(limit);
        }

        // Generate metadata
        let metadata = self.generate_result_metadata(&events, start_time, end_time);

        // Generate aggregations if requested
        let aggregations = match query.projection {
            TemporalProjection::Aggregation(ref agg_type) => {
                Some(self.generate_aggregations(&events, agg_type, start_time, end_time)?)
            }
            _ => None,
        };

        // Apply projection
        let projected_events = self.apply_projection(events, &query.projection);

        Ok(TemporalQueryResult {
            query_id,
            events: projected_events,
            metadata,
            aggregations,
            execution_time: Duration::default(), // Will be set by caller
            from_cache: false,
        })
    }

    /// Query state at specific time point
    pub async fn query_state_at_time(
        &self,
        aggregate_id: &str,
        time_point: TimePoint,
    ) -> Result<Vec<StreamEvent>> {
        let query = TemporalQuery::new()
            .at_time(time_point)
            .filter(TemporalFilter {
                aggregate_ids: Some(std::iter::once(aggregate_id.to_string()).collect()),
                ..Default::default()
            });

        let result = self.execute_query(query).await?;
        Ok(result.events)
    }

    /// Query changes between two time points
    pub async fn query_changes_between(
        &self,
        start: TimePoint,
        end: TimePoint,
        filter: Option<TemporalFilter>,
    ) -> Result<Vec<StreamEvent>> {
        let query = TemporalQuery::new()
            .in_range(TimeRange { start, end })
            .filter(filter.unwrap_or_default());

        let result = self.execute_query(query).await?;
        Ok(result.events)
    }

    /// Query timeline aggregation
    pub async fn query_timeline(
        &self,
        time_range: TimeRange,
        granularity: ChronoDuration,
        filter: Option<TemporalFilter>,
    ) -> Result<Vec<TimelinePoint>> {
        let query = TemporalQuery::new()
            .in_range(time_range)
            .filter(filter.unwrap_or_default())
            .project(TemporalProjection::Aggregation(AggregationType::Timeline(
                granularity,
            )));

        let result = self.execute_query(query).await?;
        Ok(result.aggregations.map(|a| a.timeline).unwrap_or_default())
    }

    /// Build temporal index from existing events
    async fn build_temporal_index(&self) -> Result<()> {
        info!("Building temporal index");

        let events = self
            .event_stream
            .read_events_from_position(0, usize::MAX)
            .await?;
        let mut index = self.temporal_index.write().await;

        for stored_event in events {
            index.add_event(&stored_event.event_data);
        }

        info!(
            "Temporal index built with {} events",
            index.time_index.len()
        );
        Ok(())
    }

    /// Update index with new events
    async fn update_index(
        index: Arc<RwLock<TemporalIndex>>,
        event_stream: Arc<dyn EventStream>,
    ) -> Result<()> {
        // This would typically track the last processed position
        // For simplicity, we'll just rebuild periodically
        let events = event_stream.read_events_from_position(0, 10000).await?;
        let mut idx = index.write().await;

        for stored_event in events {
            idx.add_event(&stored_event.event_data);
        }

        Ok(())
    }

    /// Resolve time range from query specification
    async fn resolve_time_range(
        &self,
        query: &TemporalQuery,
    ) -> Result<(DateTime<Utc>, DateTime<Utc>)> {
        let now = Utc::now();

        match (&query.time_point, &query.time_range) {
            (Some(time_point), None) => {
                let timestamp = self.resolve_time_point(time_point).await?;
                Ok((timestamp, timestamp))
            }
            (None, Some(time_range)) => {
                let start = self.resolve_time_point(&time_range.start).await?;
                let end = self.resolve_time_point(&time_range.end).await?;
                Ok((start, end))
            }
            (None, None) => {
                // Default to last 24 hours
                let start = now - ChronoDuration::hours(24);
                Ok((start, now))
            }
            (Some(_), Some(_)) => Err(anyhow!("Cannot specify both time_point and time_range")),
        }
    }

    /// Resolve a time point to an actual timestamp
    async fn resolve_time_point(&self, time_point: &TimePoint) -> Result<DateTime<Utc>> {
        match time_point {
            TimePoint::Timestamp(timestamp) => Ok(*timestamp),
            TimePoint::RelativeTime(duration) => Ok(Utc::now() + *duration),
            TimePoint::Version(version) => {
                // Find timestamp for this version
                let index = self.temporal_index.read().await;
                if let Some(entry) = index.version_index.get(version) {
                    Ok(entry.timestamp)
                } else {
                    Err(anyhow!("Version {} not found", version))
                }
            }
            TimePoint::EventId(event_id) => {
                // Find timestamp for this event ID
                if let Some(event) = self.load_event(*event_id).await? {
                    Ok(event.metadata().timestamp)
                } else {
                    Err(anyhow!("Event {} not found", event_id))
                }
            }
            TimePoint::Snapshot(name) => {
                // Named snapshots require an external snapshot store integration.
                // The TimeTravelEngine does not embed a snapshot store; callers that need
                // named-snapshot resolution should resolve the snapshot to a concrete
                // Timestamp or Version and use that TimePoint variant instead.
                Err(anyhow!(
                    "Named snapshot '{}' cannot be resolved: the TimeTravelEngine requires an \
                     integrated snapshot store for named-snapshot resolution. \
                     Convert the snapshot to a Timestamp or Version TimePoint.",
                    name
                ))
            }
        }
    }

    /// Find events using temporal index
    async fn find_events_with_index(
        &self,
        query: &TemporalQuery,
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
    ) -> Result<Vec<Uuid>> {
        let index = self.temporal_index.read().await;

        // Use most specific index available
        if let Some(ref aggregate_ids) = query.filter.aggregate_ids {
            if aggregate_ids.len() == 1 {
                let aggregate_id = aggregate_ids
                    .iter()
                    .next()
                    .expect("aggregate_ids validated to have exactly 1 element");
                return Ok(index.find_events_by_aggregate(aggregate_id, start_time, end_time));
            }
        }

        Ok(index.find_events_by_time_range(start_time, end_time))
    }

    /// Find events without using index (sequential scan).
    ///
    /// Used whenever `enable_temporal_indexing` is off (or as a correctness
    /// baseline): scans the full event stream and applies the same time-range
    /// (and, if present, single-aggregate) filtering that
    /// [`Self::find_events_with_index`] gets "for free" from `TemporalIndex`,
    /// so results are consistent between the indexed and non-indexed paths.
    async fn find_events_without_index(
        &self,
        query: &TemporalQuery,
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
    ) -> Result<Vec<Uuid>> {
        let stored_events = self
            .event_stream
            .read_events_from_position(0, usize::MAX)
            .await?;

        let aggregate_filter = query.filter.aggregate_ids.as_ref();

        let mut event_ids = Vec::new();
        for stored in &stored_events {
            let metadata = stored.event_data.metadata();

            if metadata.timestamp < start_time || metadata.timestamp > end_time {
                continue;
            }

            if let Some(aggregate_ids) = aggregate_filter {
                let matches = metadata
                    .context
                    .as_ref()
                    .is_some_and(|ctx| aggregate_ids.contains(ctx));
                if !matches {
                    continue;
                }
            }

            event_ids.push(Self::event_uuid(stored));
        }

        debug!(
            "Sequential scan found {} candidate event(s) in range {}..{}",
            event_ids.len(),
            start_time,
            end_time
        );

        Ok(event_ids)
    }

    /// Resolve the stable UUID identity used to reference a stored event
    /// throughout `TimeTravelEngine` (matches the derivation in
    /// [`TemporalIndex::add_event`]): parsed from `metadata.event_id` when
    /// that's a valid UUID, falling back to the event store's own
    /// `StoredEvent::event_id` otherwise (unlike the index's random fallback,
    /// this is deterministic, which is required for `load_event` to be able
    /// to look events back up by the ID `find_events_without_index` returned).
    fn event_uuid(stored: &crate::event_sourcing::StoredEvent) -> Uuid {
        Uuid::parse_str(&stored.event_data.metadata().event_id).unwrap_or(stored.event_id)
    }

    /// Load a specific event by ID.
    ///
    /// The event store trait has no direct get-by-ID lookup, so this scans
    /// the event stream (same source [`Self::find_events_without_index`] and
    /// [`Self::build_temporal_index`] use) and returns the first event whose
    /// resolved UUID identity (see [`Self::event_uuid`]) matches.
    async fn load_event(&self, event_id: Uuid) -> Result<Option<StreamEvent>> {
        let stored_events = self
            .event_stream
            .read_events_from_position(0, usize::MAX)
            .await?;

        Ok(stored_events
            .into_iter()
            .find(|stored| Self::event_uuid(stored) == event_id)
            .map(|stored| stored.event_data))
    }

    /// Check if event matches filter
    fn matches_filter(&self, event: &StreamEvent, filter: &TemporalFilter) -> bool {
        let metadata = event.metadata();
        let event_type_str = format!("{event:?}");

        if let Some(ref event_types) = filter.event_types {
            if !event_types.contains(&event_type_str) {
                return false;
            }
        }

        if let Some(ref aggregate_ids) = filter.aggregate_ids {
            if let Some(ref context) = metadata.context {
                if !aggregate_ids.contains(context) {
                    return false;
                }
            } else {
                return false;
            }
        }

        if let Some(ref user_ids) = filter.user_ids {
            if let Some(ref user) = metadata.user {
                if !user_ids.contains(user) {
                    return false;
                }
            } else {
                return false;
            }
        }

        if let Some(ref sources) = filter.sources {
            if !sources.contains(&metadata.source) {
                return false;
            }
        }

        // Apply custom filters
        for custom_filter in &filter.custom_filters {
            if !custom_filter(event) {
                return false;
            }
        }

        true
    }

    /// Apply ordering to events
    fn apply_ordering(&self, events: &mut [StreamEvent], ordering: &TemporalOrdering) {
        match ordering {
            TemporalOrdering::TimeAscending => {
                events.sort_by_key(|a| a.metadata().timestamp);
            }
            TemporalOrdering::TimeDescending => {
                events.sort_by_key(|b| std::cmp::Reverse(b.metadata().timestamp));
            }
            TemporalOrdering::VersionAscending => {
                events.sort_by(|a, b| a.metadata().version.cmp(&b.metadata().version));
            }
            TemporalOrdering::VersionDescending => {
                events.sort_by(|a, b| b.metadata().version.cmp(&a.metadata().version));
            }
            TemporalOrdering::Custom(_field, _ascending) => {
                // Custom field ordering would be implemented here
                warn!("Custom ordering not implemented");
            }
        }
    }

    /// Apply projection to events
    fn apply_projection(
        &self,
        events: Vec<StreamEvent>,
        projection: &TemporalProjection,
    ) -> Vec<StreamEvent> {
        match projection {
            TemporalProjection::FullEvents => events,
            TemporalProjection::MetadataOnly => {
                // Return events with only metadata (simplified data)
                // For metadata-only projection, we keep the event but could filter data in a real implementation
                events
            }
            TemporalProjection::Fields(_fields) => {
                // Field projection would be implemented here
                warn!("Field projection not implemented");
                events
            }
            TemporalProjection::Aggregation(_) => {
                // Aggregation results are handled separately
                Vec::new()
            }
        }
    }

    /// Generate result metadata
    fn generate_result_metadata(
        &self,
        events: &[StreamEvent],
        _start_time: DateTime<Utc>,
        _end_time: DateTime<Utc>,
    ) -> TemporalResultMetadata {
        let total_events = events.len();

        let time_range_covered = if !events.is_empty() {
            let min_time = events
                .iter()
                .map(|e| e.metadata().timestamp)
                .min()
                .expect("events validated to be non-empty");
            let max_time = events
                .iter()
                .map(|e| e.metadata().timestamp)
                .max()
                .expect("events validated to be non-empty");
            Some((min_time, max_time))
        } else {
            None
        };

        let version_range_covered = if !events.is_empty() {
            let min_version = events
                .iter()
                .filter_map(|e| e.metadata().version.parse::<u64>().ok())
                .min();
            let max_version = events
                .iter()
                .filter_map(|e| e.metadata().version.parse::<u64>().ok())
                .max();
            if let (Some(min), Some(max)) = (min_version, max_version) {
                Some((min, max))
            } else {
                None
            }
        } else {
            None
        };

        let aggregates_scanned: HashSet<String> = events
            .iter()
            .filter_map(|e| e.metadata().context.clone())
            .collect();

        TemporalResultMetadata {
            total_events,
            time_range_covered,
            version_range_covered,
            aggregates_scanned,
            index_hits: 0, // Would be tracked during execution
            index_misses: 0,
        }
    }

    /// Generate aggregations
    fn generate_aggregations(
        &self,
        events: &[StreamEvent],
        agg_type: &AggregationType,
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
    ) -> Result<TemporalAggregations> {
        match agg_type {
            AggregationType::Count => Ok(TemporalAggregations {
                count: events.len(),
                count_by_type: HashMap::new(),
                timeline: Vec::new(),
                statistics: self.calculate_statistics(events, start_time, end_time),
            }),
            AggregationType::CountBy(field) => {
                let mut count_by_type = HashMap::new();
                for event in events {
                    if field == "event_type" {
                        let event_type = format!("{event:?}");
                        *count_by_type.entry(event_type).or_insert(0) += 1;
                    }
                    // Other fields would be handled here
                }

                Ok(TemporalAggregations {
                    count: events.len(),
                    count_by_type,
                    timeline: Vec::new(),
                    statistics: self.calculate_statistics(events, start_time, end_time),
                })
            }
            AggregationType::Timeline(granularity) => {
                let timeline = self.generate_timeline(events, *granularity, start_time, end_time);

                Ok(TemporalAggregations {
                    count: events.len(),
                    count_by_type: HashMap::new(),
                    timeline,
                    statistics: self.calculate_statistics(events, start_time, end_time),
                })
            }
            AggregationType::Statistics => Ok(TemporalAggregations {
                count: events.len(),
                count_by_type: HashMap::new(),
                timeline: Vec::new(),
                statistics: self.calculate_statistics(events, start_time, end_time),
            }),
        }
    }

    /// Generate timeline aggregation
    fn generate_timeline(
        &self,
        events: &[StreamEvent],
        granularity: ChronoDuration,
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
    ) -> Vec<TimelinePoint> {
        let mut timeline = Vec::new();
        let mut current_time = start_time;

        while current_time < end_time {
            let window_end = current_time + granularity;

            let events_in_window: Vec<_> = events
                .iter()
                .filter(|e| {
                    e.metadata().timestamp >= current_time && e.metadata().timestamp < window_end
                })
                .collect();

            let mut event_types = HashMap::new();
            for event in &events_in_window {
                let event_type = format!("{event:?}");
                *event_types.entry(event_type).or_insert(0) += 1;
            }

            timeline.push(TimelinePoint {
                timestamp: current_time,
                count: events_in_window.len(),
                event_types,
            });

            current_time = window_end;
        }

        timeline
    }

    /// Calculate temporal statistics
    fn calculate_statistics(
        &self,
        events: &[StreamEvent],
        start_time: DateTime<Utc>,
        end_time: DateTime<Utc>,
    ) -> TemporalStatistics {
        let time_span = end_time.signed_duration_since(start_time);
        let events_per_second = if time_span.num_seconds() > 0 {
            events.len() as f64 / time_span.num_seconds() as f64
        } else {
            0.0
        };

        // Calculate peak throughput (events per second in busiest minute)
        let peak_throughput = if !events.is_empty() {
            let mut minute_counts = HashMap::new();
            for event in events {
                let minute = event
                    .metadata()
                    .timestamp
                    .format("%Y-%m-%d %H:%M")
                    .to_string();
                *minute_counts.entry(minute).or_insert(0) += 1;
            }
            minute_counts.values().max().copied().unwrap_or(0) as f64
        } else {
            0.0
        };

        // Calculate average event size
        let total_size: usize = events.iter().map(|e| format!("{e:?}").len()).sum();
        let average_event_size = if !events.is_empty() {
            total_size as f64 / events.len() as f64
        } else {
            0.0
        };

        let unique_aggregates = events
            .iter()
            .filter_map(|e| e.metadata().context.as_ref())
            .collect::<HashSet<_>>()
            .len();

        let unique_users = events
            .iter()
            .filter_map(|e| e.metadata().user.as_ref())
            .collect::<HashSet<_>>()
            .len();

        TemporalStatistics {
            events_per_second,
            peak_throughput,
            average_event_size,
            unique_aggregates,
            unique_users,
            time_span,
        }
    }

    /// Generate cache key for query
    fn generate_cache_key(&self, query: &TemporalQuery) -> String {
        // Simple cache key based on query structure
        format!("temporal_query_{:?}", query.query_id)
    }

    /// Get time-travel metrics
    pub async fn get_metrics(&self) -> TimeTravelMetrics {
        self.metrics.read().await.clone()
    }
}

/// Query cache for temporal queries
#[derive(Debug)]
struct QueryCache {
    config: TimeTravelConfig,
    entries: HashMap<String, CachedResult>,
}

#[derive(Debug, Clone)]
struct CachedResult {
    events: Vec<StreamEvent>,
    metadata: TemporalResultMetadata,
    aggregations: Option<TemporalAggregations>,
    cached_at: DateTime<Utc>,
}

impl QueryCache {
    fn new(config: TimeTravelConfig) -> Self {
        Self {
            config,
            entries: HashMap::new(),
        }
    }

    fn get(&self, key: &str) -> Option<CachedResult> {
        if let Some(entry) = self.entries.get(key) {
            let age = Utc::now().signed_duration_since(entry.cached_at);
            if age.num_minutes() < self.config.cache_ttl_minutes as i64 {
                return Some(entry.clone());
            }
        }
        None
    }

    fn set(&mut self, key: String, result: TemporalQueryResult) {
        let entry = CachedResult {
            events: result.events,
            metadata: result.metadata,
            aggregations: result.aggregations,
            cached_at: Utc::now(),
        };

        self.entries.insert(key, entry);
        self.evict_if_needed();
    }

    fn evict_if_needed(&mut self) {
        // Remove expired entries
        let now = Utc::now();
        self.entries.retain(|_, entry| {
            let age = now.signed_duration_since(entry.cached_at);
            age.num_minutes() < self.config.cache_ttl_minutes as i64
        });

        // Simple memory management (could be more sophisticated)
        while self.entries.len() > 1000 {
            if let Some(oldest_key) = self
                .entries
                .iter()
                .min_by_key(|(_, entry)| entry.cached_at)
                .map(|(key, _)| key.clone())
            {
                self.entries.remove(&oldest_key);
            } else {
                break;
            }
        }
    }
}

/// Time-travel engine metrics
#[derive(Debug, Clone, Default)]
pub struct TimeTravelMetrics {
    pub queries_executed: u64,
    pub queries_succeeded: u64,
    pub queries_failed: u64,
    pub active_queries: u64,
    pub cache_hits: u64,
    pub cache_misses: u64,
    pub index_hits: u64,
    pub index_misses: u64,
    pub average_query_time_ms: f64,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event_sourcing::{EventQuery, EventSnapshot, StoredEvent};
    use crate::EventMetadata;

    /// Minimal in-memory `EventStream` used only to exercise
    /// `find_events_without_index` / `load_event` against real (if
    /// synthetic) stored events, since no production `EventStream`
    /// implementor exists to construct a `TimeTravelEngine` against.
    struct MockEventStream {
        events: Vec<StoredEvent>,
    }

    #[async_trait::async_trait]
    impl EventStream for MockEventStream {
        async fn next_event(&mut self) -> Option<StoredEvent> {
            None
        }

        async fn has_events(&self) -> bool {
            !self.events.is_empty()
        }

        async fn read_events_from_position(
            &self,
            position: u64,
            max_events: usize,
        ) -> Result<Vec<StoredEvent>> {
            Ok(self
                .events
                .iter()
                .skip(position as usize)
                .take(max_events)
                .cloned()
                .collect())
        }
    }

    /// `EventStoreTrait` is required to construct a `TimeTravelEngine` but
    /// is untouched by the methods under test here, so every method is a
    /// harmless stub.
    struct MockEventStore;

    #[async_trait::async_trait]
    impl EventStoreTrait for MockEventStore {
        async fn store_event(&self, stream_id: String, event: StreamEvent) -> Result<StoredEvent> {
            Ok(test_stored_event(&stream_id, event))
        }

        async fn query_events(&self, _query: EventQuery) -> Result<Vec<StoredEvent>> {
            Ok(Vec::new())
        }

        async fn get_stream_events(
            &self,
            _stream_id: &str,
            _from_version: Option<u64>,
        ) -> Result<Vec<StoredEvent>> {
            Ok(Vec::new())
        }

        async fn replay_from_timestamp(
            &self,
            _timestamp: DateTime<Utc>,
        ) -> Result<Vec<StoredEvent>> {
            Ok(Vec::new())
        }

        async fn get_latest_snapshot(&self, _stream_id: &str) -> Result<Option<EventSnapshot>> {
            Ok(None)
        }

        async fn rebuild_stream_state(&self, _stream_id: &str) -> Result<Vec<u8>> {
            Ok(Vec::new())
        }

        async fn append_events(
            &self,
            _aggregate_id: &str,
            _events: &[StreamEvent],
            _expected_version: Option<u64>,
        ) -> Result<u64> {
            Ok(0)
        }
    }

    fn test_event(source: &str, timestamp: DateTime<Utc>) -> StreamEvent {
        StreamEvent::TripleAdded {
            subject: "http://example.org/s".to_string(),
            predicate: "http://example.org/p".to_string(),
            object: "\"o\"".to_string(),
            graph: None,
            metadata: EventMetadata {
                event_id: Uuid::new_v4().to_string(),
                timestamp,
                source: source.to_string(),
                user: None,
                context: None,
                caused_by: None,
                version: "1".to_string(),
                properties: HashMap::new(),
                checksum: None,
            },
        }
    }

    fn test_stored_event(stream_id: &str, event_data: StreamEvent) -> StoredEvent {
        let event_id =
            Uuid::parse_str(&event_data.metadata().event_id).unwrap_or_else(|_| Uuid::new_v4());
        StoredEvent {
            event_id,
            sequence_number: 0,
            stream_id: stream_id.to_string(),
            stream_version: 0,
            event_data,
            stored_at: Utc::now(),
            storage_metadata: crate::event_sourcing::StorageMetadata {
                checksum: String::new(),
                compressed_size: None,
                original_size: 0,
                storage_location: "test".to_string(),
                persistence_status: crate::event_sourcing::PersistenceStatus::Persisted,
            },
        }
    }

    fn make_engine(events: Vec<StreamEvent>) -> TimeTravelEngine {
        let stored: Vec<StoredEvent> = events
            .into_iter()
            .map(|e| test_stored_event("test-stream", e))
            .collect();
        TimeTravelEngine::new(
            TimeTravelConfig::default(),
            Arc::new(MockEventStore),
            Arc::new(MockEventStream { events: stored }),
        )
    }

    /// Regression test for time_travel.rs:750 — `find_events_without_index`
    /// used to unconditionally `warn!` and return an empty `Vec`, regardless
    /// of what was actually in the event store. It must now genuinely scan
    /// the event stream and return events whose timestamp falls in range.
    #[tokio::test]
    async fn test_find_events_without_index_scans_real_events() {
        let now = Utc::now();
        let in_range = test_event("svc-a", now);
        let out_of_range = test_event("svc-a", now - ChronoDuration::days(30));
        let engine = make_engine(vec![in_range.clone(), out_of_range]);

        let query = TemporalQuery::new();
        let start = now - ChronoDuration::hours(1);
        let end = now + ChronoDuration::hours(1);

        let found = engine
            .find_events_without_index(&query, start, end)
            .await
            .expect("sequential scan must not error");

        assert_eq!(
            found.len(),
            1,
            "sequential scan must find exactly the in-range event, not silently return empty"
        );
    }

    /// Regression test for time_travel.rs:763 — `load_event` used to always
    /// return `Ok(None)` regardless of the requested ID, breaking
    /// `TimePoint::EventId` resolution. It must now actually locate and
    /// return the matching event from the event store.
    #[tokio::test]
    async fn test_load_event_finds_real_event() {
        let now = Utc::now();
        let event = test_event("svc-a", now);
        let event_id =
            Uuid::parse_str(&event.metadata().event_id).expect("test event_id is a valid UUID");
        let engine = make_engine(vec![event]);

        let loaded = engine
            .load_event(event_id)
            .await
            .expect("load_event must not error");

        assert!(
            loaded.is_some(),
            "load_event must find a previously stored event by ID, not always return None"
        );
        assert_eq!(
            Uuid::parse_str(&loaded.expect("checked above").metadata().event_id)
                .expect("stored event_id is a valid UUID"),
            event_id
        );

        // A random, never-stored ID must still resolve to `None` (correct
        // "not found", not a fabricated hit).
        let missing = engine
            .load_event(Uuid::new_v4())
            .await
            .expect("load_event must not error for a missing ID");
        assert!(missing.is_none());
    }

    #[tokio::test]
    async fn test_time_travel_config_defaults() {
        let config = TimeTravelConfig::default();
        assert_eq!(config.max_time_window_days, 365);
        assert!(config.enable_temporal_indexing);
        assert_eq!(config.index_granularity_minutes, 60);
    }

    #[tokio::test]
    async fn test_temporal_query_builder() {
        let query = TemporalQuery::new()
            .at_time(TimePoint::Timestamp(Utc::now()))
            .filter(TemporalFilter::default())
            .order_by(TemporalOrdering::TimeDescending)
            .limit(100);

        assert!(query.time_point.is_some());
        assert!(query.limit.is_some());
        assert_eq!(query.limit.unwrap(), 100);
    }

    #[tokio::test]
    async fn test_time_point_resolution() {
        let now = Utc::now();
        let relative = TimePoint::RelativeTime(ChronoDuration::hours(-1));

        match relative {
            TimePoint::RelativeTime(duration) => {
                let resolved = now + duration;
                assert!(resolved < now);
            }
            _ => panic!("Expected RelativeTime"),
        }
    }

    #[tokio::test]
    async fn test_temporal_filter() {
        let filter = TemporalFilter {
            event_types: Some(std::iter::once("TestEvent".to_string()).collect()),
            ..Default::default()
        };

        assert!(filter.event_types.is_some());
        assert!(filter.event_types.as_ref().unwrap().contains("TestEvent"));
    }
}