//! Behavioural pin for the delta API surfaced on `AutomergeStore`
//! (peat-mesh#187 — closes the `peat-cli` round-trip-edit blocker for
//! peat-node ADR-001 Phase 4b `peat update --from <PATH>`).
//!
//! What this test pins:
//! - `AutomergeStore::diff(current, proposed)` returns a non-empty
//!   `AutomergeDelta` exactly when `proposed` has causal history beyond
//!   `current`'s heads.
//! - `AutomergeStore::apply_delta(key, &delta)` mutates the stored
//!   document in place, preserving operation history (ADR-021's
//!   create-once-evolve-through-deltas invariant): the resulting
//!   document has *one* create at the head of its history, not two,
//!   even after a full diff/apply cycle.
//! - Applying a delta fires the same observer + sync-coordinator +
//!   origin-tagged gossip notifications as `put` — the issue's explicit
//!   "decide whether apply_delta participates in the same pipeline as
//!   put: yes" requirement.
//! - `AutomergeDelta::to_bytes` / `from_bytes` round-trip preserves the
//!   delta byte-for-byte, so a serialised delta can cross process
//!   boundaries (the round-trip-edit pattern when CLI and store live
//!   in different processes).
//!
//! Pairs with the public-surface pin in `src/storage/automerge_store.rs`
//! unit tests; this file exercises the cross-module store/observer
//! wiring end-to-end.

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

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

/// Build a fresh Automerge doc containing `field = "alpha"` so we have
/// a non-trivial create-op shape to test against.
fn doc_with_alpha() -> Automerge {
    let mut doc = Automerge::new();
    let mut tx = doc.transaction();
    tx.put(ROOT, "field", "alpha").unwrap();
    tx.commit();
    doc
}

#[test]
fn diff_then_apply_roundtrips_field_change_and_preserves_history() {
    let store = AutomergeStore::in_memory();

    // Seed: put a "create" version with field=alpha.
    let initial = doc_with_alpha();
    store
        .put("delta-roundtrip-doc", &initial)
        .expect("put seed");
    let initial_change_count = initial.get_changes(&[]).len();
    assert_eq!(
        initial_change_count, 1,
        "single create-tx produces one Automerge change"
    );

    // Caller's round-trip-edit shape: get → modify → diff → apply.
    let current = store
        .get("delta-roundtrip-doc")
        .expect("get current")
        .expect("doc exists");
    let mut proposed = current.clone();
    let mut tx = proposed.transaction();
    tx.put(ROOT, "field", "beta").unwrap();
    tx.commit();

    let delta = AutomergeStore::diff(&current, &proposed);
    assert!(
        !delta.is_empty(),
        "single field edit produces a non-empty delta"
    );
    assert_eq!(
        delta.change_count(),
        1,
        "one transaction → one change in the delta"
    );

    store
        .apply_delta("delta-roundtrip-doc", &delta)
        .expect("apply_delta on existing key");

    // Stored value advanced to the proposed state.
    let after = store
        .get("delta-roundtrip-doc")
        .expect("get after")
        .expect("doc exists");
    let (value, _) = after
        .get(ROOT, "field")
        .expect("read field")
        .expect("field present");
    assert_eq!(value.to_str(), Some("beta"));

    // ADR-021 invariant: the document evolved via deltas, not via
    // re-creation. The history must show two changes (initial create
    // + the apply_delta tx), not e.g. a single new create snapshot
    // discarding the original.
    let after_change_count = after.get_changes(&[]).len();
    assert_eq!(
        after_change_count,
        initial_change_count + 1,
        "apply_delta extended history by exactly the delta's change count, \
         preserving the original create at the head"
    );
}

#[test]
fn diff_against_unchanged_doc_is_empty() {
    let a = doc_with_alpha();
    let b = a.clone();
    let delta = AutomergeStore::diff(&a, &b);
    assert!(
        delta.is_empty(),
        "diff against a byte-identical clone has nothing to apply"
    );
}

#[test]
fn apply_empty_delta_is_a_silent_noop() {
    let store = AutomergeStore::in_memory();
    store.put("noop-doc", &doc_with_alpha()).unwrap();
    let empty = AutomergeDelta::default();
    assert!(empty.is_empty());
    // Returns Ok and does not fire any side effect — the get below
    // observes the unchanged document, and no panic / error surfaces.
    store
        .apply_delta("noop-doc", &empty)
        .expect("empty delta is a no-op");
    let still_there = store.get("noop-doc").unwrap().unwrap();
    let (value, _) = still_there.get(ROOT, "field").unwrap().unwrap();
    assert_eq!(value.to_str(), Some("alpha"));
}

