use jacquard_common::types::string::Did;
use miette::{IntoDiagnostic, WrapErr};

use super::SEP;
use crate::db::types::{DbRkey, DbTid, TrimmedDid};

#[cfg(feature = "indexer_stream")]
pub const EVENT_WATERMARK_PREFIX: &[u8] = b"ewm|";

pub fn pending_key(id: u64) -> [u8; 8] {
    id.to_be_bytes()
}

#[cfg(feature = "indexer_stream")]
pub fn event_watermark_key(timestamp_secs: u64) -> Vec<u8> {
    let mut key = Vec::with_capacity(EVENT_WATERMARK_PREFIX.len() + 8);
    key.extend_from_slice(EVENT_WATERMARK_PREFIX);
    key.extend_from_slice(&timestamp_secs.to_be_bytes());
    key
}

/// segment separator in record keys: NUL terminates the trimmed DID, and in
/// history keys separates the record key from the death revision. NUL is
/// outside every segment's charset (trimmed DID bytes, NSID collections, and
/// rkey text), so it splits unambiguously.
pub const REC_SEP: u8 = 0x00;

/// length of the trimmed-DID segment at the start of a record or history key,
/// for both the current (NUL-terminated) and legacy (`|`-terminated) formats.
///
/// this cannot be a scan for NUL: plc segments are raw binary and may contain
/// NUL bytes, so the tag byte decides. web/other segments are text and cannot
/// contain NUL or `|`.
pub fn trimmed_did_len(key: &[u8]) -> Option<usize> {
    use crate::db::types::{TAG_PLC, TAG_WEB};
    match key.first()? {
        &TAG_PLC => (key.len() > 16).then_some(16),
        &TAG_WEB => key[1..]
            .iter()
            .position(|&b| b == REC_SEP || b == SEP)
            .map(|i| i + 1),
        _ => key.iter().position(|&b| b == REC_SEP || b == SEP),
    }
}

/// collection/rkey separator inside record keys. `/` is in neither charset,
/// and makes the key body read as `{collection}/{rkey}`, which sorts in
/// exactly MST order: TIDs and string rkeys interleave lexicographically,
/// and a record always sorts before records it prefix-matches.
pub const REC_PATH_SEP: u8 = b'/';

// prefix format: {DID} 00 (DID trimmed)
pub fn record_prefix_trimmed(did: &TrimmedDid) -> Vec<u8> {
    let mut prefix = Vec::with_capacity(did.len() + 1);
    did.write_to_vec(&mut prefix);
    prefix.push(REC_SEP);
    prefix
}

// prefix format: {DID} 00 (DID trimmed)
pub fn record_prefix_did(did: &Did) -> Vec<u8> {
    record_prefix_trimmed(&TrimmedDid::from(did))
}

// prefix format: {DID} 00 {collection} 2f
pub fn record_prefix_collection(did: &Did, collection: &str) -> Vec<u8> {
    let mut prefix = record_prefix_did(did);
    prefix.reserve(collection.len() + 1);
    prefix.extend_from_slice(collection.as_bytes());
    prefix.push(REC_PATH_SEP);
    prefix
}

// key format: {DID} 00 {collection} 2f {rkey text}
//
// the rkey is stored as its canonical text (13-char base32 for TIDs), not
// tagged bytes: text order is MST order, so a repo's records lay out in
// exactly MST walk order and `generate_car` no longer re-sorts.
pub fn record_key_trimmed(did: &TrimmedDid, collection: &str, rkey: &DbRkey) -> Vec<u8> {
    let mut key = record_prefix_trimmed(did);
    key.extend_from_slice(collection.as_bytes());
    key.push(REC_PATH_SEP);
    key.extend_from_slice(rkey.to_smolstr().as_bytes());
    key
}

// key format: {DID} 00 {collection} 2f {rkey text}
pub fn record_key(did: &Did, collection: &str, rkey: &DbRkey) -> Vec<u8> {
    record_key_trimmed(&TrimmedDid::from(did), collection, rkey)
}

/// parse an rkey from its text encoding. TID-shaped text becomes
/// [`DbRkey::Tid`], everything else [`DbRkey::Str`]; the text itself is the
/// lossless form, so the distinction only affects in-memory handling.
pub fn parse_rkey_text(raw: &[u8]) -> miette::Result<DbRkey> {
    let s = std::str::from_utf8(raw)
        .into_diagnostic()
        .wrap_err("record key has invalid rkey utf8")?;
    Ok(DbRkey::new(s))
}

