keywatch 0.1.1

use std::collections::HashSet;
use std::time::{Duration, Instant};

use keywatch::channel_with_starting_keys;
use keywatch::{TryRecvError, Update, channel};

#[tokio::test]
async fn add_and_recv_basic() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    tx.send("a", Update::Add(1)).unwrap();
    assert_eq!(rx.try_recv().unwrap(), ("a", Update::Add(1)));
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn delete_updates_state() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    tx.send("k", Update::Add(10)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(10))));
    tx.send("k", Update::Delete).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Delete)));
}

#[tokio::test]
async fn clear_generates_deletes_for_existing_keys() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    tx.send("a", Update::Add(1)).unwrap();
    tx.send("b", Update::Add(2)).unwrap();
    // Drain adds
    let mut seen = Vec::new();
    seen.push(rx.recv().await.unwrap());
    seen.push(rx.recv().await.unwrap());
    seen.sort_by_key(|(k, _)| *k);
    assert_eq!(seen, vec![("a", Update::Add(1)), ("b", Update::Add(2))]);
    tx.clear().unwrap();
    // Expect two deletes (order doesn't matter)
    let mut deletes = vec![rx.recv().await.unwrap(), rx.recv().await.unwrap()];
    deletes.sort_by_key(|(k, _)| *k);
    assert_eq!(deletes, vec![("a", Update::Delete), ("b", Update::Delete)]);
}

#[tokio::test]
async fn cooldown_coalesces_updates_until_sent() {
    // cooldown > 0 means after Add is consumed, further Add before cooldown expiry is queued and
    // delivered after cooldown unless a Delete occurs
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::from_millis(50));
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    // Immediately send new value; should NOT be received until cooldown expires
    tx.send("k", Update::Add(2)).unwrap();
    // try_recv should be empty right now
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    // After sleeping past cooldown we should see the coalesced Add(2)
    tokio::time::sleep(Duration::from_millis(60)).await;
    assert_eq!(rx.recv().await, Some(("k", Update::Add(2))));
}

#[tokio::test]
async fn delete_during_cooldown_supersedes_pending_add() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::from_millis(50));
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    tx.send("k", Update::Add(2)).unwrap();
    // Delete before cooldown ends
    tx.send("k", Update::Delete).unwrap();
    // Sleep long enough that if Add were pending it would appear
    tokio::time::sleep(Duration::from_millis(60)).await;
    // We expect Delete, not Add(2)
    assert_eq!(rx.recv().await, Some(("k", Update::Delete)));
    // No further events
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn disconnect_behavior() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    drop(tx);
    // After dropping sender, try_recv should return Disconnected because dropped flag is set in
    // Sender::drop
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
}

#[tokio::test]
async fn add_then_delete_before_receive_results_in_no_event() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    tx.send("k", Update::Add(1)).unwrap();
    tx.send("k", Update::Delete).unwrap();
    // The Add should have been removed by the Delete before receiver saw it; no Delete event is
    // queued either.
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    // Ensure nothing arrives later within a short window.
    tokio::select! {
        v = rx.recv() => panic!("unexpected event: {:?}", v),
        _ = tokio::time::sleep(Duration::from_millis(20)) => {}
    }
}

#[tokio::test]
async fn delete_then_add_before_receive_results_in_add_only() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    // Establish existing key so Delete has something meaningful to remove.
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    // Now queue a Delete followed immediately by an Add before receiver fetches either.
    tx.send("k", Update::Delete).unwrap();
    tx.send("k", Update::Add(42)).unwrap();
    // Expect only the Add(42); the intermediate Delete is superseded because it was never observed.
    assert_eq!(rx.try_recv().unwrap(), ("k", Update::Add(42)));
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    tokio::select! {
        v = rx.recv() => panic!("unexpected extra event: {:?}", v),
        _ = tokio::time::sleep(Duration::from_millis(20)) => {}
    }
}

#[tokio::test]
async fn delete_twice_before_receive_results_in_single_delete() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    // Establish key so it is tracked by receiver.
    tx.send("k", Update::Add(7)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(7))));
    // Queue two deletes back-to-back before receiver pulls.
    tx.send("k", Update::Delete).unwrap();
    tx.send("k", Update::Delete).unwrap();
    // Only one Delete event should be observed.
    assert_eq!(rx.try_recv().unwrap(), ("k", Update::Delete));
    // Nothing else.
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    tokio::select! {
        v = rx.recv() => panic!("unexpected extra event after double delete: {:?}", v),
        _ = tokio::time::sleep(Duration::from_millis(20)) => {}
    }
}

