truffle-core 0.3.16

use std::sync::Arc;
use std::time::Duration;

use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;

use crate::envelope::codec::JsonCodec;
use crate::envelope::EnvelopeCodec;
use crate::network::*;
use crate::session::PeerRegistry;
use crate::synced_store::{Slice, StoreEvent, SyncedStore};
use crate::transport::websocket::WebSocketTransport;
use crate::transport::WsConfig;

// ── Test data type ──────────────────────────────────────────────────

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct TestState {
    value: i32,
    label: String,
}

// ── Mock network provider ───────────────────────────────────────────

struct MockNetworkProvider {
    identity: NodeIdentity,
    local_addr: PeerAddr,
    peer_event_tx: broadcast::Sender<NetworkPeerEvent>,
    mock_peers: Arc<tokio::sync::RwLock<Vec<NetworkPeer>>>,
}

impl MockNetworkProvider {
    fn new(id: &str) -> Self {
        let (peer_event_tx, _) = broadcast::channel(64);
        Self {
            identity: NodeIdentity {
                id: id.to_string(),
                hostname: format!("truffle-test-{id}"),
                name: format!("Test Node {id}"),
                dns_name: None,
                ip: Some("127.0.0.1".parse().unwrap()),
            },
            local_addr: PeerAddr {
                ip: Some("127.0.0.1".parse().unwrap()),
                hostname: format!("truffle-test-{id}"),
                dns_name: None,
            },
            peer_event_tx,
            mock_peers: Arc::new(tokio::sync::RwLock::new(Vec::new())),
        }
    }

    fn event_sender(&self) -> broadcast::Sender<NetworkPeerEvent> {
        self.peer_event_tx.clone()
    }
}

impl NetworkProvider for MockNetworkProvider {
    fn local_identity(&self) -> NodeIdentity {
        self.identity.clone()
    }
    fn local_addr(&self) -> PeerAddr {
        self.local_addr.clone()
    }
    fn peer_events(&self) -> broadcast::Receiver<NetworkPeerEvent> {
        self.peer_event_tx.subscribe()
    }
    async fn start(&mut self) -> Result<(), NetworkError> {
        Ok(())
    }
    async fn stop(&mut self) -> Result<(), NetworkError> {
        Ok(())
    }
    async fn peers(&self) -> Vec<NetworkPeer> {
        self.mock_peers.read().await.clone()
    }
    async fn dial_tcp(
        &self,
        _addr: &str,
        _port: u16,
    ) -> Result<tokio::net::TcpStream, NetworkError> {
        Err(NetworkError::DialFailed("mock".into()))
    }
    async fn listen_tcp(&self, _port: u16) -> Result<NetworkTcpListener, NetworkError> {
        Err(NetworkError::ListenFailed("mock".into()))
    }
    async fn unlisten_tcp(&self, _port: u16) -> Result<(), NetworkError> {
        Ok(())
    }
    async fn bind_udp(&self, _port: u16) -> Result<NetworkUdpSocket, NetworkError> {
        Err(NetworkError::NotRunning)
    }
    async fn ping(&self, _addr: &str) -> Result<PingResult, NetworkError> {
        Ok(PingResult {
            latency: Duration::from_millis(1),
            connection: "direct".to_string(),
            peer_addr: None,
        })
    }
    async fn health(&self) -> HealthInfo {
        HealthInfo {
            state: "running".to_string(),
            healthy: true,
            ..Default::default()
        }
    }
}

// ── Helpers ─────────────────────────────────────────────────────────

fn ws_config(port: u16) -> WsConfig {
    WsConfig {
        port,
        ping_interval: Duration::from_secs(300),
        pong_timeout: Duration::from_secs(300),
        ..Default::default()
    }
}

async fn random_port() -> u16 {
    let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    l.local_addr().unwrap().port()
}

async fn make_test_node(
    id: &str,
    ws_port: u16,
) -> (
    Arc<crate::node::Node<MockNetworkProvider>>,
    broadcast::Sender<NetworkPeerEvent>,
) {
    let provider = MockNetworkProvider::new(id);
    let event_tx = provider.event_sender();
    let network = Arc::new(provider);
    let ws_transport = Arc::new(WebSocketTransport::new(network.clone(), ws_config(ws_port)));
    let session = Arc::new(PeerRegistry::new(network.clone(), ws_transport));
    session.start().await;

    let codec: Arc<dyn EnvelopeCodec> = Arc::new(JsonCodec);
    let node = Arc::new(crate::node::Node::from_parts(network, session, codec));
    (node, event_tx)
}

// ── Tests ───────────────────────────────────────────────────────────

#[tokio::test]
async fn test_set_and_local() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    // Initially empty.
    assert!(store.local().await.is_none());

    // Set data.
    let state = TestState {
        value: 42,
        label: "hello".to_string(),
    };
    store.set(state.clone()).await;

    // Now readable.
    let result = store.local().await.unwrap();
    assert_eq!(result, state);

    store.stop().await;
}

#[tokio::test]
async fn test_version_increments() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    assert_eq!(store.version(), 0);

    store
        .set(TestState {
            value: 1,
            label: "v1".into(),
        })
        .await;
    assert_eq!(store.version(), 1);

    store
        .set(TestState {
            value: 2,
            label: "v2".into(),
        })
        .await;
    assert_eq!(store.version(), 2);

    store
        .set(TestState {
            value: 3,
            label: "v3".into(),
        })
        .await;
    assert_eq!(store.version(), 3);

    store.stop().await;
}

