//! Property-based invariants for `MemoryJobQueue`.

use bytes::Bytes;
use klieo_bus_memory::MemoryBus;
use klieo_core::bus::Job;
use proptest::prelude::*;
use std::time::Duration;
use tokio::runtime::Runtime;

#[derive(Debug, Clone)]
enum Op {
    Enqueue { payload: u8 },
    Claim,
    Ack,
    Nak { delay_ms: u64 },
    DeadLetter,
}

fn op_strategy() -> impl Strategy<Value = Op> {
    prop_oneof![
        any::<u8>().prop_map(|payload| Op::Enqueue { payload }),
        Just(Op::Claim),
        Just(Op::Ack),
        (0u64..50).prop_map(|delay_ms| Op::Nak { delay_ms }),
        Just(Op::DeadLetter),
    ]
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(1024))]

    #[test]
    fn no_enqueued_job_is_lost(ops in proptest::collection::vec(op_strategy(), 0..40)) {
        let rt = Runtime::new().unwrap();
        rt.block_on(async move {
            let bus = MemoryBus::new();
            let mut enqueued: Vec<u8> = Vec::new();
            let mut acked: Vec<u8> = Vec::new();
            let mut dead_lettered: Vec<u8> = Vec::new();

            // Subscribe to DLQ so we can count dead-lettered payloads.
            let mut dlq_sub = bus
                .pubsub
                .subscribe("q.dlq", klieo_core::ids::DurableName::new("dlq"))
                .await
                .unwrap();

            let mut current_claim: Option<klieo_core::bus::ClaimedJob> = None;

            for op in ops.iter() {
                match op {
                    Op::Enqueue { payload } => {
                        bus.jobs
                            .enqueue("q", Job::new(Bytes::from(vec![*payload])))
                            .await
                            .unwrap();
                        enqueued.push(*payload);
                    }
                    Op::Claim => {
                        if current_claim.is_none() {
                            current_claim = bus
                                .jobs
                                .claim("q", "w", Duration::from_millis(50))
                                .await
                                .unwrap();
                        }
                    }
                    Op::Ack => {
                        if let Some(c) = current_claim.take() {
                            let payload = c.payload[0];
                            c.ack().await.unwrap();
                            acked.push(payload);
                        }
                    }
                    Op::Nak { delay_ms } => {
                        if let Some(c) = current_claim.take() {
                            c.nak(Duration::from_millis(*delay_ms)).await.unwrap();
                        }
                    }
                    Op::DeadLetter => {
                        if let Some(c) = current_claim.take() {
                            c.dead_letter("test").await.unwrap();
                        }
                    }
                }
            }

            // Drain DLQ for dead-lettered payloads.
            use tokio_stream::StreamExt;
            tokio::time::sleep(Duration::from_millis(50)).await;
            while let Ok(Some(Ok(msg))) =
                tokio::time::timeout(Duration::from_millis(20), dlq_sub.next()).await
            {
                dead_lettered.push(msg.payload[0]);
            }

            // Drain remaining queue (everything still claimable).
            let mut remaining: Vec<u8> = Vec::new();
            while let Some(claimed) = bus
                .jobs
                .claim("q", "w_drain", Duration::from_secs(60))
                .await
                .unwrap()
            {
                remaining.push(claimed.payload[0]);
                claimed.ack().await.unwrap();
            }

            // Invariant: every enqueued payload is accounted for at least
            // once across acked / dead_lettered / remaining. Multiset
            // semantics — duplicates are legal at this layer (the
            // runtime guarantees at-least-once delivery, not
            // exactly-once). For each enqueued item, assert it appears
            // at least once across the three result vectors.
            let total: Vec<u8> = acked
                .iter()
                .chain(dead_lettered.iter())
                .chain(remaining.iter())
                .copied()
                .collect();
            for needle in &enqueued {
                prop_assert!(
                    total.contains(needle),
                    "enqueued payload {} disappeared (acked={:?}, dlq={:?}, remaining={:?})",
                    needle, acked, dead_lettered, remaining,
                );
            }
            Ok(())
        })?;
    }

    #[test]
    fn dedup_key_idempotent(payload in any::<u8>()) {
        let rt = Runtime::new().unwrap();
        rt.block_on(async move {
            let bus = MemoryBus::new();
            let mut j1 = Job::new(Bytes::from(vec![payload]));
            j1.dedup_key = Some("dk".into());
            let id1 = bus.jobs.enqueue("q", j1).await.unwrap();
            let mut j2 = Job::new(Bytes::from(vec![payload.wrapping_add(1)]));
            j2.dedup_key = Some("dk".into());
            let id2 = bus.jobs.enqueue("q", j2).await.unwrap();
            prop_assert_eq!(id1, id2);
            Ok(())
        })?;
    }

    #[test]
    fn nak_past_max_attempts_routes_to_dlq(
        payload in any::<u8>(),
        max_attempts in 1u32..5,
    ) {
        let rt = Runtime::new().unwrap();
        rt.block_on(async move {
            let bus = MemoryBus::new();
            let mut dlq_sub = bus
                .pubsub
                .subscribe("q.dlq", klieo_core::ids::DurableName::new("dlq"))
                .await
                .unwrap();

            let mut job = Job::new(Bytes::from(vec![payload]));
            job.max_attempts = Some(max_attempts);
            bus.jobs.enqueue("q", job).await.unwrap();

            // Nak exactly max_attempts times; the (max_attempts)-th nak
            // routes to DLQ.
            for _ in 0..max_attempts {
                let claim = bus
                    .jobs
                    .claim("q", "w", Duration::from_secs(1))
                    .await
                    .unwrap()
                    .expect("job must be claimable");
                claim.nak(Duration::from_millis(0)).await.unwrap();
            }

            // Queue must be empty.
            let none = bus
                .jobs
                .claim("q", "w", Duration::from_secs(1))
                .await
                .unwrap();
            prop_assert!(none.is_none(), "queue must be empty after DLQ routing");

            // DLQ must contain the payload exactly once with the
            // canonical reason header.
            use tokio_stream::StreamExt;
            let msg = tokio::time::timeout(Duration::from_millis(200), dlq_sub.next())
                .await
                .expect("dlq delivery within 200ms")
                .expect("dlq stream open")
                .expect("no error");
            prop_assert_eq!(msg.payload[0], payload);
            prop_assert_eq!(
                msg.headers.get("dlq_reason").map(String::as_str),
                Some("max_attempts_exceeded")
            );
            Ok(())
        })?;
    }

    #[test]
    fn lease_expiry_redelivers_job(
        payload in any::<u8>(),
        ttl_ms in 5u64..30,
    ) {
        let rt = Runtime::new().unwrap();
        rt.block_on(async move {
            let bus = MemoryBus::new();
            bus.jobs
                .enqueue("q", Job::new(Bytes::from(vec![payload])))
                .await
                .unwrap();

            // First worker claims with a short lease, then drops the
            // claim without ack/nak. Lazy lease expiry means the next
            // claim() call past the deadline must redeliver.
            let claim1 = bus
                .jobs
                .claim("q", "w1", Duration::from_millis(ttl_ms))
                .await
                .unwrap()
                .expect("first claim must succeed");
            drop(claim1);

            // Wait past the TTL.
            tokio::time::sleep(Duration::from_millis(ttl_ms * 3 + 5)).await;

            let claim2 = bus
                .jobs
                .claim("q", "w2", Duration::from_secs(1))
                .await
                .unwrap()
                .expect("second claim must succeed via lease expiry");
            prop_assert_eq!(claim2.payload[0], payload);
            claim2.ack().await.unwrap();
            Ok(())
        })?;
    }
}
