rsp-rs 0.3.5

use crate::{QuadContainer, WindowInstance};
use oxigraph::model::Quad;
use oxigraph::store::Store;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

/// Report strategy for window content emission
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReportStrategy {
    NonEmptyContent,
    OnContentChange,
    OnWindowClose,
    Periodic,
}

/// Tick mechanism for window progression
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Tick {
    TimeDriven,
    TupleDriven,
    BatchDriven,
}

/// Output stream type
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum StreamType {
    RStream,
    IStream,
    DStream,
}

/// Callback type for window content emission
pub type WindowCallback = Arc<dyn Fn(QuadContainer) + Send + Sync>;

/// CSPARQL Window implementation
pub struct CSPARQLWindow {
    pub name: String,
    pub width: i64,
    pub slide: i64,
    pub time: i64,
    pub t0: i64,
    pub active_windows: HashMap<WindowInstance, QuadContainer>,
    pub report: ReportStrategy,
    pub tick: Tick,
    callbacks: HashMap<StreamType, Vec<WindowCallback>>,
    pub debug_mode: bool,
}

impl CSPARQLWindow {
    pub fn new(
        name: String,
        width: i64,
        slide: i64,
        report: ReportStrategy,
        tick: Tick,
        start_time: i64,
    ) -> Self {
        Self {
            name,
            width,
            slide,
            report,
            tick,
            time: start_time,
            t0: start_time,
            active_windows: HashMap::new(),
            callbacks: HashMap::new(),
            debug_mode: false,
        }
    }

    /// Get window content at a specific timestamp
    /// Returns the window with the smallest close time that contains the timestamp
    pub fn get_content(&self, timestamp: i64) -> Option<&QuadContainer> {
        let mut max_window: Option<&WindowInstance> = None;
        let mut max_time = i64::MAX;

        for (window, _container) in &self.active_windows {
            if window.open <= timestamp && timestamp <= window.close {
                if window.close < max_time {
                    max_time = window.close;
                    max_window = Some(window);
                }
            }
        }

        max_window.and_then(|w| self.active_windows.get(w))
    }

    /// Add a quad to the window at the given timestamp
    pub fn add(&mut self, quad: Quad, timestamp: i64) {
        if self.debug_mode {
            eprintln!(
                "[WINDOW {}] Received element ({:?},{}) ",
                self.name, quad, timestamp
            );
        }

        // Create a new quad with the window's graph name
        // This ensures the quad's graph matches the SPARQL query's GRAPH clause
        let quad_with_window_graph = oxigraph::model::Quad::new(
            quad.subject.clone(),
            quad.predicate.clone(),
            quad.object.clone(),
            oxigraph::model::GraphName::NamedNode(
                oxigraph::model::NamedNode::new(&self.name).unwrap_or_else(|_| {
                    // Fallback if window name isn't a valid IRI
                    oxigraph::model::NamedNode::new("http://default-window").unwrap()
                }),
            ),
        );

        let mut to_evict = Vec::new();
        let t_e = timestamp;

        if self.time > t_e {
            eprintln!("OUT OF ORDER NOT HANDLED");
        }

        self.scope(t_e);

        // Add element to appropriate windows
        for (window, container) in &mut self.active_windows {
            if self.debug_mode {
                eprintln!(
                    "[WINDOW {}] Processing Window [{},{}) for element ({:?},{})",
                    self.name, window.open, window.close, quad_with_window_graph, timestamp
                );
            }

            if window.open <= t_e && t_e < window.close {
                if self.debug_mode {
                    eprintln!(
                        "[WINDOW {}] Adding element to Window [{},{})",
                        self.name, window.open, window.close
                    );
                }
                container.add(quad_with_window_graph.clone(), timestamp);
                if self.debug_mode {
                    eprintln!(
                        "[WINDOW {}] Window [{},{}) now has {} quads",
                        self.name,
                        window.open,
                        window.close,
                        container.len()
                    );
                }
            } else if t_e >= window.close {
                if self.debug_mode {
                    eprintln!(
                        "[WINDOW {}] Scheduling for Eviction [{},{})",
                        self.name, window.open, window.close
                    );
                }
                // Don't add to eviction list yet - windows need to report before being evicted
                // to_evict.push(window.clone());
            }
        }

        // Find the window to report
        if self.debug_mode {
            eprintln!(
                "[WINDOW {}] Active windows before reporting check: {}",
                self.name,
                self.active_windows.len()
            );
        }

        let mut max_window: Option<WindowInstance> = None;
        let mut max_time = 0i64;

        for (window, container) in &self.active_windows {
            if self.compute_report(window, container, timestamp) {
                if self.debug_mode {
                    eprintln!(
                        "[WINDOW {}] Window [{},{}) should report (has {} quads)",
                        self.name,
                        window.open,
                        window.close,
                        container.len()
                    );
                }
                if window.close > max_time {
                    max_time = window.close;
                    max_window = Some(window.clone());
                }
                // Mark window for eviction after it reports
                to_evict.push(window.clone());
            }
        }

        // Emit window content if conditions are met
        if let Some(window) = max_window {
            if self.debug_mode {
                eprintln!(
                    "[WINDOW {}] Max window selected for reporting: [{},{})",
                    self.name, window.open, window.close
                );
            }
            if self.tick == Tick::TimeDriven {
                if timestamp > self.time {
                    self.time = timestamp;
                    if let Some(content) = self.active_windows.get(&window) {
                        if self.debug_mode {
                            eprintln!(
                                "[WINDOW {}] Emitting {} quads at t={} for window [{},{})",
                                self.name,
                                content.len(),
                                timestamp,
                                window.open,
                                window.close
                            );
                        }
                        self.emit(StreamType::RStream, content.clone());
                    } else {
                        if self.debug_mode {
                            eprintln!(
                                "[WINDOW {}] ERROR: Window [{},{}) not found in active_windows!",
                                self.name, window.open, window.close
                            );
                        }
                    }
                }
            }
        }

        // Evict old windows
        for window in to_evict {
            if self.debug_mode {
                eprintln!(
                    "[WINDOW {}] Evicting [{},{})",
                    self.name, window.open, window.close
                );
            }
            self.active_windows.remove(&window);
        }
    }

