use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use bobbin_knot_proxy::MirrorProxy;
use chrono::{DateTime, Datelike, Utc};
use futures::StreamExt;
use http::HeaderMap;
use jacquard_common::DefaultStr;
use jacquard_common::types::did::Did;
use jacquard_common::types::nsid::Nsid;
use serde::Deserialize;
use tokio::sync::{Notify, Semaphore};

const COMMIT_STATS_NSID: &str = "sh.tangled.git.temp2.getCommitStats";
const FETCH_MONTHS: u32 = 24;
const MAX_RESPONSE_BYTES: usize = 64 * 1024;
const DEFAULT_CAPACITY: usize = 4_096;
const DEFAULT_FRESHNESS: Duration = Duration::from_secs(30);
const FAILURE_RETRY: Duration = Duration::from_secs(5);
const MAX_CONCURRENT_REFRESHES: usize = 16;

#[derive(Clone, Debug, Deserialize)]
struct CommitMonth {
    start: String,
    commits: i64,
}

#[derive(Debug, Deserialize)]
struct CommitStats {
    months: Vec<CommitMonth>,
}

#[derive(Clone)]
struct Entry {
    counts: BTreeMap<i32, i64>,
    refresh_after: Option<Instant>,
    refreshing: bool,
    loaded: bool,
    notify: Arc<Notify>,
    access: u64,
}

struct CacheState {
    entries: HashMap<String, Entry>,
    access: u64,
}

pub(crate) struct CommitStatsCache {
    state: Mutex<CacheState>,
    capacity: usize,
    freshness: Duration,
    refreshes: Arc<Semaphore>,
    capacity_notify: Arc<Notify>,
}

impl Default for CommitStatsCache {
    fn default() -> Self {
        Self::new(DEFAULT_CAPACITY, DEFAULT_FRESHNESS)
    }
}

impl CommitStatsCache {
    fn new(capacity: usize, freshness: Duration) -> Self {
        Self {
            state: Mutex::new(CacheState {
                entries: HashMap::new(),
                access: 0,
            }),
            capacity: capacity.max(1),
            freshness,
            refreshes: Arc::new(Semaphore::new(MAX_CONCURRENT_REFRESHES)),
            capacity_notify: Arc::new(Notify::new()),
        }
    }

    pub(crate) async fn get(
        self: &Arc<Self>,
        mirror: Option<Arc<MirrorProxy>>,
        actor: &Did<DefaultStr>,
    ) -> BTreeMap<i32, i64> {
        let Some(mirror) = mirror else {
            return BTreeMap::new();
        };
        let key = actor.to_string();

        loop {
            let (counts, loaded, should_refresh, waiter) = {
                let mut state = self
                    .state
                    .lock()
                    .expect("commit stats cache mutex poisoned");
                state.access = state.access.wrapping_add(1);
                let access = state.access;
                let capacity_waiter = if !state.entries.contains_key(&key) {
                    if evict_one(&mut state.entries, self.capacity) {
                        state.entries.insert(
                            key.clone(),
                            Entry {
                                counts: BTreeMap::new(),
                                refresh_after: None,
                                refreshing: false,
                                loaded: false,
                                notify: Arc::new(Notify::new()),
                                access,
                            },
                        );
                        None
                    } else {
                        Some(Arc::clone(&self.capacity_notify).notified_owned())
                    }
                } else {
                    None
                };
                if let Some(waiter) = capacity_waiter {
                    (BTreeMap::new(), false, false, Some(waiter))
                } else {
                    let entry = state.entries.get_mut(&key).expect("entry inserted above");
                    entry.access = access;
                    if !entry.loaded && entry.refreshing {
                        (
                            BTreeMap::new(),
                            false,
                            false,
                            Some(Arc::clone(&entry.notify).notified_owned()),
                        )
                    } else {
                        let fresh = entry
                            .refresh_after
                            .is_some_and(|refresh_after| Instant::now() < refresh_after);
                        let should_refresh = !fresh && !entry.refreshing;
                        if should_refresh {
                            entry.refreshing = true;
                        }
                        (entry.counts.clone(), entry.loaded, should_refresh, None)
                    }
                }
            };

            if loaded {
                if should_refresh {
                    match Arc::clone(&self.refreshes).try_acquire_owned() {
                        Ok(permit) => {
                            let cache = Arc::clone(self);
                            let mirror = Arc::clone(&mirror);
                            let key = key.clone();
                            let mut guard = RefreshGuard::new(Arc::clone(&cache), key.clone());
                            tokio::spawn(async move {
                                let result = fetch_commit_stats(&mirror, &key).await;
                                drop(permit);
                                match result {
                                    Ok(counts) => cache.finish(key, Some(counts)),
                                    Err(error) => {
                                        tracing::warn!(actor = %key, %error, "refreshing commit stats failed");
                                        cache.finish(key, None);
                                    }
                                }
                                guard.disarm();
                            });
                        }
                        Err(_) => self.finish(key.clone(), None),
                    }
                }
                return counts;
            }

            if let Some(waiter) = waiter {
                waiter.await;
                continue;
            }

            let mut guard = RefreshGuard::new(Arc::clone(self), key.clone());
            let Ok(permit) = Arc::clone(&self.refreshes).acquire_owned().await else {
                self.finish(key, None);
                guard.disarm();
                return BTreeMap::new();
            };
            let result = fetch_commit_stats(&mirror, &key).await;
            drop(permit);
            return match result {
                Ok(counts) => {
                    self.finish(key, Some(counts.clone()));
                    guard.disarm();
                    counts
                }
                Err(error) => {
                    tracing::warn!(actor = %key, %error, "loading commit stats failed");
                    self.finish(key, None);
                    guard.disarm();
                    BTreeMap::new()
                }
            };
        }
    }