#[tokio::test]
async fn add_second_then_delete_before_second_seen_results_in_delete() {
    let (tx, mut rx) = channel::<&'static str, i32>(Duration::ZERO);
    // First add
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    // Second add followed immediately by delete before receiver pulls again
    tx.send("k", Update::Add(2)).unwrap();
    tx.send("k", Update::Delete).unwrap();
    // Receiver should observe a Delete (not the intermediate Add(2))
    assert_eq!(rx.recv().await, Some(("k", Update::Delete)));
    // And nothing further
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn clear_discards_cooldown_value_and_emits_delete() {
    let cooldown = Duration::from_millis(100);
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);
    // First add and receive
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    // Second add goes into cooldown slot (not yet visible)
    tx.send("k", Update::Add(2)).unwrap();
    // Clear before cooldown expires
    // ensure some time passes but less than cooldown
    tokio::time::sleep(Duration::from_millis(10)).await;
    tx.clear().unwrap();
    // We should receive a Delete (not Add(2))
    assert_eq!(rx.recv().await, Some(("k", Update::Delete)));
    // After cooldown would have expired, no Add should surface
    tokio::time::sleep(cooldown).await;
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn clear_mixed_after_second_send_before_other_matures() {
    // Scenario: A and B added and received. Second send for A happens while both still cooling.
    // Then we wait until only A has matured but B has not, and call clear. We expect only Deletes.
    let cooldown = Duration::from_millis(100);
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);

    // Initial adds
    tx.send("A", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("A", Update::Add(1))));
    tokio::time::sleep(Duration::from_millis(20)).await; // stagger start times
    tx.send("B", Update::Add(10)).unwrap();
    assert_eq!(rx.recv().await, Some(("B", Update::Add(10))));

    // While both cooling, send an updated value for A (replaces cooldown slot value, keeps original expiry)
    tokio::time::sleep(Duration::from_millis(50)).await; // t ≈ 70ms since A inserted, 50ms since B
    tx.send("A", Update::Add(2)).unwrap();

    // Wait until A matured (~100ms) but before B (~120ms)
    tokio::time::sleep(Duration::from_millis(35)).await; // total ≈ 105ms; A matured, B not

    tx.clear().unwrap();

    // Expect Deletes for both A and B (order agnostic)
    let mut deletes = vec![rx.recv().await.unwrap(), rx.recv().await.unwrap()];
    deletes.sort_by_key(|(k, _)| *k);
    assert_eq!(deletes, vec![("A", Update::Delete), ("B", Update::Delete)]);

    // No further events
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn recv_waits_for_cooldown_expiry() {
    // Ensure that recv() actually awaits the cooldown future (not a notification) for a matured
    // cooldown key.
    let cooldown = Duration::from_millis(120);
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);

    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));

    // Second add enters cooldown
    tx.send("k", Update::Add(2)).unwrap();

    // Immediately call recv; it should block until cooldown matures.
    let start = Instant::now();
    let received = rx.recv().await;
    let elapsed = start.elapsed();

    assert_eq!(received, Some(("k", Update::Add(2))));
    // Allow a small tolerance for timer jitter.
    assert!(
        elapsed >= cooldown - Duration::from_millis(25),
        "recv returned too early: {:?} < {:?}",
        elapsed,
        cooldown
    );
}

#[tokio::test]
async fn recv_wakes_on_notification() {
    // Long cooldown; we expect recv to complete well before it if woken by a send notification.
    let cooldown = Duration::from_secs(2);
    let (tx, rx) = channel::<&'static str, i32>(cooldown);
    let start = Instant::now();
    let handle = tokio::spawn(async move {
        let mut rx = rx;
        rx.recv().await
    });
    // Give the task a brief moment to park in recv loop.
    tokio::time::sleep(Duration::from_millis(30)).await;
    tx.send("k", Update::Add(1)).unwrap();
    let out = handle.await.unwrap();
    let elapsed = start.elapsed();
    assert_eq!(out, Some(("k", Update::Add(1))));
    assert!(
        elapsed < cooldown / 4,
        "recv waited too long: {:?}",
        elapsed
    );
}

