//! Store-agnostic GraphQL resolvers for the causal inspector.
//!
//! Queries read from `Arc<dyn InspectorReadModel>` (injected into context).
//! Subscriptions read from `tokio::sync::broadcast::Sender<PersistedEvent>`
//! (injected into context — how events get broadcast is the consumer's concern).

use std::sync::Arc;

use async_graphql::{Context, Object, Result, Subscription};
use chrono::{DateTime, Utc};
use futures::Stream;

use crate::display::EventDisplay;
use crate::read_model::{InspectorReadModel, EventQuery, StoredEvent};
use crate::types::*;

/// Generic inspector query resolvers for causal event tables.
///
/// The `D` parameter controls how events are displayed (names, summaries).
///
/// # Context requirements
///
/// The async-graphql context must contain:
/// - `Arc<dyn InspectorReadModel>` — the store to query
pub struct CausalInspectorQuery<D: EventDisplay> {
    display: Arc<D>,
}

impl<D: EventDisplay> CausalInspectorQuery<D> {
    pub fn new(display: D) -> Self {
        Self {
            display: Arc::new(display),
        }
    }
}

fn stored_to_inspector(events: Vec<StoredEvent>, display: &dyn EventDisplay) -> Vec<InspectorEvent> {
    events.iter().map(|e| e.to_inspector_event(display)).collect()
}

