//! Behavioural pin for peat-mesh#202 — `AutomergeStore::delete` (and
//! the receive-side path that routes through it via
//! `AutomergeSyncCoordinator::apply_tombstone`) must fire the
//! `observer_tx` / `gossip_tx` broadcast channels so a CDC consumer
//! subscribed to `subscribe_to_observer_changes` sees the delete
//! event.
//!
//! Pre-rc.29 the documented "fires for ALL document changes" contract
//! on `subscribe_to_observer_changes` held for inserts/updates but
//! not for deletes — a remote tombstone propagated to this node would
//! silently remove the local document and the observer channel never
//! woke. peat-cli's `peat observe`-with-`peat delete` cross-CLI flow
//! depends on the channel firing on deletes, so this test pins the
//! corrected contract end-to-end through the public surface
//! consumers actually call.
//!
//! Pairs with the in-tree dispatch-tier pin
//! (`src/storage/automerge_sync.rs::tests::...tombstone_receive_fires_observer_tx`)
//! that exercises the full sync-receive path; this file pins the
//! direct `store.delete` path that local deletes take.

#![cfg(feature = "automerge-backend")]

use automerge::transaction::Transactable;
use automerge::{Automerge, ROOT};
use peat_mesh::storage::{AutomergeStore, ChangeOrigin};
use std::time::Duration;
use tokio::time::timeout;

fn doc_with_field(value: &str) -> Automerge {
    let mut doc = Automerge::new();
    let mut tx = doc.transaction();
    tx.put(ROOT, "field", value).unwrap();
    tx.commit();
    doc
}

#[tokio::test]
async fn delete_fires_observer_tx() {
    let store = AutomergeStore::in_memory();
    store
        .put("contacts:c-1", &doc_with_field("alice"))
        .expect("put seed");

    // Subscribe AFTER the seed put so the seed doesn't pollute the
    // assertion stream.
    let mut observer_rx = store.subscribe_to_observer_changes();

    store.delete("contacts:c-1").expect("delete");

    let key = timeout(Duration::from_millis(500), observer_rx.recv())
        .await
        .expect("observer_tx must fire within 500ms of delete (peat-mesh#202)")
        .expect("observer_tx must yield Ok");
    assert_eq!(key, "contacts:c-1");

    // And a subsequent get returns None — the CDC consumer's
    // "Some(_) → insert/update, None → delete" pattern observable
    // through the public surface.
    let after = store.get("contacts:c-1").expect("get post-delete");
    assert!(after.is_none(), "document must be removed");
}

#[tokio::test]
async fn delete_fires_local_change_observer_and_gossip_channels() {
    let store = AutomergeStore::in_memory();
    store
        .put("contacts:c-2", &doc_with_field("bob"))
        .expect("put seed");

    let mut change_rx = store.subscribe_to_changes();
    let mut observer_rx = store.subscribe_to_observer_changes();
    let mut gossip_rx = store.subscribe_to_changes_with_origin();

    store.delete("contacts:c-2").expect("delete");

    // change_tx (local-only sync trigger) — fires because the default
    // `delete` uses Local origin.
    let change_key = timeout(Duration::from_millis(500), change_rx.recv())
        .await
        .expect("change_tx must fire within 500ms")
        .expect("change_tx must yield Ok");
    assert_eq!(change_key, "contacts:c-2");

    // observer_tx — always fires for any mutation.
    let observer_key = timeout(Duration::from_millis(500), observer_rx.recv())
        .await
        .expect("observer_tx must fire within 500ms")
        .expect("observer_tx must yield Ok");
    assert_eq!(observer_key, "contacts:c-2");

    // gossip_tx — origin-tagged Local for the default delete call.
    let gossip = timeout(Duration::from_millis(500), gossip_rx.recv())
        .await
        .expect("gossip_tx must fire within 500ms")
        .expect("gossip_tx must yield Ok");
    assert_eq!(gossip.key, "contacts:c-2");
    assert_eq!(gossip.origin, ChangeOrigin::Local);
}

#[tokio::test]
async fn delete_with_remote_origin_suppresses_change_tx_and_tags_gossip() {
    let store = AutomergeStore::in_memory();
    store
        .put("contacts:c-3", &doc_with_field("carol"))
        .expect("put seed");

    let mut change_rx = store.subscribe_to_changes();
    let mut observer_rx = store.subscribe_to_observer_changes();
    let mut gossip_rx = store.subscribe_to_changes_with_origin();

    store
        .delete_with_origin("contacts:c-3", ChangeOrigin::Remote("peer-xyz".into()))
        .expect("delete_with_origin Remote");

    // observer_tx still fires — this is the rc.29 contract repair
    // (the issue under fix). A CDC consumer subscribed to the
    // documented channel MUST see the delete even when it originated
    // from a remote peer.
    let observer_key = timeout(Duration::from_millis(500), observer_rx.recv())
        .await
        .expect("observer_tx must fire on remote-origin delete (peat-mesh#202)")
        .expect("observer_tx must yield Ok");
    assert_eq!(observer_key, "contacts:c-3");

    // gossip_tx fires with peer attribution so transitive-gossip
    // drivers can suppress the echo back to peer-xyz.
    let gossip = timeout(Duration::from_millis(500), gossip_rx.recv())
        .await
        .expect("gossip_tx must fire within 500ms")
        .expect("gossip_tx must yield Ok");
    assert_eq!(gossip.key, "contacts:c-3");
    assert_eq!(gossip.origin, ChangeOrigin::Remote("peer-xyz".into()));

    // peat-mesh#115 ping-pong contract: Remote origin must suppress
    // change_tx (no sync-back to the source peer).
    let change_recv = timeout(Duration::from_millis(150), change_rx.recv()).await;
    assert!(
        change_recv.is_err(),
        "change_tx must NOT fire for Remote-origin delete — got: {change_recv:?}"
    );
}

#[tokio::test]
async fn delete_of_missing_key_still_fires_observer() {
    // Delete is documented as a no-op error-wise when the key is
    // absent (the underlying redb `remove` on a missing key returns
    // Ok). The observer-fires contract should still hold: a CDC
    // consumer that subscribed expecting deletion events shouldn't
    // distinguish "deleted a present doc" from "deleted an absent
    // doc" — both result in the post-state "key is absent."
    //
    // This is a defensive pin: even if some future caller calls
    // delete speculatively (e.g. dedup on apply_tombstone), the
    // event still reaches downstream consumers consistently.
    let store = AutomergeStore::in_memory();
    let mut observer_rx = store.subscribe_to_observer_changes();

    store.delete("never-existed").expect("delete absent key");

    let key = timeout(Duration::from_millis(500), observer_rx.recv())
        .await
        .expect("observer_tx must fire even when the key was absent")
        .expect("observer_tx must yield Ok");
    assert_eq!(key, "never-existed");
}
