use crate::db::keys;
use fjall::compaction::filter::Context;
use lsm_tree::compaction::{CompactionFilter, Factory};
use lsm_tree::compaction::{ItemAccessor, Verdict};
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};

pub struct CountsGcFilterFactory {
    pub delta_gc_watermark: Arc<AtomicU64>,
}

struct CountsGcFilter {
    delta_gc_watermark: Arc<AtomicU64>,
}

impl CompactionFilter for CountsGcFilter {
    fn filter_item(&mut self, item: ItemAccessor<'_>, _: &Context) -> lsm_tree::Result<Verdict> {
        let key = item.key();

        if key.starts_with(keys::COUNT_DELTA_PREFIX)
            && let Ok((id, _)) = keys::parse_count_delta_key(key)
            && id <= self.delta_gc_watermark.load(Ordering::Relaxed)
        {
            return Ok(Verdict::Destroy);
        }

        Ok(Verdict::Keep)
    }
}

impl Factory for CountsGcFilterFactory {
    fn name(&self) -> &str {
        "counts_gc"
    }

    fn make_filter(&self, _: &Context) -> Box<dyn CompactionFilter> {
        Box::new(CountsGcFilter {
            delta_gc_watermark: self.delta_gc_watermark.clone(),
        })
    }
}

/// drops history entries whose death rev is older than the configured
/// retention window. the rev is a TID in the key tail, so the verdict is
/// purely key-local: age is decodable without touching the value, and a
/// filter on `history` can never see a head, so there is nothing to
/// disambiguate. reclamation is lazy - it lands when compaction next covers
/// the range.
pub struct HistoryRetentionFilterFactory {
    pub ttl: Arc<OnceLock<Option<std::time::Duration>>>,
}

struct HistoryRetentionFilter {
    cutoff_micros: Option<i64>,
}

/// decodes the timestamp micros out of the trailing 8-byte big-endian TID:
/// 53 bits of micros over 10 bits of clock id (jacquard encodes
/// `(micros << 10) | clkid`; shifting by 11 would halve the timestamp and
/// make fresh entries look decades old)
fn tid_micros(key: &[u8]) -> Option<i64> {
    let rev: [u8; 8] = key.get(key.len().checked_sub(8)?..)?.try_into().ok()?;
    Some((u64::from_be_bytes(rev) >> 10) as i64)
}

fn retention_cutoff_micros(now_micros: i64, ttl: std::time::Duration) -> i64 {
    let ttl_micros = i64::try_from(ttl.as_micros()).unwrap_or(i64::MAX);
    now_micros.saturating_sub(ttl_micros)
}

impl CompactionFilter for HistoryRetentionFilter {
    fn filter_item(&mut self, item: ItemAccessor<'_>, _: &Context) -> lsm_tree::Result<Verdict> {
        match (tid_micros(item.key()), self.cutoff_micros) {
            (Some(micros), Some(cutoff_micros)) if micros < cutoff_micros => Ok(Verdict::Destroy),
            _ => Ok(Verdict::Keep),
        }
    }
}

impl Factory for HistoryRetentionFilterFactory {
    fn name(&self) -> &str {
        "history_retention"
    }

    fn make_filter(&self, _: &Context) -> Box<dyn CompactionFilter> {
        let cutoff_micros = self
            .ttl
            .get()
            .copied()
            .flatten()
            .map(|ttl| retention_cutoff_micros(chrono::Utc::now().timestamp_micros(), ttl));
        Box::new(HistoryRetentionFilter { cutoff_micros })
    }
}

#[cfg(all(test, feature = "indexer"))]
mod tests {
    use super::*;
    use miette::{IntoDiagnostic, Result};
    use std::time::Duration;

    const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";

    fn tid_at(micros: i64) -> crate::db::types::DbTid {
        // same layout jacquard uses: (micros << 10) | clkid
        crate::db::types::DbTid::new_from_bytes((((micros as u64) << 10) | 0x2A).to_be_bytes())
    }