#[tokio::test]
async fn recv_wakes_on_clear() {
    // Scenario: receiver is idle (no pending events), a key exists (so clear will generate a Delete),
    // and clear() is called while recv() is awaiting. recv() should wake promptly to deliver Delete.
    use std::time::Instant;
    let cooldown = Duration::from_secs(5); // long so we rely on notification, not cooldown expiry
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1)))); // drain initial add

    let start = Instant::now();
    let recv_fut = tokio::spawn(async move { rx.recv().await });
    // Give task time to park
    tokio::time::sleep(Duration::from_millis(40)).await;
    tx.clear().unwrap();
    let out = recv_fut.await.unwrap();
    let elapsed = start.elapsed();
    assert_eq!(out, Some(("k", Update::Delete)));
    // Should wake well before cooldown window.
    assert!(elapsed < cooldown / 10, "recv waited too long for clear notification: {:?}", elapsed);
}

#[tokio::test]
async fn send_fails_after_receiver_drop() {
    let (tx, rx) = channel::<&'static str, i32>(Duration::ZERO);
    drop(rx); // trigger dropped flag
    let err = tx
        .send("k", Update::Add(42))
        .err()
        .expect("expected send error");
    match err.0 {
        Update::Add(v) => assert_eq!(v, 42),
        other => panic!("unexpected update in error: {:?}", other),
    }
}

#[tokio::test]
async fn clear_fails_after_receiver_drop() {
    let (tx, rx) = channel::<&'static str, i32>(Duration::ZERO);
    drop(rx);
    let err = tx.clear().err().expect("expected clear error");
    match err.0 {
        Update::Delete => {}
        other => panic!("unexpected update in clear error: {:?}", other),
    }
}

#[tokio::test]
async fn starting_key_delete_emitted_unknown_key_delete_suppressed() {
    // starting key "known" should produce Delete; unknown should not.
    let (tx, mut rx) = channel_with_starting_keys::<&'static str, i32>(
        HashSet::from_iter(["known"]),
        Duration::ZERO,
    );
    // Delete for starting key should appear.
    tx.send("known", Update::Delete).unwrap();
    assert_eq!(rx.try_recv().unwrap(), ("known", Update::Delete));
    // Delete for unknown key should be suppressed.
    tx.send("unknown", Update::Delete).unwrap();
    assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
}

#[tokio::test]
async fn try_recv_now_bypasses_cooldown() {
    let cooldown = Duration::from_millis(150);
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);
    tx.send("k", Update::Add(1)).unwrap();
    assert_eq!(rx.recv().await, Some(("k", Update::Add(1))));
    // Send replacement during cooldown (value stored in cooldown slot, not yet observable).
    tx.send("k", Update::Add(2)).unwrap();
    // Normal try_recv cannot see it yet due to cooldown.
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
    // Bypass cooldown: should release the cooled value early.
    assert_eq!(rx.try_recv_now().unwrap(), ("k", Update::Add(2)));
    // Subsequent tries are empty until another send.
    assert!(matches!(rx.try_recv_now(), Err(TryRecvError::Empty)));
    // Wait past original cooldown to ensure it doesn't reappear.
    tokio::time::sleep(Duration::from_millis(170)).await;
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn try_recv_now_returns_pending_non_cooled_add() {
    // Scenario: An Add not yet received (so not in cooldown cycle) should be retrievable via try_recv_now.
    let cooldown = Duration::from_millis(200);
    let (tx, mut rx) = channel::<&'static str, i32>(cooldown);
    tx.send("k", Update::Add(10)).unwrap();
    // Use try_recv_now instead of recv/try_recv; should return the Add immediately.
    assert_eq!(rx.try_recv_now().unwrap(), ("k", Update::Add(10)));
    // Nothing else is pending; no replacement value has been sent yet.
    assert!(matches!(rx.try_recv_now(), Err(TryRecvError::Empty)));
    // After cooldown elapses without a replacement Add, nothing new appears.
    tokio::time::sleep(Duration::from_millis(220)).await;
    assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
}