use std::collections::{HashMap, VecDeque};
use std::str::FromStr;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use jacquard_common::DefaultStr;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::string::{AtUri, Cid};
use serde::Deserialize;
use smallvec::SmallVec;
use tokio::sync::oneshot;
use tokio::time::Instant;

// a settlement only has to outlive the round trip from the writer's pds back to us
const RETENTION: Duration = Duration::from_secs(30);
// best effort, counted in distinct at-uris so a writer rewriting one record
// cant push everyone else out
const CAPACITY: usize = 4096;
// enough revisions to still answer a writer whose record was superseded mid flight
const MAX_REVISIONS: usize = 8;
// the detail can echo caller-controlled bytes (parse errors)
const DETAIL_LIMIT: usize = 200;

/// why a record bobbin saw will not show up in its index
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Rejection {
    UnknownCollection,
    Undecodable,
    MissingBody,
    NotIndexed,
}

impl Rejection {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::UnknownCollection => "unknownCollection",
            Self::Undecodable => "undecodable",
            Self::MissingBody => "missingBody",
            Self::NotIndexed => "notIndexed",
        }
    }
}

/// what bobbin did with a record once it came off the firehose
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RecordOutcome {
    Indexed,
    Deleted,
    Rejected {
        reason: Rejection,
        detail: Option<Box<str>>,
    },
}

impl RecordOutcome {
    pub fn rejected(reason: Rejection, detail: Option<&str>) -> Self {
        Self::Rejected {
            reason,
            detail: detail.map(clamp),
        }
    }

    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::Indexed => "indexed",
            Self::Deleted => "deleted",
            Self::Rejected { .. } => "rejected",
        }
    }

    pub const fn rejection(&self) -> Option<Rejection> {
        match self {
            Self::Rejected { reason, .. } => Some(*reason),
            Self::Indexed | Self::Deleted => None,
        }
    }

    pub fn detail(&self) -> Option<&str> {
        match self {
            Self::Rejected { detail, .. } => detail.as_deref(),
            Self::Indexed | Self::Deleted => None,
        }
    }
}

fn clamp(s: &str) -> Box<str> {
    let mut end = DETAIL_LIMIT.min(s.len());
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    s[..end].into()
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct ParsedCid(IpldCid);

impl From<IpldCid> for ParsedCid {
    fn from(cid: IpldCid) -> Self {
        Self(cid)
    }
}

impl FromStr for ParsedCid {
    type Err = <IpldCid as FromStr>::Err;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        s.parse::<IpldCid>().map(Self)
    }
}

impl serde::Serialize for ParsedCid {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.collect_str(&self.0)
    }
}

impl<'de> Deserialize<'de> for ParsedCid {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let raw = Cid::<DefaultStr>::deserialize(deserializer)?;
        raw.to_ipld().map(Self).map_err(serde::de::Error::custom)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Settlement {
    pub source: AtUri<DefaultStr>,
    pub cid: Option<ParsedCid>,
    pub outcome: RecordOutcome,
}

#[derive(Debug, Default)]
pub struct Settlements {
    inner: Mutex<Inner>,
}

#[derive(Debug, Default)]
struct Inner {
    // per-revision history so a superseded record can still be resolved
    recent: HashMap<AtUri<DefaultStr>, SmallVec<[Recent; 1]>>,
    // first-seen order, exactly one entry per recent key
    order: VecDeque<AtUri<DefaultStr>>,
    waiters: HashMap<AtUri<DefaultStr>, Vec<Waiter>>,
    next_seq: u64,
}

#[derive(Debug)]
struct Recent {
    at: Instant,
    cid: Option<ParsedCid>,
    outcome: RecordOutcome,
}

#[derive(Debug)]
struct Waiter {
    seq: u64,
    cid: Option<ParsedCid>,
    tx: oneshot::Sender<Settlement>,
}

pub enum Watch {
    Settled(Settlement),
    /// the registration's Drop cancels the wait, so timeouts and
    /// abandoned requests don't leave a waiter behind
    Waiting(Registration, oneshot::Receiver<Settlement>),
}

/// one entry in the waiter list. dropping it removes the entry, so a
/// settlement that arrives later is only remembered, not sent
pub struct Registration {
    watch: Arc<Settlements>,
    source: AtUri<DefaultStr>,
    seq: u64,
}

impl Drop for Registration {
    fn drop(&mut self) {
        let mut inner = self.watch.lock();
        let Some(waiting) = inner.waiters.get_mut(&self.source) else {
            return;
        };
        waiting.retain(|w| w.seq != self.seq);
        if waiting.is_empty() {
            inner.waiters.remove(&self.source);
        }
    }
}

impl Settlements {
    pub fn new() -> Self {
        Self::default()
    }