#[Object]
impl<D: EventDisplay + 'static> CausalInspectorQuery<D> {
    /// Paginated reverse-chronological event listing with optional filters.
    async fn inspector_events(
        &self,
        ctx: &Context<'_>,
        limit: Option<i32>,
        cursor: Option<i64>,
        search: Option<String>,
        from: Option<DateTime<Utc>>,
        to: Option<DateTime<Utc>>,
        correlation_id: Option<String>,
        aggregate_key: Option<String>,
    ) -> Result<InspectorEventsPage> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;
        let lim = (limit.unwrap_or(50) as usize).min(200);

        let query = EventQuery {
            limit: lim,
            cursor,
            search,
            from,
            to,
            correlation_id,
            aggregate_key,
        };

        let stored = read_model
            .list_events(&query)
            .await
            .map_err(|e| async_graphql::Error::new(format!("Failed to load events: {e}")))?;

        let next_cursor = if stored.len() == lim {
            stored.last().map(|e| e.seq)
        } else {
            None
        };

        let events = stored_to_inspector(stored, self.display.as_ref());
        Ok(InspectorEventsPage { events, next_cursor })
    }

    /// Walk the causal tree for an event (ancestors + descendants).
    async fn inspector_causal_tree(&self, ctx: &Context<'_>, seq: i64) -> Result<InspectorCausalTree> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let (stored, root_seq) = read_model
            .causal_tree(seq)
            .await
            .map_err(|e| async_graphql::Error::new(format!("Failed to load causal tree: {e}")))?;

        let events = stored_to_inspector(stored, self.display.as_ref());
        Ok(InspectorCausalTree { events, root_seq })
    }

    /// Fetch all events sharing a correlation_id (causal flow DAG viewer).
    async fn inspector_causal_flow(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<InspectorCausalFlow> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let stored = read_model
            .causal_flow(&correlation_id)
            .await
            .map_err(|e| async_graphql::Error::new(format!("Failed to load causal flow: {e}")))?;

        let events = stored_to_inspector(stored, self.display.as_ref());
        Ok(InspectorCausalFlow { events })
    }

    /// Fetch reactor logs for a specific event + reactor.
    async fn inspector_reactor_logs(
        &self,
        ctx: &Context<'_>,
        event_id: String,
        reactor_id: String,
    ) -> Result<Vec<ReactorLog>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let event_uuid = uuid::Uuid::parse_str(&event_id)
            .map_err(|e| async_graphql::Error::new(format!("Invalid event_id: {e}")))?;

        let entries = read_model
            .reactor_logs(event_uuid, &reactor_id)
            .await
            .map_err(|e| async_graphql::Error::new(format!("Failed to load reactor logs: {e}")))?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorLog {
                event_id: r.event_id.to_string(),
                reactor_id: r.reactor_id,
                level: r.level,
                message: r.message,
                data: r.data,
                logged_at: r.logged_at,
            })
            .collect())
    }

    /// Fetch all reactor logs for a correlation chain.
    async fn inspector_reactor_logs_by_correlation(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<ReactorLog>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_logs_by_correlation(&correlation_id)
            .await
            .map_err(|e| async_graphql::Error::new(format!("Failed to load reactor logs: {e}")))?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorLog {
                event_id: r.event_id.to_string(),
                reactor_id: r.reactor_id,
                level: r.level,
                message: r.message,
                data: r.data,
                logged_at: r.logged_at,
            })
            .collect())
    }

    /// Fetch reactor describe() blocks for a correlation chain.
    async fn inspector_reactor_descriptions(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<ReactorDescription>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_descriptions(&correlation_id)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load reactor descriptions: {e}"))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorDescription {
                reactor_id: r.reactor_id,
                blocks: r.description,
            })
            .collect())
    }

    /// Fetch per-event aggregate state timeline for a correlation chain.
    async fn inspector_aggregate_timeline(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<AggregateTimelineEntry>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .aggregate_state_timeline(&correlation_id)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!(
                    "Failed to load aggregate state timeline: {e}"
                ))
            })?;

        // Group by seq → AggregateTimelineEntry with multiple AggregateStateEntry
        let mut by_seq: std::collections::BTreeMap<
            i64,
            (String, String, Vec<AggregateStateEntry>),
        > = std::collections::BTreeMap::new();

        for entry in entries {
            let row = by_seq
                .entry(entry.seq)
                .or_insert_with(|| (entry.event_id.to_string(), entry.event_type.clone(), Vec::new()));
            row.2.push(AggregateStateEntry {
                key: entry.aggregate_key,
                state: entry.state,
            });
        }

        Ok(by_seq
            .into_iter()
            .map(|(seq, (event_id, event_type, aggregates))| {
                AggregateTimelineEntry {
                    seq,
                    event_id,
                    event_type,
                    aggregates,
                }
            })
            .collect())
    }

    /// Fetch per-event reactor description snapshots for a correlation chain.
    async fn inspector_reactor_description_snapshots(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<ReactorDescriptionSnapshot>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_description_snapshots(&correlation_id)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!(
                    "Failed to load reactor description snapshots: {e}"
                ))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorDescriptionSnapshot {
                seq: r.seq,
                event_id: r.event_id.to_string(),
                reactor_id: r.reactor_id,
                blocks: r.description,
            })
            .collect())
    }

    /// List all correlation chains with summary stats (paginated).
    async fn inspector_correlations(
        &self,
        ctx: &Context<'_>,
        search: Option<String>,
        limit: Option<i32>,
        cursor: Option<String>,
    ) -> Result<CorrelationSummaryPage> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;
        let lim = (limit.unwrap_or(50) as usize).min(200);

        let cursor_ts = cursor
            .as_deref()
            .map(|s| s.parse::<DateTime<Utc>>())
            .transpose()
            .map_err(|e| async_graphql::Error::new(format!("Invalid cursor: {e}")))?;

        let entries = read_model
            .list_correlations(search.as_deref(), lim, cursor_ts)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load correlations: {e}"))
            })?;

        let next_cursor = if entries.len() == lim {
            entries.last().map(|e| e.last_ts.to_rfc3339())
        } else {
            None
        };

        let correlations = entries
            .into_iter()
            .map(|r| CorrelationSummary {
                correlation_id: r.correlation_id,
                event_count: r.event_count,
                first_ts: r.first_ts,
                last_ts: r.last_ts,
                root_event_type: r.root_event_type,
                has_errors: r.has_errors,
            })
            .collect();

        Ok(CorrelationSummaryPage { correlations, next_cursor })
    }

    /// Derive the reactor dependency graph from the event log.
    async fn inspector_reactor_dependencies(
        &self,
        ctx: &Context<'_>,
    ) -> Result<Vec<ReactorDependency>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_dependencies()
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load reactor dependencies: {e}"))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorDependency {
                reactor_id: r.reactor_id,
                input_event_types: r.input_event_types,
                output_event_types: r.output_event_types,
            })
            .collect())
    }

    /// List all known aggregate keys.
    async fn inspector_aggregate_keys(&self, ctx: &Context<'_>) -> Result<Vec<String>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        read_model
            .list_aggregate_keys()
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load aggregate keys: {e}"))
            })
    }

    /// Fetch aggregate lifecycle — all state snapshots for a specific aggregate across correlations.
    async fn inspector_aggregate_lifecycle(
        &self,
        ctx: &Context<'_>,
        aggregate_key: String,
        limit: Option<i32>,
    ) -> Result<Vec<AggregateLifecycleEntry>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;
        let lim = (limit.unwrap_or(100) as usize).min(500);

        let entries = read_model
            .aggregate_lifecycle(&aggregate_key, lim)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load aggregate lifecycle: {e}"))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| AggregateLifecycleEntry {
                seq: r.seq,
                event_id: r.event_id.to_string(),
                event_type: r.event_type,
                ts: r.ts,
                correlation_id: r.correlation_id,
                aggregate_key: r.aggregate_key,
                state: r.state,
            })
            .collect())
    }

    /// Fetch per-attempt execution history for a correlation chain.
    async fn inspector_reactor_attempts(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<ReactorAttempt>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_attempt_history(&correlation_id)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load reactor attempts: {e}"))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorAttempt {
                event_id: r.event_id.to_string(),
                reactor_id: r.reactor_id,
                correlation_id: r.correlation_id,
                attempt: r.attempt,
                status: r.status,
                error: r.error,
                started_at: r.started_at,
                completed_at: r.completed_at,
            })
            .collect())
    }

    /// Fetch aggregated reactor execution outcomes for a correlation chain.
    async fn inspector_reactor_outcomes(
        &self,
        ctx: &Context<'_>,
        correlation_id: String,
    ) -> Result<Vec<ReactorOutcome>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?;

        let entries = read_model
            .reactor_outcomes(&correlation_id)
            .await
            .map_err(|e| {
                async_graphql::Error::new(format!("Failed to load reactor outcomes: {e}"))
            })?;

        Ok(entries
            .into_iter()
            .map(|r| ReactorOutcome {
                reactor_id: r.reactor_id,
                status: r.status,
                error: r.error,
                attempts: r.attempts,
                started_at: r.started_at,
                completed_at: r.completed_at,
                triggering_event_ids: r.triggering_event_ids,
            })
            .collect())
    }
}

