//! the writes a full backfill makes: the new repo's records merged against the repo's stored
//! records, both read in key order, so neither side is ever held whole.

use std::cmp::Ordering;
use std::iter::Fuse;

use cid::Cid as IpldCid;
use miette::Result;
use crate::mst::MstKey;
use tracing::trace;

/// one write that brings the stored records in line with the new repo
#[derive(Debug)]
pub(crate) enum RecordOp<K, T> {
    /// a record that `stored` does not have
    Create { key: MstKey, record: T },
    /// a record whose cid differs from the stored one
    Update { key: MstKey, record: T },
    /// a stored record the new repo doesn't have
    Delete(K),
}

/// the writes that turn `stored` into `new`, in key order. a record whose cid didn't change
/// is left alone.
///
/// `new` yields each record's mst key, its cid and whatever the caller needs to write it.
/// `stored` yields each stored record's key after the did prefix and its cid. a heads key
/// ends in the same `collection/rkey` text as the mst key, so both sides sort the same way.
pub(crate) fn diff<N, S, K, T>(new: N, stored: S) -> Diff<N, S, K, T>
where
    N: Iterator<Item = Result<(MstKey, IpldCid, T)>>,
    S: Iterator<Item = Result<(K, IpldCid)>>,
    K: AsRef<[u8]>,
{
    Diff {
        new: new.fuse(),
        stored: stored.fuse(),
        next_new: None,
        next_stored: None,
        last_new: None,
        failed: false,
    }
}

pub(crate) struct Diff<N: Iterator, S: Iterator, K, T> {
    new: Fuse<N>,
    stored: Fuse<S>,
    next_new: Option<(MstKey, IpldCid, T)>,
    next_stored: Option<(K, IpldCid)>,
    last_new: Option<MstKey>,
    failed: bool,
}

impl<N, S, K, T> Diff<N, S, K, T>
where
    N: Iterator<Item = Result<(MstKey, IpldCid, T)>>,
    S: Iterator<Item = Result<(K, IpldCid)>>,
    K: AsRef<[u8]>,
{
    fn step(&mut self) -> Result<Option<RecordOp<K, T>>> {
        loop {
            if self.next_new.is_none() {
                self.next_new = self.pull_new()?;
            }
            if self.next_stored.is_none() {
                self.next_stored = self.stored.next().transpose()?;
            }
            let (new, stored) = match (self.next_new.take(), self.next_stored.take()) {
                (None, None) => return Ok(None),
                (Some((key, _, record)), None) => {
                    return Ok(Some(RecordOp::Create { key, record }));
                }
                (None, Some((stored, _))) => return Ok(Some(RecordOp::Delete(stored))),
                (Some(new), Some(stored)) => (new, stored),
            };
            match new.0.as_bytes().cmp(stored.0.as_ref()) {
                Ordering::Less => {
                    self.next_stored = Some(stored);
                    let (key, _, record) = new;
                    return Ok(Some(RecordOp::Create { key, record }));
                }
                Ordering::Greater => {
                    self.next_new = Some(new);
                    return Ok(Some(RecordOp::Delete(stored.0)));
                }
                Ordering::Equal if new.1 != stored.1 => {
                    let (key, _, record) = new;
                    return Ok(Some(RecordOp::Update { key, record }));
                }
                Ordering::Equal => trace!(key = %new.0, cid = %new.1, "skip unchanged record"),
            }
        }
    }

    /// the next new record. a merge against records out of order would delete and recreate
    /// them, so that's an error instead
    fn pull_new(&mut self) -> Result<Option<(MstKey, IpldCid, T)>> {
        let Some(next) = self.new.next().transpose()? else {
            return Ok(None);
        };
        if let Some(last) = self.last_new.as_ref().filter(|last| **last >= next.0) {
            miette::bail!("record {} does not sort after {last}", next.0);
        }
        self.last_new = Some(next.0.clone());
        Ok(Some(next))
    }
}