    fn lock(&self) -> MutexGuard<'_, Inner> {
        self.inner.lock().unwrap_or_else(|e| e.into_inner())
    }

    pub fn settle(&self, settlement: Settlement) {
        let now = Instant::now();
        let winners = {
            let mut inner = self.lock();
            let winners = inner.take_matching(&settlement);
            inner.remember(now, settlement.clone());
            inner.prune(now);
            winners
        };
        for tx in winners {
            let _ = tx.send(settlement.clone());
        }
    }

    pub fn watch(self: &Arc<Self>, source: &AtUri<DefaultStr>, cid: Option<ParsedCid>) -> Watch {
        let now = Instant::now();
        let mut inner = self.lock();
        inner.prune(now);
        if let Some(recent) = inner.find(source, cid) {
            return Watch::Settled(Settlement {
                source: source.clone(),
                cid: recent.cid,
                outcome: recent.outcome.clone(),
            });
        }
        let seq = inner.next_seq();
        let (tx, rx) = oneshot::channel();
        inner
            .waiters
            .entry(source.clone())
            .or_default()
            .push(Waiter { seq, cid, tx });
        let registration = Registration {
            watch: self.clone(),
            source: source.clone(),
            seq,
        };
        Watch::Waiting(registration, rx)
    }
}

impl Inner {
    fn next_seq(&mut self) -> u64 {
        self.next_seq += 1;
        self.next_seq
    }

    fn take_matching(&mut self, s: &Settlement) -> Vec<oneshot::Sender<Settlement>> {
        if let Some(waiting) = self.waiters.remove(&s.source) {
            let (hit, miss): (Vec<_>, Vec<_>) = waiting
                .into_iter()
                .partition(|w| w.cid.is_none_or(|want| s.cid == Some(want)));
            if !miss.is_empty() {
                self.waiters.insert(s.source.clone(), miss);
            }
            hit.into_iter().map(|w| w.tx).collect()
        } else {
            Vec::new()
        }
    }

    fn find(&self, source: &AtUri<DefaultStr>, cid: Option<ParsedCid>) -> Option<&Recent> {
        self.recent
            .get(source)?
            .iter()
            .rev()
            .find(|r| cid.is_none_or(|want| r.cid == Some(want)))
    }

    fn remember(&mut self, now: Instant, settlement: Settlement) {
        let entries = match self.recent.entry(settlement.source) {
            std::collections::hash_map::Entry::Vacant(v) => {
                self.order.push_back(v.key().clone());
                v.insert(SmallVec::new())
            }
            std::collections::hash_map::Entry::Occupied(o) => o.into_mut(),
        };
        entries.push(Recent {
            at: now,
            cid: settlement.cid,
            outcome: settlement.outcome,
        });
        entries.retain(|r| now.duration_since(r.at) < RETENTION);
        while entries.len() > MAX_REVISIONS {
            entries.remove(0);
        }
    }