/// split a record key's `{collection} 2f {rkey}` suffix (everything after the
/// `{DID} 00` prefix) into its parts.
pub fn split_record_suffix(suffix: &[u8]) -> miette::Result<(&str, DbRkey)> {
    let sep = suffix
        .iter()
        .position(|&b| b == REC_PATH_SEP)
        .ok_or_else(|| miette::miette!("record key missing collection/rkey separator"))?;
    let collection = std::str::from_utf8(&suffix[..sep])
        .into_diagnostic()
        .wrap_err("record key has invalid collection utf8")?;
    let rkey = parse_rkey_text(&suffix[sep + 1..])?;
    Ok((collection, rkey))
}

// key format: {record key} 00 {death_rev: u64 BE}
//
// a record's superseded bodies, keyed by the rev at which they died. the NUL
// sorts a record's history immediately after the record itself, and BE revs
// sort deaths chronologically within it.
pub fn history_key(record_key: &[u8], death_rev: &DbTid) -> Vec<u8> {
    let mut key = Vec::with_capacity(record_key.len() + 9);
    key.extend_from_slice(record_key);
    key.push(REC_SEP);
    key.extend_from_slice(death_rev.as_bytes());
    key
}

/// recover the record-key prefix from a canonical death-keyed history entry.
pub fn history_record_key(key: &[u8]) -> Option<&[u8]> {
    let record_end = key.len().checked_sub(9)?;
    (key.get(record_end) == Some(&REC_SEP)).then_some(&key[..record_end])
}

/// durable operator-redaction identity: `{record key} 00 {cid}`.
///
/// this lives in its own keyspace. keeping the cid in the key makes replay
/// suppression a point read rather than an unbounded scan of record history.
pub fn redaction_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
    record_cid_key(record_key, cid)
}

fn record_cid_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
    let cid = cid.to_bytes();
    let mut key = Vec::with_capacity(record_key.len() + 1 + cid.len());
    key.extend_from_slice(record_key);
    key.push(REC_SEP);
    key.extend_from_slice(&cid);
    key
}

/// split a `{record key} 00 {cid}` key and require hydrant's canonical
/// CIDv1/dag-cbor/sha256 record identity.
pub fn split_record_cid_key(key: &[u8]) -> Option<(&[u8], jacquard_common::types::cid::IpldCid)> {
    const CID_LEN: usize = 36;
    let record_end = key.len().checked_sub(CID_LEN + 1)?;
    if key.get(record_end) != Some(&REC_SEP) {
        return None;
    }
    let raw = &key[record_end + 1..];
    if !raw.starts_with(&[0x01, 0x71, 0x12, 0x20]) {
        return None;
    }
    let cid = cid::Cid::read_bytes(raw).ok()?;
    (cid.to_bytes().as_slice() == raw).then_some((&key[..record_end], cid))
}

/// exact body location for a pre-v10 pointer event.
///
/// the cid is part of the key because forks can produce two different bodies
/// for the same record and revision. unlike ordinary history, this finite
/// compatibility archive does not pretend an event's birth revision is a death
/// time.
#[cfg(feature = "indexer_stream")]
pub fn event_body_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
    record_cid_key(record_key, cid)
}

/// the range of history entries for `record_key` with death rev strictly
/// greater than `after`: (`key` 00 `after`, `key` 01).
///
/// read-side only: stream inflation is today's sole history reader
///
/// the NUL/01 byte pair brackets this record's entries exactly in byte order,
/// but keys of *longer* records that share this one as a prefix still fall
/// inside; filter hits with [`history_hit_for`].
#[cfg(feature = "indexer_stream")]
pub fn history_range_after(
    record_key: &[u8],
    after: &DbTid,
) -> (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>) {
    use std::ops::Bound;
    let mut lower = Vec::with_capacity(record_key.len() + 9);
    lower.extend_from_slice(record_key);
    lower.push(REC_SEP);
    lower.extend_from_slice(after.as_bytes());
    let mut upper = Vec::with_capacity(record_key.len() + 1);
    upper.extend_from_slice(record_key);
    upper.push(REC_SEP + 1);
    (Bound::Excluded(lower), Bound::Excluded(upper))
}

/// a history range hit only names this record if it is exactly
/// `{record key} 00 {rev}`; longer records sharing the prefix must be rejected.
#[cfg(feature = "indexer_stream")]
pub fn history_hit_for(record_key: &[u8], hit_key: &[u8]) -> bool {
    hit_key.len() == record_key.len() + 9 && hit_key.starts_with(record_key)
}

// key format: r|{DID}|{collection} (DID trimmed)
pub fn count_collection_key(did: &Did, collection: &str) -> Vec<u8> {
    let mut key = super::did_collection_prefix(did);
    key.extend_from_slice(collection.as_bytes());
    key
}