impl<N, S, K, T> Iterator for Diff<N, S, K, T>
where
    N: Iterator<Item = Result<(MstKey, IpldCid, T)>>,
    S: Iterator<Item = Result<(K, IpldCid)>>,
    K: AsRef<[u8]>,
{
    type Item = Result<RecordOp<K, T>>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.failed {
            return None;
        }
        let step = self.step();
        self.failed = step.is_err();
        step.transpose()
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use smol_str::{SmolStr, ToSmolStr};

    use super::*;
    use crate::db::keys;
    use crate::db::types::{DbAction, DbRkey, DbTid};
    use crate::mst::tests::cid;

    /// every stored record in a map, each new record taken out of it, and whatever stays behind
    /// deleted. returns (key, action) in key order
    fn oracle(
        new: &[(String, IpldCid)],
        stored: &[(String, IpldCid)],
    ) -> Vec<(String, &'static str)> {
        let mut existing: HashMap<(SmolStr, DbRkey), IpldCid> = stored
            .iter()
            .map(|(key, cid)| {
                let (collection, rkey) = keys::split_record_suffix(key.as_bytes()).unwrap();
                ((collection.into(), rkey), *cid)
            })
            .collect();
        let mut ops = Vec::new();
        for (key, cid) in new {
            let (collection, rkey) = crate::ops::parse_path(key).unwrap();
            let path = (SmolStr::new(collection), DbRkey::new(rkey));
            let action = existing
                .remove(&path)
                .map_or(Some(DbAction::Create), |existing| {
                    (existing != *cid).then_some(DbAction::Update)
                });
            if let Some(action) = action {
                ops.push((key.clone(), action.as_str()));
            }
        }
        ops.extend(
            existing.into_keys().map(|(collection, rkey)| {
                (format!("{collection}/{}", rkey.to_smolstr()), "delete")
            }),
        );
        ops.sort();
        ops
    }

    fn merged(
        new: &[(String, IpldCid)],
        stored: &[(String, IpldCid)],
    ) -> Vec<(String, &'static str)> {
        let new = new
            .iter()
            .map(|(key, cid)| Ok((MstKey::new(key), *cid, ())));
        let stored = stored.iter().map(|(key, cid)| Ok((key.clone(), *cid)));
        diff(new, stored)
            .map(|op| match op.unwrap() {
                RecordOp::Create { key, .. } => (key.to_string(), "create"),
                RecordOp::Update { key, .. } => (key.to_string(), "update"),
                RecordOp::Delete(key) => (key, "delete"),
            })
            .collect()
    }

    /// xorshift, so a failing case can be rerun from its seed
    struct Rng(u64);

    impl Rng {
        fn next(&mut self) -> u64 {
            self.0 ^= self.0 << 13;
            self.0 ^= self.0 >> 7;
            self.0 ^= self.0 << 17;
            self.0
        }

        fn below(&mut self, n: usize) -> usize {
            (self.next() % n as u64) as usize
        }
    }

    #[test]
    fn the_merge_makes_the_same_writes_as_the_map_it_replaced() {
        // collections where one is a prefix of another or differs by a char either side of
        // `/`, which is where byte order and (collection, rkey) order could part
        let collections = [
            "a.b.c",
            "a.b.c.d",
            "a.b.cd",
            "a.b-c.d",
            "app.bsky.feed.post",
        ];
        let strings = [
            "self",
            "a",
            "a.b",
            "a-b",
            "a_b",
            "~",
            "z:z",
            "3k",
            "2222222222222x",
        ];
        for seed in 1..=2000u64 {
            let mut rng = Rng(seed.wrapping_mul(0x9e37_79b9_7f4a_7c15));
            let tids = (0..4)
                .map(|_| DbTid::new_from_bytes((rng.next() >> 1).to_be_bytes()).to_smolstr())
                .collect::<Vec<_>>();
            let mut keys = (0..12)
                .map(|_| {
                    let collection = collections[rng.below(collections.len())];
                    let rkey = match rng.below(2) {
                        0 => tids[rng.below(tids.len())].to_string(),
                        _ => strings[rng.below(strings.len())].to_string(),
                    };
                    format!("{collection}/{rkey}")
                })
                .collect::<Vec<_>>();
            keys.sort();
            keys.dedup();
            // either side is empty now and then, and cids come from a few values so
            // unchanged records are common
            let mut pick = |odds: usize| {
                let mut picked = Vec::new();
                for key in &keys {
                    if odds > 0 && rng.below(odds + 1) > 0 {
                        picked.push((key.clone(), cid(rng.below(3) as u8)));
                    }
                }
                picked
            };
            let new = pick(seed as usize % 4);
            let stored = pick(seed as usize % 3);

            let merged = merged(&new, &stored);
            assert_eq!(merged, oracle(&new, &stored), "seed {seed}");
            assert!(
                merged.windows(2).all(|pair| pair[0].0 < pair[1].0),
                "seed {seed}: {merged:?}"
            );
        }
    }

    #[test]
    fn deletes_land_between_the_puts_in_key_order() {
        let new = [
            ("a.b.c/1".to_string(), cid(1)),
            ("a.b.c/3".to_string(), cid(1)),
        ];
        let stored = [
            ("a.b.c/2".to_string(), cid(1)),
            ("a.b.c/3".to_string(), cid(2)),
        ];
        assert_eq!(
            merged(&new, &stored),
            [
                ("a.b.c/1".to_string(), "create"),
                ("a.b.c/2".to_string(), "delete"),
                ("a.b.c/3".to_string(), "update"),
            ]
        );
    }

    #[test]
    fn new_records_out_of_order_are_an_error() {
        let new = [("a.b.c/2", cid(1)), ("a.b.c/1", cid(1))]
            .map(|(key, cid)| Ok((MstKey::new(key), cid, ())));
        let mut ops = diff(
            new.into_iter(),
            std::iter::empty::<Result<(Vec<u8>, IpldCid)>>(),
        );
        assert!(matches!(ops.next(), Some(Ok(RecordOp::Create { .. }))));
        let err = ops.next().unwrap().unwrap_err();
        assert!(err.to_string().contains("does not sort after"), "{err}");
        assert!(ops.next().is_none());
    }
}