#[test]
fn apply_delta_on_missing_key_errors() {
    let store = AutomergeStore::in_memory();

    // Build a non-empty delta against an unrelated doc so the apply
    // path runs the missing-key check (an empty delta would short-
    // circuit before the get).
    let current = Automerge::new();
    let proposed = doc_with_alpha();
    let delta = AutomergeStore::diff(&current, &proposed);
    assert!(!delta.is_empty());

    let err = store
        .apply_delta("no-such-key", &delta)
        .expect_err("apply_delta on missing key must fail");
    let msg = format!("{err:#}");
    assert!(
        msg.contains("no-such-key") && msg.contains("not found"),
        "diagnostic must name the missing key + the not-found condition; got: {msg}"
    );
}

#[tokio::test]
async fn apply_delta_fires_local_change_observer_and_gossip_channels() {
    let store = AutomergeStore::in_memory();
    store
        .put("observed-doc", &doc_with_alpha())
        .expect("put seed");

    // Subscribe BEFORE we mutate. broadcast::Receiver only sees messages
    // sent after subscription (the seed put above is intentionally
    // pre-subscription so it doesn't pollute the assertion stream).
    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();

    let current = store.get("observed-doc").unwrap().unwrap();
    let mut proposed = current.clone();
    let mut tx = proposed.transaction();
    tx.put(ROOT, "field", "gamma").unwrap();
    tx.commit();
    let delta = AutomergeStore::diff(&current, &proposed);
    store
        .apply_delta("observed-doc", &delta)
        .expect("apply_delta");

    // change_tx (local-only sync trigger) — fires because the default
    // `apply_delta` 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, "observed-doc");

    // 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, "observed-doc");

    // gossip_tx — origin-tagged Local for the default apply_delta 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, "observed-doc");
    assert_eq!(gossip.origin, ChangeOrigin::Local);
}

#[tokio::test]
async fn apply_delta_with_remote_origin_suppresses_change_tx_and_tags_gossip() {
    let store = AutomergeStore::in_memory();
    store
        .put("remote-applied-doc", &doc_with_alpha())
        .expect("put seed");

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

    let current = store.get("remote-applied-doc").unwrap().unwrap();
    let mut proposed = current.clone();
    let mut tx = proposed.transaction();
    tx.put(ROOT, "field", "delta").unwrap();
    tx.commit();
    let delta = AutomergeStore::diff(&current, &proposed);
    store
        .apply_delta_with_origin(
            "remote-applied-doc",
            &delta,
            ChangeOrigin::Remote("peer-xyz".into()),
        )
        .expect("apply_delta_with_origin Remote");

    // Mirrors `put_with_origin`'s peat-mesh#115 contract: Remote origin
    // suppresses 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 apply_delta — got: {change_recv:?}"
    );

    // gossip_tx carries the 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, "remote-applied-doc");
    assert_eq!(gossip.origin, ChangeOrigin::Remote("peer-xyz".into()));
}

#[test]
fn delta_to_bytes_from_bytes_roundtrip() {
    // Build a multi-change delta so the length-prefix framing is
    // exercised across more than one entry.
    let current = doc_with_alpha();
    let mut proposed = current.clone();

    let mut tx1 = proposed.transaction();
    tx1.put(ROOT, "field", "beta").unwrap();
    tx1.commit();
    let mut tx2 = proposed.transaction();
    tx2.put(ROOT, "extra", 7_i64).unwrap();
    tx2.commit();

    let delta = AutomergeStore::diff(&current, &proposed);
    assert_eq!(
        delta.change_count(),
        2,
        "two distinct transactions → two changes in the delta"
    );

    let bytes = delta.to_bytes().expect("to_bytes within u32 bounds");
    let reparsed = AutomergeDelta::from_bytes(&bytes).expect("from_bytes round-trip");
    assert_eq!(reparsed.change_count(), delta.change_count());
    assert_eq!(
        reparsed.to_bytes().expect("reparsed to_bytes"),
        bytes,
        "round-tripped delta serialises to the same bytes"
    );

    // And applying the reparsed delta yields the same target state as
    // applying the original.
    let store = AutomergeStore::in_memory();
    store.put("rtdoc", &current).unwrap();
    store
        .apply_delta("rtdoc", &reparsed)
        .expect("apply reparsed delta");
    let after = store.get("rtdoc").unwrap().unwrap();
    let (field, _) = after.get(ROOT, "field").unwrap().unwrap();
    let (extra, _) = after.get(ROOT, "extra").unwrap().unwrap();
    assert_eq!(field.to_str(), Some("beta"));
    assert_eq!(extra.to_i64(), Some(7));
}

#[test]
fn delta_from_bytes_rejects_truncated_buffer() {
    let proposed = doc_with_alpha();
    let delta = AutomergeStore::diff(&Automerge::new(), &proposed);
    let bytes = delta.to_bytes().expect("to_bytes within u32 bounds");
    assert!(!bytes.is_empty());

    // Drop the last byte of the change body → must error, not silently
    // truncate.
    let truncated = &bytes[..bytes.len() - 1];
    let err = AutomergeDelta::from_bytes(truncated).expect_err("truncated must fail");
    let msg = format!("{err:#}");
    assert!(
        msg.contains("truncated"),
        "from_bytes diagnostic must name the truncation; got: {msg}"
    );
}
