use std::collections::{BTreeSet, HashMap};
use std::sync::Arc;

use chrono::{DateTime, Utc};
use fabro_types::{Run, RunId, RunProjection};
use tokio::sync::Mutex;

use crate::ListRunsQuery;
use crate::run_state::build_summary;

#[derive(Debug, Clone)]
pub struct CachedRunProjection {
    pub run_id:     RunId,
    pub summary:    Run,
    pub projection: Arc<RunProjection>,
    pub last_seq:   u32,
}

impl CachedRunProjection {
    pub(crate) fn from_projection(run_id: RunId, projection: RunProjection, last_seq: u32) -> Self {
        let summary = build_summary(&projection, &run_id);
        Self {
            run_id,
            summary,
            projection: Arc::new(projection),
            last_seq,
        }
    }
}

#[derive(Debug, Default)]
pub(crate) struct RunProjectionCache {
    state: Mutex<RunProjectionCacheState>,
}

#[derive(Debug, Default)]
struct RunProjectionCacheState {
    entries:            HashMap<RunId, CachedRunProjection>,
    children_by_parent: HashMap<RunId, BTreeSet<RunId>>,
}

impl RunProjectionCacheState {
    fn replace_all(&mut self, entries: Vec<CachedRunProjection>) {
        self.entries.clear();
        self.children_by_parent.clear();
        for entry in entries {
            self.insert(entry);
        }
    }

    fn insert(&mut self, entry: CachedRunProjection) {
        let run_id = entry.run_id;
        let parent_id = entry.summary.parent_id;
        if let Some(previous) = self.entries.insert(run_id, entry) {
            self.remove_parent_index(&previous);
        }
        if let Some(parent_id) = parent_id {
            self.children_by_parent
                .entry(parent_id)
                .or_default()
                .insert(run_id);
        }
    }

    fn remove(&mut self, run_id: &RunId) {
        if let Some(entry) = self.entries.remove(run_id) {
            self.remove_parent_index(&entry);
        }
    }

    fn remove_parent_index(&mut self, entry: &CachedRunProjection) {
        let Some(parent_id) = entry.summary.parent_id else {
            return;
        };
        self.remove_parent_link(&parent_id, &entry.run_id);
    }

    fn remove_parent_link(&mut self, parent_id: &RunId, run_id: &RunId) {
        let Some(children) = self.children_by_parent.get_mut(parent_id) else {
            return;
        };
        children.remove(run_id);
        if children.is_empty() {
            self.children_by_parent.remove(parent_id);
        }
    }

    fn count_children(&self, run_id: &RunId) -> u64 {
        self.children_by_parent
            .get(run_id)
            .map_or(0, |children| children.len() as u64)
    }

    fn with_children_count(&self, mut entry: CachedRunProjection) -> CachedRunProjection {
        entry.summary.children_count = self.count_children(&entry.run_id);
        entry
    }
}

/// Apply read-time overlays to a cached entry. Pure: does not touch the cache
/// state, so it can run outside the cache mutex.
fn apply_read_overlays(entry: &mut CachedRunProjection, now: DateTime<Utc>) {
    // `Conclusion::timing` is the authoritative terminal snapshot and is
    // already present in cached terminal summaries. Only fill missing timing
    // with the best-effort live projection.
    if entry.summary.timing.is_none() {
        entry.summary.timing = entry.projection.live_run_timing(now);
    }
}

impl RunProjectionCache {
    pub(crate) async fn replace_all(&self, entries: Vec<CachedRunProjection>) {
        self.state.lock().await.replace_all(entries);
    }

    pub(crate) async fn replace(&self, entry: CachedRunProjection) {
        self.state.lock().await.insert(entry);
    }

    pub(crate) async fn list(
        &self,
        query: &ListRunsQuery,
        now: DateTime<Utc>,
    ) -> Vec<CachedRunProjection> {
        let entries = {
            let state = self.state.lock().await;
            let raw = match query.parent_id {
                Some(parent_id) => state
                    .children_by_parent
                    .get(&parent_id)
                    .into_iter()
                    .flat_map(|children| children.iter())
                    .filter_map(|run_id| state.entries.get(run_id).cloned())
                    .collect::<Vec<_>>(),
                None => state.entries.values().cloned().collect::<Vec<_>>(),
            };
            raw.into_iter()
                .map(|entry| state.with_children_count(entry))
                .collect::<Vec<_>>()
        };
        let mut entries = entries
            .into_iter()
            .filter(|entry| {
                let created_at = entry.run_id.created_at();
                if query.start.is_some_and(|start| created_at < start) {
                    return false;
                }
                if query.end.is_some_and(|end| created_at > end) {
                    return false;
                }
                true
            })
            .collect::<Vec<_>>();
        // Apply per-entry live overlays outside the cache mutex, after any
        // date filtering so skipped entries do not sum stage timings.
        for entry in &mut entries {
            apply_read_overlays(entry, now);
        }
        entries.sort_by(|left, right| {
            right
                .run_id
                .created_at()
                .cmp(&left.run_id.created_at())
                .then_with(|| right.run_id.cmp(&left.run_id))
        });
        entries
    }

    pub(crate) async fn get(&self, run_id: &RunId) -> Option<CachedRunProjection> {
        let state = self.state.lock().await;
        state
            .entries
            .get(run_id)
            .cloned()
            .map(|entry| state.with_children_count(entry))
    }

    /// Projection and last sequence for `run_id`, without the summary clone
    /// and children count that `get` computes under the cache mutex.
    pub(crate) async fn projection_snapshot(
        &self,
        run_id: &RunId,
    ) -> Option<(Arc<RunProjection>, u32)> {
        self.state
            .lock()
            .await
            .entries
            .get(run_id)
            .map(|entry| (Arc::clone(&entry.projection), entry.last_seq))
    }

    /// Run ids whose latest explicit pull request creation is still pending,
    /// oldest request first. Clones only ids and timestamps, so callers can
    /// poll on an interval without materializing run summaries.
    pub(crate) async fn pending_pull_request_creations(&self) -> Vec<RunId> {
        let mut pending = self
            .state
            .lock()
            .await
            .entries
            .values()
            .filter_map(|entry| {
                let creation = entry.projection.pull_request_creation.as_ref()?;
                creation
                    .is_pending()
                    .then_some((creation.requested_at, entry.run_id))
            })
            .collect::<Vec<_>>();
        pending.sort_unstable();
        pending.into_iter().map(|(_, run_id)| run_id).collect()
    }

    pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
        let mut entry = {
            let state = self.state.lock().await;
            state
                .entries
                .get(run_id)
                .cloned()
                .map(|entry| state.with_children_count(entry))?
        };
        apply_read_overlays(&mut entry, now);
        Some(entry.summary)
    }

    pub(crate) async fn remove(&self, run_id: &RunId) {
        self.state.lock().await.remove(run_id);
    }
}