// key format: {DID}|{rev}
pub fn resync_buffer_key(did: &Did, rev: DbTid) -> Vec<u8> {
    let repo = TrimmedDid::from(did);
    let mut key = Vec::with_capacity(repo.len() + 1 + 8);
    repo.write_to_vec(&mut key);
    key.push(SEP);
    key.extend_from_slice(rev.as_bytes());
    key
}

// prefix format: {DID}| (DID trimmed)
pub fn resync_buffer_prefix(did: &Did) -> Vec<u8> {
    let repo = TrimmedDid::from(did);
    let mut prefix = Vec::with_capacity(repo.len() + 1);
    repo.write_to_vec(&mut prefix);
    prefix.push(SEP);
    prefix
}

/// key format: `ret|<did bytes>`
pub const CRAWLER_RETRY_PREFIX: &[u8] = b"ret|";

pub fn crawler_retry_key(did: &Did) -> Vec<u8> {
    let repo = TrimmedDid::from(did);
    let mut key = Vec::with_capacity(CRAWLER_RETRY_PREFIX.len() + repo.len());
    key.extend_from_slice(CRAWLER_RETRY_PREFIX);
    repo.write_to_vec(&mut key);
    key
}

pub fn crawler_retry_parse_key(key: &[u8]) -> miette::Result<TrimmedDid<'_>> {
    let did = key
        .strip_prefix(CRAWLER_RETRY_PREFIX)
        .ok_or_else(|| miette::miette!("invalid crawler retry key"))?;
    TrimmedDid::try_from(did)
}

pub const CRAWLER_CURSOR_PREFIX: &[u8] = b"crawler_cursor|";

pub fn crawler_cursor_key(relay: &str) -> Vec<u8> {
    let mut key = CRAWLER_CURSOR_PREFIX.to_vec();
    key.extend_from_slice(relay.as_bytes());
    key
}

pub const RELAY_FIRST_CURSOR_PREFIX: &[u8] = b"relay_first_cursor|";

pub fn relay_first_cursor_prefix(relay: &str) -> Vec<u8> {
    let mut prefix = Vec::with_capacity(RELAY_FIRST_CURSOR_PREFIX.len() + relay.len() + 1);
    prefix.extend_from_slice(RELAY_FIRST_CURSOR_PREFIX);
    prefix.extend_from_slice(relay.as_bytes());
    prefix.push(SEP);
    prefix
}

fn relay_first_cursor_key(relay: &str, suffix: &str) -> Vec<u8> {
    let mut key = relay_first_cursor_prefix(relay);
    key.extend_from_slice(suffix.as_bytes());
    key
}

pub fn relay_first_pass_key(relay: &str) -> Vec<u8> {
    relay_first_cursor_key(relay, "pass")
}

pub fn relay_first_handled_key(relay: &str) -> Vec<u8> {
    relay_first_cursor_key(relay, "handled")
}

pub fn relay_first_hosts_cursor_key(relay: &str) -> Vec<u8> {
    relay_first_cursor_key(relay, "hosts")
}

pub fn relay_first_pds_cursor_key(relay: &str, host: &str) -> Vec<u8> {
    relay_first_cursor_key(relay, &format!("pds|{host}"))
}

pub const BY_COLLECTION_CURSOR_PREFIX: &[u8] = b"by_collection_cursor|";

/// prefix for all by-collection cursors belonging to a given index URL.
pub fn by_collection_cursor_prefix(url: &str) -> Vec<u8> {
    let mut prefix = BY_COLLECTION_CURSOR_PREFIX.to_vec();
    prefix.extend_from_slice(url.as_bytes());
    prefix.push(SEP);
    prefix
}

pub fn by_collection_cursor_key(url: &str, collection: &str) -> Vec<u8> {
    let mut key = by_collection_cursor_prefix(url);
    key.extend_from_slice(collection.as_bytes());
    key
}

pub const CRAWLER_SOURCE_PREFIX: &[u8] = b"src|";

pub fn crawler_source_key(url: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(CRAWLER_SOURCE_PREFIX.len() + url.len());
    key.extend_from_slice(CRAWLER_SOURCE_PREFIX);
    key.extend_from_slice(url.as_bytes());
    key
}

// key format: {collection}|{cid_bytes}
pub fn block_key(collection: &str, cid: &[u8]) -> Vec<u8> {
    let mut key = Vec::with_capacity(collection.len() + 1 + cid.len());
    key.extend_from_slice(collection.as_bytes());
    key.push(SEP);
    key.extend_from_slice(cid);
    key
}