    fn finish(&self, key: String, counts: Option<BTreeMap<i32, i64>>) {
        let mut state = self
            .state
            .lock()
            .expect("commit stats cache mutex poisoned");
        state.access = state.access.wrapping_add(1);
        let access = state.access;
        let Some(entry) = state.entries.get_mut(&key) else {
            return;
        };
        entry.refreshing = false;
        entry.loaded = true;
        entry.access = access;
        match counts {
            Some(counts) => {
                entry.counts = counts;
                entry.refresh_after = Some(Instant::now() + self.freshness);
            }
            None => entry.refresh_after = Some(Instant::now() + FAILURE_RETRY),
        }
        let notify = Arc::clone(&entry.notify);
        drop(state);
        notify.notify_waiters();
        self.capacity_notify.notify_waiters();
    }

    fn cancel_refresh(&self, key: &str) {
        let mut state = self
            .state
            .lock()
            .expect("commit stats cache mutex poisoned");
        let Some(entry) = state.entries.get_mut(key) else {
            return;
        };
        entry.refreshing = false;
        let notify = Arc::clone(&entry.notify);
        drop(state);
        notify.notify_waiters();
        self.capacity_notify.notify_waiters();
    }

    #[cfg(test)]
    fn seed(&self, actor: &str, counts: BTreeMap<i32, i64>, refresh_after: Option<Instant>) {
        self.state
            .lock()
            .expect("commit stats cache mutex poisoned")
            .entries
            .insert(
                actor.to_owned(),
                Entry {
                    counts,
                    refresh_after,
                    refreshing: false,
                    loaded: true,
                    notify: Arc::new(Notify::new()),
                    access: 0,
                },
            );
    }
}

struct RefreshGuard {
    cache: Arc<CommitStatsCache>,
    key: String,
    active: bool,
}

impl RefreshGuard {
    fn new(cache: Arc<CommitStatsCache>, key: String) -> Self {
        Self {
            cache,
            key,
            active: true,
        }
    }

    fn disarm(&mut self) {
        self.active = false;
    }
}

impl Drop for RefreshGuard {
    fn drop(&mut self) {
        if self.active {
            self.cache.cancel_refresh(&self.key);
        }
    }
}

fn evict_one(entries: &mut HashMap<String, Entry>, capacity: usize) -> bool {
    if entries.len() < capacity {
        return true;
    }
    let Some(key) = entries
        .iter()
        .filter(|(_, entry)| !entry.refreshing)
        .min_by_key(|(_, entry)| entry.access)
        .map(|(key, _)| key.clone())
    else {
        return false;
    };
    entries.remove(&key);
    true
}