#[tokio::test]
async fn test_subscribe_local_changed() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");
    let mut events = store.subscribe();

    let state = TestState {
        value: 7,
        label: "event".into(),
    };
    store.set(state.clone()).await;

    let event = tokio::time::timeout(Duration::from_millis(100), events.recv())
        .await
        .expect("should receive event within timeout")
        .expect("channel should not be closed");

    match event {
        StoreEvent::LocalChanged(data) => assert_eq!(data, state),
        other => panic!("expected LocalChanged, got {other:?}"),
    }

    store.stop().await;
}

#[tokio::test]
async fn test_device_ids_and_store_id() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("my-sessions");

    assert_eq!(store.store_id(), "my-sessions");
    assert_eq!(store.device_id(), "device-a");

    // No device_ids before set.
    assert!(store.device_ids().await.is_empty());

    store
        .set(TestState {
            value: 1,
            label: "test".into(),
        })
        .await;

    let ids = store.device_ids().await;
    assert_eq!(ids, vec!["device-a".to_string()]);

    store.stop().await;
}

#[tokio::test]
async fn test_all_includes_local() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    store
        .set(TestState {
            value: 99,
            label: "local".into(),
        })
        .await;

    let all = store.all().await;
    assert_eq!(all.len(), 1);
    assert_eq!(all["device-a"].data.value, 99);
    assert_eq!(all["device-a"].version, 1);

    store.stop().await;
}

#[tokio::test]
async fn test_remote_slice_via_sync_message() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    // Simulate receiving an Update from a remote peer by injecting a
    // sync message into the namespace channel via send_typed to ourselves.
    // Since mock networking doesn't support real WS, we directly manipulate
    // the store's internal state to test the logic.
    {
        let remote_slice = Slice {
            device_id: "device-b".to_string(),
            data: TestState {
                value: 100,
                label: "from-b".into(),
            },
            version: 1,
            updated_at: 12345,
        };
        let mut remotes = store.inner.remotes.write().await;
        remotes.insert("device-b".to_string(), remote_slice);
    }

    // Verify remote data is readable.
    let slice = store.get("device-b").await.unwrap();
    assert_eq!(slice.data.value, 100);
    assert_eq!(slice.version, 1);

    // Verify all() includes both local and remote.
    store
        .set(TestState {
            value: 1,
            label: "local".into(),
        })
        .await;

    let all = store.all().await;
    assert_eq!(all.len(), 2);
    assert!(all.contains_key("device-a"));
    assert!(all.contains_key("device-b"));

    store.stop().await;
}

#[tokio::test]
async fn test_stale_remote_update_rejected() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    // Insert a remote slice at version 5.
    {
        let remote_slice = Slice {
            device_id: "device-b".to_string(),
            data: TestState {
                value: 50,
                label: "v5".into(),
            },
            version: 5,
            updated_at: 12345,
        };
        let mut remotes = store.inner.remotes.write().await;
        remotes.insert("device-b".to_string(), remote_slice);
    }

    // Try to apply a stale version 3 — should be ignored.
    // We call the internal apply logic directly.
    {
        let stale_slice = Slice {
            device_id: "device-b".to_string(),
            data: TestState {
                value: 30,
                label: "v3-stale".into(),
            },
            version: 3,
            updated_at: 10000,
        };

        // Manually check: the existing version should win.
        let remotes = store.inner.remotes.read().await;
        let existing = remotes.get("device-b").unwrap();
        assert!(stale_slice.version <= existing.version);
    }

    // Verify the original data is still there.
    let slice = store.get("device-b").await.unwrap();
    assert_eq!(slice.data.value, 50);
    assert_eq!(slice.version, 5);

    store.stop().await;
}

#[tokio::test]
async fn test_peer_leave_removes_slice() {
    let ws_port = random_port().await;
    let (node, event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");
    let mut events = store.subscribe();

    // Insert a remote slice.
    {
        let remote_slice = Slice {
            device_id: "device-b".to_string(),
            data: TestState {
                value: 100,
                label: "from-b".into(),
            },
            version: 1,
            updated_at: 12345,
        };
        let mut remotes = store.inner.remotes.write().await;
        remotes.insert("device-b".to_string(), remote_slice);
    }

    assert!(store.get("device-b").await.is_some());

    // Simulate peer leaving.
    let peer_b = NetworkPeer {
        id: "device-b".to_string(),
        hostname: "truffle-test-device-b".to_string(),
        ip: "127.0.0.1".parse().unwrap(),
        online: true,
        cur_addr: Some("127.0.0.1:41641".to_string()),
        relay: None,
        os: None,
        last_seen: None,
        key_expiry: None,
        dns_name: None,
    };
    // First join, then leave (the sync task needs Joined to register the peer).
    let _ = event_tx.send(NetworkPeerEvent::Joined(peer_b));
    tokio::time::sleep(Duration::from_millis(50)).await;
    let _ = event_tx.send(NetworkPeerEvent::Left("device-b".to_string()));
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Remote slice should be removed.
    assert!(store.get("device-b").await.is_none());

    // Should receive PeerRemoved event.
    let event = tokio::time::timeout(Duration::from_millis(200), async {
        loop {
            if let Ok(StoreEvent::PeerRemoved { device_id }) = events.recv().await {
                return device_id;
            }
        }
    })
    .await
    .expect("should receive PeerRemoved within timeout");

    assert_eq!(event, "device-b");

    store.stop().await;
}

#[tokio::test]
async fn test_stop_cancels_task() {
    let ws_port = random_port().await;
    let (node, _event_tx) = make_test_node("device-a", ws_port).await;

    let store: Arc<SyncedStore<TestState>> = node.synced_store("test-store");

    // Set some data to prove the store is working.
    store
        .set(TestState {
            value: 1,
            label: "test".into(),
        })
        .await;
    assert_eq!(store.version(), 1);

    // Stop should not panic.
    store.stop().await;

    // Calling stop again should be a no-op.
    store.stop().await;
}