    fn prune(&mut self, now: Instant) {
        while let Some(source) = self.order.pop_front() {
            let over = self.recent.len() > CAPACITY;
            let keep = match self.recent.get_mut(&source) {
                None => false,
                Some(entries) => {
                    entries.retain(|r| now.duration_since(r.at) < RETENTION);
                    !over && !entries.is_empty()
                }
            };
            if keep {
                self.order.push_front(source);
                break;
            }
            self.recent.remove(&source);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn uri(rkey: &str) -> AtUri<DefaultStr> {
        AtUri::new_owned(format!("at://did:plc:alice/sh.tangled.feed.star/{rkey}")).unwrap()
    }

    const CID_A: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
    const CID_B: &str = "bafkreigh2akiscaildc7gnvtklbsfhdgwz72eolmpckbqr5ej26byp3uli";

    fn cid(s: &str) -> ParsedCid {
        s.parse().unwrap()
    }

    // a structurally valid cidv1 whose digest is a counter, so every mint differs
    fn mint(i: u32) -> ParsedCid {
        let mut raw = vec![0x01u8, 0x55, 0x12, 0x20];
        raw.extend_from_slice(&[0u8; 28]);
        raw.extend_from_slice(&i.to_be_bytes());
        ParsedCid::from(IpldCid::try_from(raw.as_slice()).unwrap())
    }

    fn indexed(rkey: &str, c: Option<ParsedCid>) -> Settlement {
        Settlement {
            source: uri(rkey),
            cid: c,
            outcome: RecordOutcome::Indexed,
        }
    }

    #[test]
    fn a_long_detail_is_clamped_on_a_char_boundary() {
        let long = "\u{1f600}".repeat(200);
        let outcome = RecordOutcome::rejected(Rejection::Undecodable, Some(&long));
        let detail = outcome.detail().expect("detail kept");
        assert!(detail.len() <= DETAIL_LIMIT);
        assert!(long.starts_with(detail));
    }

    fn settled(watch: Watch) -> Option<Settlement> {
        match watch {
            Watch::Settled(s) => Some(s),
            Watch::Waiting(..) => None,
        }
    }

    #[tokio::test(start_paused = true)]
    async fn settlement_before_the_ask_is_still_there() {
        let watch = Arc::new(Settlements::new());
        watch.settle(indexed("a", Some(cid(CID_A))));
        let got = settled(watch.watch(&uri("a"), Some(cid(CID_A)))).expect("already settled");
        assert_eq!(got.outcome, RecordOutcome::Indexed);
    }

    #[tokio::test(start_paused = true)]
    async fn a_waiter_wakes_on_its_own_cid() {
        let watch = Arc::new(Settlements::new());
        let Watch::Waiting(_reg, rx) = watch.watch(&uri("a"), Some(cid(CID_B))) else {
            panic!("nothing has settled yet");
        };
        watch.settle(indexed("a", Some(cid(CID_A))));
        watch.settle(Settlement {
            source: uri("a"),
            cid: Some(cid(CID_B)),
            outcome: RecordOutcome::rejected(Rejection::Undecodable, None),
        });
        let got = rx.await.expect("waiter woken");
        assert_eq!(
            got.outcome,
            RecordOutcome::rejected(Rejection::Undecodable, None)
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_cidless_waiter_takes_the_next_settlement() {
        let watch = Arc::new(Settlements::new());
        let Watch::Waiting(_reg, rx) = watch.watch(&uri("a"), None) else {
            panic!("nothing has settled yet");
        };
        watch.settle(Settlement {
            source: uri("a"),
            cid: None,
            outcome: RecordOutcome::Deleted,
        });
        assert_eq!(rx.await.unwrap().outcome, RecordOutcome::Deleted);
    }

    // a second write to the same uri must not eat the answer the first
    // writer is still waiting for
    #[tokio::test(start_paused = true)]
    async fn a_superseded_revision_keeps_its_own_outcome() {
        let watch = Arc::new(Settlements::new());
        watch.settle(Settlement {
            source: uri("a"),
            cid: Some(cid(CID_A)),
            outcome: RecordOutcome::rejected(Rejection::Undecodable, None),
        });
        watch.settle(indexed("a", Some(cid(CID_B))));
        let old = settled(watch.watch(&uri("a"), Some(cid(CID_A)))).expect("older revision");
        assert_eq!(
            old.outcome,
            RecordOutcome::rejected(Rejection::Undecodable, None)
        );
        let new = settled(watch.watch(&uri("a"), Some(cid(CID_B)))).expect("newer revision");
        assert_eq!(new.outcome, RecordOutcome::Indexed);
        let any = settled(watch.watch(&uri("a"), None)).expect("newest wins without a cid");
        assert_eq!(any.cid, Some(cid(CID_B)));
    }

    #[tokio::test(start_paused = true)]
    async fn an_unrelated_revision_does_not_wake_a_waiter() {
        let watch = Arc::new(Settlements::new());
        let Watch::Waiting(_reg, mut rx) = watch.watch(&uri("a"), Some(cid(CID_B))) else {
            panic!("nothing has settled yet");
        };
        watch.settle(indexed("a", Some(cid(CID_A))));
        assert!(rx.try_recv().is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn dropping_the_registration_takes_the_waiter_back_out() {
        let watch = Arc::new(Settlements::new());
        let Watch::Waiting(reg, rx) = watch.watch(&uri("a"), None) else {
            panic!("nothing has settled yet");
        };
        drop(reg);
        assert!(watch.lock().waiters.is_empty());
        drop(rx);
    }

    #[tokio::test(start_paused = true)]
    async fn settlements_expire() {
        let watch = Arc::new(Settlements::new());
        watch.settle(indexed("a", None));
        tokio::time::advance(RETENTION + Duration::from_secs(1)).await;
        watch.settle(indexed("b", None));
        assert!(settled(watch.watch(&uri("a"), None)).is_none());
        assert!(settled(watch.watch(&uri("b"), None)).is_some());
    }

    #[tokio::test(start_paused = true)]
    async fn a_later_revision_outlives_an_expired_earlier_one() {
        let watch = Arc::new(Settlements::new());
        watch.settle(indexed("a", Some(cid(CID_A))));
        tokio::time::advance(RETENTION - Duration::from_secs(1)).await;
        watch.settle(indexed("a", Some(cid(CID_B))));
        tokio::time::advance(Duration::from_secs(2)).await;
        watch.settle(indexed("z", None));
        assert!(settled(watch.watch(&uri("a"), Some(cid(CID_A)))).is_none());
        let got = settled(watch.watch(&uri("a"), Some(cid(CID_B)))).expect("newer survives");
        assert_eq!(got.outcome, RecordOutcome::Indexed);
    }

    #[tokio::test(start_paused = true)]
    async fn the_recent_map_stays_bounded() {
        let watch = Arc::new(Settlements::new());
        let keys = vec!["a", "b", "c", "d", "e"];
        for key in keys {
            watch.settle(indexed(key, None));
        }
        for i in 0..(MAX_REVISIONS * 3) {
            watch.settle(indexed("c", Some(mint(i as u32))));
        }
        for i in 0..3 {
            watch.settle(indexed("a", Some(mint(i as u32))));
        }
        let inner = watch.lock();
        assert_eq!(inner.order.len(), inner.recent.len());
        for entries in inner.recent.values() {
            assert!(entries.len() <= MAX_REVISIONS);
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_rewrite_flood_does_not_evict_other_records() {
        let watch = Arc::new(Settlements::new());
        watch.settle(indexed("b", None));
        for i in 0..(CAPACITY + 100) {
            watch.settle(indexed("a", Some(mint(i as u32))));
        }
        let watch_b = watch.watch(&uri("b"), None);
        assert!(matches!(watch_b, Watch::Settled(_)));
        let inner = watch.lock();
        assert_eq!(inner.order.len(), inner.recent.len());
        assert_eq!(inner.order.len(), 2);
    }

    #[tokio::test(start_paused = true)]
    async fn only_the_last_few_revisions_of_a_record_survive() {
        let watch = Arc::new(Settlements::new());
        let u = uri("a");
        for i in 0..(MAX_REVISIONS + 5) {
            watch.settle(indexed("a", Some(mint(i as u32))));
        }
        for i in 0..5 {
            assert!(
                settled(watch.watch(&u, Some(mint(i as u32)))).is_none(),
                "cid c{i} should be gone"
            );
        }
        for i in 5..(MAX_REVISIONS + 5) {
            let s = settled(watch.watch(&u, Some(mint(i as u32)))).expect("should be findable");
            assert_eq!(s.cid, Some(mint(i as u32)));
        }
    }

    #[tokio::test(start_paused = true)]
    async fn too_many_records_evicts_the_oldest() {
        let watch = Arc::new(Settlements::new());
        for i in 0..(CAPACITY + 10) {
            watch.settle(indexed(&format!("r{i}"), None));
        }
        let inner = watch.lock();
        assert!(inner.recent.len() <= CAPACITY);
        drop(inner);
        assert!(
            settled(watch.watch(&uri("r0"), None)).is_none(),
            "first should be gone"
        );
        assert!(
            settled(watch.watch(&uri(&format!("r{}", CAPACITY + 9)), None)).is_some(),
            "last should survive"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn expires_after_retention_with_no_pressure() {
        let watch = Arc::new(Settlements::new());
        watch.settle(indexed("a", None));
        tokio::time::advance(RETENTION + Duration::from_secs(1)).await;
        assert!(settled(watch.watch(&uri("a"), None)).is_none());
    }
}