#[cfg(test)]
mod tests {
    use super::*;
    use jacquard_common::types::string::Tid;

    fn did() -> Did {
        Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap()
    }

    fn tid_rkey(s: &str) -> DbRkey {
        DbRkey::Tid(DbTid::from(&Tid::new(s).unwrap()))
    }

    #[test]
    fn record_key_roundtrips_collection_and_rkey() {
        let did = did();
        for rkey in [
            DbRkey::Str("self".into()),
            DbRkey::Str("abc".into()),
            DbRkey::Str("~tilde".into()),
            tid_rkey("3kzbif5moe22m"),
        ] {
            let key = record_key(&did, "app.bsky.feed.post", &rkey);
            let prefix = record_prefix_did(&did);
            assert!(key.starts_with(&prefix));
            let (col, parsed) = split_record_suffix(&key[prefix.len()..]).unwrap();
            assert_eq!(col, "app.bsky.feed.post");
            assert_eq!(parsed.to_smolstr(), rkey.to_smolstr());
        }
    }

    #[test]
    fn record_keys_sort_in_mst_order() {
        let did = did();
        let mut keys: Vec<Vec<u8>> = [
            ("app.bsky.actor.profile", DbRkey::Str("self".into())),
            ("app.bsky.feed.like", tid_rkey("3kzbif5moe22m")),
            ("app.bsky.feed.like", DbRkey::Str("abc".into())),
            ("app.bsky.feed.like", DbRkey::Str("abcd".into())),
            ("app.bsky.feed.post", tid_rkey("3kzbif5abc22a")),
            ("app.bsky.feed.post", DbRkey::Str("z-string".into())),
        ]
        .into_iter()
        .map(|(col, rk)| record_key(&did, col, &rk))
        .collect();
        let mut sorted = keys.clone();
        sorted.sort();
        // mst order is plain lexicographic order of `{collection}/{rkey}` text
        keys.sort_by_key(|k| {
            let prefix_len = record_prefix_did(&did).len();
            k[prefix_len..].to_vec()
        });
        assert_eq!(sorted, keys);
        // and the prefix pair nests correctly: "abc" before "abcd"
        let a = record_key(&did, "app.bsky.feed.like", &DbRkey::Str("abc".into()));
        let b = record_key(&did, "app.bsky.feed.like", &DbRkey::Str("abcd".into()));
        assert!(a < b);
    }

