//! m8p10 BUG 2 reproduction: repeated reconcile of two ALREADY-CONVERGED nodes
//! must be a fixpoint on decay scores (no creep).
//!
//! The UAT engineer observed: after WAL-relaying the same signals to both nodes,
//! repeated `/cluster/reconcile` makes decay scores creep (0.5 → 0.375 → …)
//! instead of being a no-op. The root cause is CRDT attribution: each node folds
//! the replicated event into its own local hot accumulator and then
//! `take_crdt_snapshot` attributes that whole accumulator to its OWN local shard.
//! A signal that physically exists on both nodes is therefore represented as TWO
//! disjoint per-shard contributions, which `CrdtSignalState::merge` sums.
//!
//! The load-bearing property: reconcile of two converged nodes is a no-op on
//! scores; the merge attributes a replicated event to its ORIGINATING shard on
//! every node that holds it, so it is one contribution, not N.
#![allow(clippy::unwrap_used, clippy::float_cmp)]

use std::{sync::Arc, time::Duration};

use tidaldb::{
    TidalDb,
    db::config::{NodeConfig, NodeRole},
    replication::{
        WalSegmentId,
        shard::ShardId,
        state::ReplicationState,
        transport::{Transport, TransportError, WalSegmentPayload},
    },
    schema::{DecaySpec, EntityId, EntityKind, SchemaBuilder, Timestamp, Window},
    signals::{NoopWalWriter, SignalLedger},
    wal::format::batch::{EventRecord, encode_batch},
};

struct ChannelTransport {
    rx: crossbeam::channel::Receiver<WalSegmentPayload>,
    local: ShardId,
}

impl Transport for ChannelTransport {
    fn send_segment(
        &self,
        _to: ShardId,
        _payload: WalSegmentPayload,
    ) -> Result<(), TransportError> {
        Ok(())
    }
    fn recv_segment(&self) -> Option<WalSegmentPayload> {
        self.rx.recv().ok()
    }
    fn local_shard(&self) -> ShardId {
        self.local
    }
}

fn make_schema() -> tidaldb::schema::Schema {
    let mut builder = SchemaBuilder::new();
    let _ = builder
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::AllTime])
        .velocity(false)
        .add();
    builder.build().unwrap()
}

fn resolve_view_type_id(schema: &tidaldb::schema::Schema) -> tidaldb::signals::SignalTypeId {
    let ledger = SignalLedger::new(schema.clone(), Box::new(NoopWalWriter));
    ledger.resolve_signal_type("view").unwrap()
}

fn open_node(
    schema: tidaldb::schema::Schema,
    role: NodeRole,
    shard: ShardId,
    peers: &[ShardId],
) -> TidalDb {
    TidalDb::builder()
        .ephemeral()
        .with_schema(schema)
        .with_cluster(NodeConfig {
            role,
            shard_id: shard,
            peer_shards: peers.to_vec(),
            ..NodeConfig::default()
        })
        .open()
        .expect("node opens")
}

fn wait_for_applied(state: &ReplicationState, shard: ShardId, expected: u64) {
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    loop {
        if state.applied_seqno(shard) == Some(expected) {
            return;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "replication did not reach applied seqno {expected} (last: {:?})",
            state.applied_seqno(shard)
        );
        std::thread::sleep(Duration::from_millis(1));
    }
}