/// Generic inspector subscription — streams live events.
///
/// # Context requirements
///
/// The async-graphql context must contain:
/// - `Arc<dyn InspectorReadModel>` — for catch-up queries
/// - `tokio::sync::broadcast::Sender<StoredEvent>` — for live event stream
pub struct CausalInspectorSubscription<D: EventDisplay> {
    display: Arc<D>,
}

impl<D: EventDisplay> CausalInspectorSubscription<D> {
    pub fn new(display: D) -> Self {
        Self {
            display: Arc::new(display),
        }
    }
}

#[Subscription]
impl<D: EventDisplay + 'static> CausalInspectorSubscription<D> {
    /// Stream live events. If `last_seq` is provided, replays missed events first.
    async fn inspector_event_added(
        &self,
        ctx: &Context<'_>,
        last_seq: Option<i64>,
    ) -> Result<impl Stream<Item = InspectorEvent>> {
        let read_model = ctx.data::<Arc<dyn InspectorReadModel>>()?.clone();
        let tx = ctx
            .data::<tokio::sync::broadcast::Sender<StoredEvent>>()?
            .clone();
        let mut rx = tx.subscribe();
        let display = Arc::clone(&self.display);

        Ok(async_stream::stream! {
            let mut high_water: i64 = 0;

            // Catch-up phase
            if let Some(start) = last_seq {
                let catch_up_start = start + 1;
                match read_model.events_from_seq(catch_up_start, 500).await {
                    Ok(events) => {
                        for event in events {
                            if event.seq > high_water {
                                high_water = event.seq;
                            }
                            yield event.to_inspector_event(display.as_ref());
                        }
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "Subscription catch-up query failed");
                    }
                }
            }

            // Live phase
            loop {
                match rx.recv().await {
                    Ok(event) => {
                        if event.seq > high_water {
                            high_water = event.seq;
                            yield event.to_inspector_event(display.as_ref());
                        }
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(missed = n, "Subscription receiver lagged");
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                        break;
                    }
                }
            }
        })
    }
}