    /// Compute whether to report this window based on the report strategy
    fn compute_report(
        &self,
        window: &WindowInstance,
        _content: &QuadContainer,
        timestamp: i64,
    ) -> bool {
        match self.report {
            ReportStrategy::OnWindowClose => window.close < timestamp,
            ReportStrategy::NonEmptyContent => !_content.is_empty(),
            ReportStrategy::OnContentChange => true, // TODO : Tracking content changes needed here but for now always true as a placeholder for future implementation
            ReportStrategy::Periodic => true, // TODO : Implement periodic reporting logic here as content is always true for now
        }
    }

    /// Calculate and create windows based on the event time
    pub fn scope(&mut self, t_e: i64) {
        if self.t0 == 0 {
            self.t0 = t_e;
        }

        // Use integer arithmetic to avoid precision loss with large timestamps
        // This computes ceiling division: ceil(dividend / divisor) = (dividend + divisor - 1) / divisor
        let delta = (t_e - self.t0).abs();
        let c_sup = self.t0 + ((delta + self.slide - 1) / self.slide) * self.slide;
        let mut o_i = c_sup - self.width;

        if self.debug_mode {
            eprintln!(
                "[WINDOW {}] Calculating the Windows to Open. First one opens at [{}] and closes at [{}]",
                self.name, o_i, c_sup
            );
        }

        while o_i <= t_e {
            if self.debug_mode {
                eprintln!(
                    "[WINDOW {}] Computing Window [{},{}) if absent",
                    self.name,
                    o_i,
                    o_i + self.width
                );
            }

            let window = WindowInstance::new(o_i, o_i + self.width);
            self.compute_window_if_absent(window);
            o_i += self.slide;
        }
    }

    /// Add window if it doesn't already exist
    fn compute_window_if_absent(&mut self, key: WindowInstance) {
        self.active_windows
            .entry(key)
            .or_insert_with(|| QuadContainer::new(HashSet::new(), 0));
    }

    /// Subscribe a callback to window emissions
    pub fn subscribe<F>(&mut self, stream_type: StreamType, callback: F)
    where
        F: Fn(QuadContainer) + Send + Sync + 'static,
    {
        let callbacks = self.callbacks.entry(stream_type).or_insert_with(Vec::new);
        callbacks.push(Arc::new(callback));
    }

    /// Emit window content to subscribers
    fn emit(&self, stream_type: StreamType, content: QuadContainer) {
        if let Some(callbacks) = self.callbacks.get(&stream_type) {
            for callback in callbacks {
                callback(content.clone());
            }
        }
    }

    /// Get content from window at specific timestamp (alternative method name for compatibility)
    pub fn get_content_from_window(&self, timestamp: i64) -> Option<&QuadContainer> {
        self.get_content(timestamp)
    }

    /// Get the current number of active windows
    pub fn get_active_window_count(&self) -> usize {
        self.active_windows.len()
    }

    /// Get the timestamp range of active windows
    pub fn get_active_window_ranges(&self) -> Vec<(i64, i64)> {
        self.active_windows
            .iter()
            .map(|(window, _)| (window.open, window.close))
            .collect()
    }

    /// Enable or disable debug mode for verbose logging
    pub fn set_debug_mode(&mut self, enabled: bool) {
        self.debug_mode = enabled;
    }
}

use oxigraph::sparql::QueryResults;

pub fn execute_query<'a>(
    container: &'a QuadContainer,
    query: &str,
) -> Result<QueryResults<'a>, Box<dyn std::error::Error>> {
    let store = Store::new()?;
    for quad in &container.elements {
        store.insert(quad)?;
    }
    use oxigraph::sparql::SparqlEvaluator;
    let results = SparqlEvaluator::new()
        .parse_query(query)?
        .on_store(&store)
        .execute()
        .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;
    Ok(results)
}