    #[test]
    fn new_record_keys_sort_before_legacy_keys() {
        // the v10 migration scans ascending and must only meet legacy keys:
        // new keys (`did 00`) sort before legacy keys (`did 7c`) of the same repo
        let did = did();
        let new = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("self".into()));
        let mut legacy = record_prefix_did(&did);
        *legacy.last_mut().unwrap() = super::SEP;
        legacy.extend_from_slice(b"app.bsky.feed.post|sself");
        assert!(new < legacy);
    }

    #[cfg(feature = "indexer_stream")]
    #[test]
    fn history_key_brackets_and_hit_guard() {
        let did = did();
        let rec = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abc".into()));
        let longer = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abcd".into()));
        let rev = DbTid::from(&Tid::new("3kzbif5moe22m").unwrap());
        let death = history_key(&rec, &rev);

        // a record's history sorts immediately after the record itself
        assert!(rec < death);

        // the range finds this record's deaths, including one at rev+1
        let later = DbTid::from(&Tid::new("3kzbif5moe22n").unwrap());
        let later_death = history_key(&rec, &later);
        let (lo, hi) = history_range_after(&rec, &rev);
        let in_range = |k: &Vec<u8>| {
            (match &lo {
                std::ops::Bound::Excluded(l) => k > l,
                _ => unreachable!(),
            }) && (match &hi {
                std::ops::Bound::Excluded(h) => k < h,
                _ => unreachable!(),
            })
        };
        assert!(in_range(&later_death));
        // a death exactly at `after` is excluded (strictly-after semantics)
        assert!(!in_range(&death));

        // a longer record sharing this one as a prefix sorts above the upper
        // bound (`rec 01` < `rec d`), so the bounded range only ever contains
        // this record's own `rec 00 rev` entries...
        let longer_death = history_key(&longer, &later);
        assert!(!in_range(&longer));
        assert!(!in_range(&longer_death));
        // ...and a different record entirely also sorts outside
        let other = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abd".into()));
        assert!(!in_range(&other));

        // the guard is defense in depth: even if a foreign key were hit (e.g.
        // via an unbounded seek), it is accepted only for this exact record
        assert!(history_hit_for(&rec, &death));
        assert!(!history_hit_for(&rec, &longer));
        assert!(!history_hit_for(&rec, &longer_death));
    }

    #[test]
    fn history_death_revs_sort_chronologically() {
        let did = did();
        let rec = record_key(&did, "app.bsky.feed.post", &tid_rkey("3kzbif5moe22m"));
        let early = history_key(&rec, &DbTid::from(&Tid::new("3kzbif5moe22m").unwrap()));
        let late = history_key(&rec, &DbTid::from(&Tid::new("3kzbif5mof33m").unwrap()));
        assert!(early < late);
    }
    #[test]
    fn indexer_key_codec_inventory_contracts() {
        let did = did();
        let trimmed = TrimmedDid::from(&did);
        let rkey = DbRkey::Str("stinkpot".into());
        let rev = DbTid::from(&Tid::new("3kzbif5moe22m").unwrap());

        assert_eq!(pending_key(42), 42_u64.to_be_bytes());
        #[cfg(feature = "indexer_stream")]
        assert_eq!(
            event_watermark_key(42),
            [EVENT_WATERMARK_PREFIX, 42_u64.to_be_bytes().as_slice()].concat()
        );

        let record_prefix = record_prefix_trimmed(&trimmed);
        assert_eq!(trimmed_did_len(&record_prefix), Some(trimmed.len()));
        assert_eq!(record_prefix, record_prefix_did(&did));
        let collection_prefix = record_prefix_collection(&did, "app.bsky.feed.post");
        let record = record_key_trimmed(&trimmed, "app.bsky.feed.post", &rkey);
        assert_eq!(record, record_key(&did, "app.bsky.feed.post", &rkey));
        assert!(record.starts_with(&collection_prefix));
        assert_eq!(parse_rkey_text(b"stinkpot").unwrap(), rkey);
        assert!(parse_rkey_text(&[0xff]).is_err());
        assert!(split_record_suffix(b"missing-separator").is_err());

        let history = history_key(&record, &rev);
        assert_eq!(history_record_key(&history), Some(record.as_slice()));
        assert_eq!(history_record_key(b"short"), None);

        let cid = jacquard_repo::mst::util::compute_cid(b"body").unwrap();
        let redaction = redaction_key(&record, &cid);
        assert_eq!(
            split_record_cid_key(&redaction),
            Some((record.as_slice(), cid))
        );
        assert!(split_record_cid_key(&record).is_none());
        #[cfg(feature = "indexer_stream")]
        assert_eq!(event_body_key(&record, &cid), redaction);

        let count_prefix = super::super::did_collection_prefix(&did);
        assert_eq!(
            count_collection_key(&did, "app.bsky.feed.post"),
            [count_prefix.as_slice(), b"app.bsky.feed.post"].concat()
        );
        let resync_prefix = resync_buffer_prefix(&did);
        assert_eq!(
            resync_buffer_key(&did, rev),
            [resync_prefix.as_slice(), rev.as_bytes()].concat()
        );

        let retry = crawler_retry_key(&did);
        assert_eq!(crawler_retry_parse_key(&retry).unwrap(), trimmed);
        assert!(crawler_retry_parse_key(b"bad").is_err());
        assert!(crawler_retry_parse_key(b"nope|did:web:example.com").is_err());

        assert_eq!(
            crawler_cursor_key("https://relay.example"),
            b"crawler_cursor|https://relay.example"
        );
        assert_eq!(
            relay_first_cursor_prefix("https://relay.example"),
            b"relay_first_cursor|https://relay.example|"
        );
        assert_eq!(
            relay_first_pass_key("https://relay.example"),
            b"relay_first_cursor|https://relay.example|pass"
        );
        assert_eq!(
            relay_first_handled_key("https://relay.example"),
            b"relay_first_cursor|https://relay.example|handled"
        );
        assert_eq!(
            relay_first_hosts_cursor_key("https://relay.example"),
            b"relay_first_cursor|https://relay.example|hosts"
        );
        assert_eq!(
            relay_first_pds_cursor_key("https://relay.example", "pds.example"),
            b"relay_first_cursor|https://relay.example|pds|pds.example"
        );
        assert_eq!(
            by_collection_cursor_prefix("https://relay.example"),
            b"by_collection_cursor|https://relay.example|"
        );
        assert_eq!(
            by_collection_cursor_key("https://relay.example", "app.bsky.feed.post"),
            b"by_collection_cursor|https://relay.example|app.bsky.feed.post"
        );
        assert_eq!(
            crawler_source_key("https://relay.example"),
            b"src|https://relay.example"
        );
        assert_eq!(
            block_key("app.bsky.feed.post", &[1, 2, 3]),
            b"app.bsky.feed.post|\x01\x02\x03"
        );
    }
}