    #[test]
    fn tid_micros_decodes_real_jacquard_tids() {
        // regression: the filter must decode tids produced by jacquard, not
        // by a private encoder that could drift from it
        let tid = jacquard_common::types::string::Tid::now_0();
        let db_tid = crate::db::types::DbTid::from(&tid);
        let mut key = b"record-key".to_vec();
        key.extend_from_slice(db_tid.as_bytes());
        let decoded = tid_micros(&key).unwrap();
        let expected = chrono::Utc::now().timestamp_micros();
        assert!(
            (decoded - expected).abs() < 5_000_000,
            "decoded {decoded} should be within 5s of now ({expected})"
        );
    }

    #[test]
    fn tid_micros_decodes_the_key_tail() {
        let micros = 1_754_000_000_123_456i64;
        let mut key = b"record-key".to_vec();
        key.extend_from_slice((((micros as u64) << 10) | 0x3FF).to_be_bytes().as_ref());
        assert_eq!(tid_micros(&key), Some(micros));
        assert_eq!(tid_micros(b"short"), None);
        assert_eq!(tid_micros(b""), None);
    }

    #[test]
    fn huge_ttl_cannot_wrap_the_cutoff_into_the_future() {
        let now = chrono::Utc::now().timestamp_micros();
        assert!(retention_cutoff_micros(now, Duration::MAX) < 0);
    }

    fn record_key() -> Result<Vec<u8>> {
        Ok(crate::db::keys::record_key(
            &jacquard_common::types::string::Did::new_static(DID).into_diagnostic()?,
            "app.bsky.feed.post",
            &crate::db::types::DbRkey::new("3kzbif5moe22m"),
        ))
    }

    fn db_with_history_ttl(ttl: Option<Duration>, tmp: &std::path::Path) -> Result<crate::db::Db> {
        let mut cfg = crate::config::Config::default();
        cfg.database_path = tmp.to_path_buf();
        cfg.history_ttl = ttl;
        crate::db::Db::open(&cfg)
    }

    #[test]
    fn retention_filter_drops_expired_history_entries() -> Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let db = db_with_history_ttl(Some(Duration::from_secs(3600)), tmp.path())?;

        let now_micros = chrono::Utc::now().timestamp_micros();
        let expired = tid_at(now_micros - 2 * 3600 * 1_000_000);
        let fresh = tid_at(now_micros);
        let key = record_key()?;

        let mut batch = db.inner.batch();
        db.indexer.stage_history(
            &mut batch,
            crate::db::keys::history_key(&key, &expired),
            b"old".as_slice(),
        );
        db.indexer.stage_history(
            &mut batch,
            crate::db::keys::history_key(&key, &fresh),
            b"new".as_slice(),
        );
        batch.commit().into_diagnostic()?;

        db.indexer
            .history
            .rotate_memtable_and_wait()
            .into_diagnostic()?;
        db.indexer.history.major_compact().into_diagnostic()?;

        let entries: Vec<_> = db
            .indexer
            .history
            .iter()
            .filter_map(|g| g.into_inner().ok())
            .collect();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].1.as_ref(), b"new");

        Ok(())
    }

    #[test]
    fn without_retention_config_history_survives_compaction() -> Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let db = db_with_history_ttl(None, tmp.path())?;

        let ancient = tid_at(1_000_000_000);
        let key = record_key()?;

        let mut batch = db.inner.batch();
        db.indexer.stage_history(
            &mut batch,
            crate::db::keys::history_key(&key, &ancient),
            b"old".as_slice(),
        );
        batch.commit().into_diagnostic()?;

        db.indexer
            .history
            .rotate_memtable_and_wait()
            .into_diagnostic()?;
        db.indexer.history.major_compact().into_diagnostic()?;

        assert_eq!(db.indexer.history.iter().count(), 1);

        Ok(())
    }
}