async fn fetch_commit_stats(
    mirror: &MirrorProxy,
    actor: &str,
) -> Result<BTreeMap<i32, i64>, String> {
    let nsid = Nsid::new_static(COMMIT_STATS_NSID).expect("commit stats NSID is valid");
    let months = FETCH_MONTHS.to_string();
    let response = mirror
        .forward_raw(
            &nsid,
            &[("actor", actor), ("months", months.as_str())],
            HeaderMap::new(),
        )
        .await
        .map_err(|error| error.to_string())?;
    if !response.status().is_success() {
        let status = response.status();
        response.discard().await;
        return Err(format!("mirror returned {status}"));
    }

    let mut body = response.into_body_stream();
    let mut bytes = Vec::new();
    while let Some(chunk) = body.next().await {
        let chunk = chunk.map_err(|error| error.to_string())?;
        if bytes.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES {
            return Err("mirror commit stats response is too large".to_owned());
        }
        bytes.extend_from_slice(&chunk);
    }
    let output: CommitStats = serde_json::from_slice(&bytes).map_err(|error| error.to_string())?;
    if output.months.len() > FETCH_MONTHS as usize {
        return Err("mirror returned too many commit stat months".to_owned());
    }
    output
        .months
        .into_iter()
        .try_fold(BTreeMap::new(), |mut counts, month| -> Result<_, String> {
            let start = DateTime::parse_from_rfc3339(&month.start)
                .map_err(|error| format!("invalid month start: {error}"))?
                .with_timezone(&Utc);
            if month.commits < 0 {
                return Err("mirror returned a negative commit count".to_owned());
            }
            counts.insert(month_number(start), month.commits);
            Ok(counts)
        })
}

pub(crate) fn month_number(value: DateTime<Utc>) -> i32 {
    value.year() * 12 + value.month0() as i32
}

#[cfg(test)]
mod tests {
    use super::*;
    use bobbin_runtime::{RuntimeHasher, SystemClock};
    use serde_json::json;
    use wiremock::matchers::{method, path, query_param};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    fn actor() -> Did<DefaultStr> {
        Did::new_static("did:plc:pushcacheactor").unwrap()
    }

    fn mirror(server: &MockServer) -> Arc<MirrorProxy> {
        Arc::new(
            MirrorProxy::new(
                &url::Url::parse(&server.uri()).unwrap(),
                Arc::new(SystemClock::new()),
                RuntimeHasher::default(),
            )
            .unwrap(),
        )
    }