/// Two nodes, both holding the SAME WAL-relayed signal (leader writes locally,
/// follower applies the replicated segment). After convergence, repeated
/// reconcile in BOTH directions must leave both nodes' decay scores unchanged.
#[test]
fn reconcile_of_converged_nodes_is_a_fixpoint() {
    let schema = make_schema();
    // Leader is shard 0, follower is shard 1 — exactly the multi-process layout.
    // Both processes open their region as NodeRole::Single (writeable, can apply
    // replicated writes + reconcile), as the multi-process region node does. The
    // follower tracks the leader's shard as a peer so the receiver can advance
    // its applied-seqno for the leader's stream.
    let leader = open_node(schema.clone(), NodeRole::Single, ShardId(0), &[ShardId(1)]);
    let follower = open_node(schema.clone(), NodeRole::Single, ShardId(1), &[ShardId(0)]);

    let type_id = resolve_view_type_id(&schema);
    let follower_state = follower.replication_state().clone();

    let (tx, rx) = crossbeam::channel::bounded(16);
    let transport = Arc::new(ChannelTransport {
        rx,
        local: ShardId(1),
    });
    follower.start_replication(Arc::clone(&transport)).unwrap();

    // Leader writes a signal locally and relays it to the follower (the WAL
    // relay path — NOT reconcile). Both nodes now physically hold entity 100.
    // Use a near-now timestamp so the 7-day half-life signal has not decayed to
    // zero by query time (read_decay_score reads at now()).
    let ts = Timestamp::now();
    let entity = EntityId::new(100);
    leader.signal("view", entity, 1.0, ts).unwrap();

    let events = vec![EventRecord::signal(
        100,
        type_id.as_u16() as u8,
        1.0,
        ts.as_nanos(),
    )];
    let bytes = encode_batch(&events, 1, 1).unwrap();
    tx.send(WalSegmentPayload {
        id: WalSegmentId::new(tidaldb::replication::RegionId::SINGLE, ShardId::SINGLE, 1),
        bytes,
        event_count: 1,
        leader_last_seq: 1,
        stream_baseline: 0,
    })
    .unwrap();
    wait_for_applied(&follower_state, ShardId::SINGLE, 1);

    // Baseline: both nodes agree on the score via the WAL relay alone (this is
    // the contract — reconcile is NOT needed for convergence).
    let leader_baseline = leader.read_decay_score(entity, "view", 0).unwrap().unwrap();
    let follower_baseline = follower
        .read_decay_score(entity, "view", 0)
        .unwrap()
        .unwrap();
    assert!(
        (leader_baseline - follower_baseline).abs() < 1e-9,
        "WAL relay alone must converge: leader={leader_baseline} follower={follower_baseline}"
    );
    assert!(
        (leader_baseline - 1.0).abs() < 1e-3,
        "baseline score should be ~1.0, got {leader_baseline}"
    );

    // Now exchange CRDT snapshots and reconcile BOTH ways, three rounds. Because
    // both nodes already hold the identical signal, each reconcile MUST be a
    // no-op on scores. Bug 2: scores creep (double-count → re-attribution).
    let mut prev_leader = leader_baseline;
    let mut prev_follower = follower_baseline;
    for round in 1..=3 {
        let leader_snap = leader.take_crdt_snapshot().unwrap();
        let follower_snap = follower.take_crdt_snapshot().unwrap();
        leader.reconcile_with(&follower_snap).unwrap();
        follower.reconcile_with(&leader_snap).unwrap();

        let leader_now = leader.read_decay_score(entity, "view", 0).unwrap().unwrap();
        let follower_now = follower
            .read_decay_score(entity, "view", 0)
            .unwrap()
            .unwrap();

        assert!(
            (leader_now - prev_leader).abs() < 1e-9,
            "round {round}: leader score must not change across reconcile of converged nodes: \
             {prev_leader} → {leader_now} (creep = bug 2)"
        );
        assert!(
            (follower_now - prev_follower).abs() < 1e-9,
            "round {round}: follower score must not change across reconcile of converged nodes: \
             {prev_follower} → {follower_now} (creep = bug 2)"
        );
        prev_leader = leader_now;
        prev_follower = follower_now;
    }

    drop(tx);
    leader.close().unwrap();
    follower.close().unwrap();
}

/// Diverged convergence still works: the leader accepts a signal the follower
/// MISSES during a partition (no WAL relay), then a single reconcile in the
/// follower's direction folds in the leader's state without double-counting, and
/// a SECOND reconcile is an exact no-op (fixpoint). This proves the canonical-
/// shard attribution does not regress the legitimate anti-entropy path.
#[test]
fn diverged_reconcile_converges_then_is_a_fixpoint() {
    let schema = make_schema();
    let leader = open_node(schema.clone(), NodeRole::Single, ShardId(0), &[ShardId(1)]);
    let follower = open_node(schema, NodeRole::Single, ShardId(1), &[ShardId(0)]);

    let entity = EntityId::new(200);
    let ts = Timestamp::now();
    // Leader writes; the follower is "partitioned" so it never receives the relay.
    leader.signal("view", entity, 3.0, ts).unwrap();

    // Follower has nothing for this entity yet.
    assert!(
        follower
            .read_decay_score(entity, "view", 0)
            .unwrap()
            .is_none(),
        "follower must start without the partitioned signal"
    );

    let leader_score = leader.read_decay_score(entity, "view", 0).unwrap().unwrap();

    // First reconcile: fold the leader's snapshot into the follower.
    let leader_snap = leader.take_crdt_snapshot().unwrap();
    follower.reconcile_with(&leader_snap).unwrap();
    let follower_after = follower
        .read_decay_score(entity, "view", 0)
        .unwrap()
        .unwrap();
    assert!(
        (follower_after - leader_score).abs() < 1e-6,
        "follower must converge to the leader's score {leader_score}, got {follower_after}"
    );

    // Second reconcile with the SAME snapshot: exact no-op (no double-count).
    let leader_snap2 = leader.take_crdt_snapshot().unwrap();
    follower.reconcile_with(&leader_snap2).unwrap();
    let follower_again = follower
        .read_decay_score(entity, "view", 0)
        .unwrap()
        .unwrap();
    assert!(
        (follower_again - follower_after).abs() < 1e-9,
        "second reconcile must be a fixpoint: {follower_after} → {follower_again}"
    );

    leader.close().unwrap();
    follower.close().unwrap();
}