    #[tokio::test]
    async fn a_cold_read_waits_for_the_first_value() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats"))
            .and(query_param("actor", actor().as_ref()))
            .and(query_param("months", "24"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "months": [{"start": "2026-09-01T00:00:00Z", "commits": 7}]
            })))
            .expect(1)
            .mount(&server)
            .await;

        let cache = Arc::new(CommitStatsCache::new(4, Duration::from_secs(60)));
        let counts = cache.get(Some(mirror(&server)), &actor()).await;
        assert_eq!(counts.get(&(2026 * 12 + 8)), Some(&7));
        assert_eq!(
            cache
                .get(Some(mirror(&server)), &actor())
                .await
                .get(&(2026 * 12 + 8)),
            Some(&7),
        );
        server.verify().await;
    }

    #[tokio::test]
    async fn rejects_more_months_than_requested() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "months": (0..=FETCH_MONTHS)
                    .map(|month| json!({
                        "start": format!("2024-{:02}-01T00:00:00Z", month % 12 + 1),
                        "commits": 1,
                    }))
                    .collect::<Vec<_>>()
            })))
            .mount(&server)
            .await;

        let error = fetch_commit_stats(&mirror(&server), actor().as_ref())
            .await
            .unwrap_err();
        assert!(error.contains("too many commit stat months"));
    }

    #[tokio::test]
    async fn concurrent_cold_reads_share_one_request() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_delay(Duration::from_millis(25))
                    .set_body_json(json!({
                        "months": [{"start": "2026-09-01T00:00:00Z", "commits": 7}]
                    })),
            )
            .expect(1)
            .mount(&server)
            .await;
        let cache = Arc::new(CommitStatsCache::new(4, Duration::from_secs(60)));
        let actor = actor();
        let first = cache.get(Some(mirror(&server)), &actor);
        let second = cache.get(Some(mirror(&server)), &actor);

        let (first, second) = tokio::join!(first, second);
        assert_eq!(first.get(&(2026 * 12 + 8)), Some(&7));
        assert_eq!(second.get(&(2026 * 12 + 8)), Some(&7));
        server.verify().await;
    }

    #[tokio::test]
    async fn a_cold_read_waits_for_a_bounded_refresh_permit() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({"months": []})))
            .expect(1)
            .mount(&server)
            .await;
        let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO));
        let permits = Arc::clone(&cache.refreshes)
            .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32)
            .await
            .unwrap();
        let read = tokio::spawn({
            let cache = Arc::clone(&cache);
            let mirror = mirror(&server);
            async move { cache.get(Some(mirror), &actor()).await }
        });

        for _ in 0..50 {
            if cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| entry.refreshing)
            {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert!(
            cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| entry.refreshing && !entry.loaded)
        );
        assert!(server.received_requests().await.unwrap().is_empty());
        drop(permits);
        assert!(read.await.unwrap().is_empty());
        server.verify().await;
    }

    #[tokio::test]
    async fn cancelling_a_cold_read_releases_waiters() {
        let server = MockServer::start().await;
        let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO));
        let permits = Arc::clone(&cache.refreshes)
            .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32)
            .await
            .unwrap();
        let read = tokio::spawn({
            let cache = Arc::clone(&cache);
            let mirror = mirror(&server);
            async move { cache.get(Some(mirror), &actor()).await }
        });
        for _ in 0..50 {
            if cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| entry.refreshing)
            {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert!(
            cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| entry.refreshing && !entry.loaded)
        );
        read.abort();
        assert!(read.await.unwrap_err().is_cancelled());
        assert!(
            cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| !entry.refreshing && !entry.loaded)
        );
        drop(permits);
    }

    #[tokio::test]
    async fn cold_entries_do_not_exceed_capacity_while_refreshes_wait() {
        let server = MockServer::start().await;
        let cache = Arc::new(CommitStatsCache::new(1, Duration::ZERO));
        let permits = Arc::clone(&cache.refreshes)
            .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32)
            .await
            .unwrap();
        let first_actor = Did::new_static("did:plc:firstcoldactor").unwrap();
        let second_actor = Did::new_static("did:plc:secondcoldactor").unwrap();
        let first = tokio::spawn({
            let cache = Arc::clone(&cache);
            let mirror = mirror(&server);
            async move { cache.get(Some(mirror), &first_actor).await }
        });
        for _ in 0..50 {
            if cache.state.lock().unwrap().entries.len() == 1 {
                break;
            }
            tokio::task::yield_now().await;
        }
        let second = tokio::spawn({
            let cache = Arc::clone(&cache);
            let mirror = mirror(&server);
            async move { cache.get(Some(mirror), &second_actor).await }
        });
        for _ in 0..50 {
            tokio::task::yield_now().await;
        }
        assert_eq!(cache.state.lock().unwrap().entries.len(), 1);

        first.abort();
        assert!(first.await.unwrap_err().is_cancelled());
        for _ in 0..50 {
            if cache
                .state
                .lock()
                .unwrap()
                .entries
                .contains_key("did:plc:secondcoldactor")
            {
                break;
            }
            tokio::task::yield_now().await;
        }
        let state = cache.state.lock().unwrap();
        assert_eq!(state.entries.len(), 1);
        assert!(state.entries.contains_key("did:plc:secondcoldactor"));
        drop(state);
        second.abort();
        assert!(second.await.unwrap_err().is_cancelled());
        drop(permits);
    }

    #[test]
    fn dropping_a_refresh_guard_clears_stale_refresh_state() {
        let cache = Arc::new(CommitStatsCache::new(1, Duration::ZERO));
        cache.seed(actor().as_ref(), BTreeMap::new(), Some(Instant::now()));
        cache
            .state
            .lock()
            .unwrap()
            .entries
            .get_mut(actor().as_ref())
            .unwrap()
            .refreshing = true;
        drop(RefreshGuard::new(Arc::clone(&cache), actor().to_string()));
        assert!(
            !cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .unwrap()
                .refreshing
        );
    }

    #[tokio::test]
    async fn a_failed_refresh_keeps_stale_counts() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats"))
            .respond_with(ResponseTemplate::new(503))
            .expect(1)
            .mount(&server)
            .await;

        let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO));
        cache.seed(
            actor().as_ref(),
            BTreeMap::from([(2026 * 12 + 8, 5)]),
            Some(Instant::now()),
        );
        let counts = cache.get(Some(mirror(&server)), &actor()).await;
        assert_eq!(counts.get(&(2026 * 12 + 8)), Some(&5));
        for _ in 0..50 {
            let refreshing = cache
                .state
                .lock()
                .unwrap()
                .entries
                .get(actor().as_ref())
                .is_some_and(|entry| entry.refreshing);
            if !refreshing {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            cache
                .get(Some(mirror(&server)), &actor())
                .await
                .get(&(2026 * 12 + 8)),
            Some(&5),
        );
        server.verify().await;
    }
}