//! RFC 010 c10 — pid targeting + auto-serialization. The Phase 3 gate:
//! cross-node call/reply with no ceremony, under the subprocess harness.
//!
//! Local suite (`run()`, no network): serialize/deserialize shapes,
//! self-collapse, the outside-runtime contract, the local send-site
//! incarnation check with a probe proving **no frame is emitted**.
//!
//! Cross-process: two nodes. The *server* exposes a `Name<Req>`; the
//! *client* sends a `Req` carrying its own `Pid<Reply>` (auto-serialized to
//! a `RemotePid` on the wire); the server replies via `send_to_remote`
//! straight back to that pid — no name at the client end, no ceremony. A
//! third-node roundtrip: the client's pid travels client→server→relay→
//! server→client, and still delivers.
#![cfg(feature = "cluster")]

mod common;

use common::{maybe_child, spawn_node};
use smarm::cluster::envelope::{encode_payload, Frame, NodeMeta};
use smarm::cluster::expose::{expose, type_hash};
use smarm::cluster::membership::{subscribe, NodeEvent};
use smarm::cluster::remote::{self, send_to_remote, RemoteName, RemotePid, ToRemoteError};
use smarm::cluster::{start, Config, StaticSeeds, Timing};
use smarm::pg::Incarnation;
use smarm::{channel, install, register, run, Addressable, Name, Pid};
use std::time::Duration;

// ---- message types (std-only payloads; the crate's serde is derive-less,
// so wire types are hand-rolled with serde's tuple/seq API via `serde::ser`
// impls below — the same thing a user's derive would generate) ------------

/// A request carrying a reply-to. Serialize/Deserialize are written by hand
/// here for exactly one reason: this crate deliberately does not pull in
/// serde-derive. Field 1 is the auto-serializing pid.
#[derive(Debug, PartialEq)]
struct Req {
    text: String,
    reply_to: RemotePid<Replier>,
}

#[derive(Debug, PartialEq)]
struct Reply(String);

struct Replier;
impl Addressable for Replier {
    type Msg = Reply;
}

impl serde::Serialize for Req {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeTuple;
        let mut t = s.serialize_tuple(2)?;
        t.serialize_element(&self.text)?;
        t.serialize_element(&self.reply_to)?;
        t.end()
    }
}
impl<'de> serde::Deserialize<'de> for Req {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let (text, reply_to) = <(String, RemotePid<Replier>)>::deserialize(d)?;
        Ok(Req { text, reply_to })
    }
}
impl serde::Serialize for Reply {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        self.0.serialize(s)
    }
}
impl<'de> serde::Deserialize<'de> for Reply {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        String::deserialize(d).map(Reply)
    }
}

// ================= local suite =========================================

/// A local `Pid<A>` serializes as a `RemotePid<A>` stamped with this node's
/// identity; deserializing it back on the same node collapses to the same
/// local pid (`local()` is `Some`, `Pid` round-trips).
#[test]
fn local_pid_serializes_and_collapses_on_self() {
    maybe_child(ROLES);
    run(|| {
        // The local identity is set by cluster::start; the local suite sets
        // it directly.
        remote::set_local_identity("me", Incarnation::new(7));
        let (tx, _rx) = channel::<Reply>();
        let me: Pid<Replier> = install::<Replier>(tx);

        let bytes = encode_payload(&me).unwrap();
        let rp: RemotePid<Replier> = smarm::cluster::envelope::decode_payload(&bytes).unwrap();
        assert_eq!(rp.node(), "me");
        assert_eq!(rp.incarnation(), Incarnation::new(7));
        assert_eq!(
            rp.local(),
            Some(me),
            "self-node pid collapses to the local pid"
        );

        // Deserializing straight into Pid<A> works for a self-node pid...
        let back: Pid<Replier> = smarm::cluster::envelope::decode_payload(&bytes).unwrap();
        assert_eq!(back, me);

        // ...and FAILS for a foreign one (collapse is literal: node == self).
        let foreign = RemotePid::<Replier>::from_parts("elsewhere", Incarnation::new(1), 3, 1);
        let fbytes = encode_payload(&foreign).unwrap();
        assert!(smarm::cluster::envelope::decode_payload::<Pid<Replier>>(&fbytes).is_err());
        assert_eq!(foreign.local(), None);
    });
}

/// `send_to_remote` short-circuits locally for a self-node pid — the
/// zero-copy-equivalent collapse: the message object itself lands in the
/// local channel, no encode, no frame.
#[test]
fn send_to_remote_collapses_locally_for_self() {
    maybe_child(ROLES);
    run(|| {
        remote::set_local_identity("me", Incarnation::new(7));
        let (tx, rx) = channel::<Reply>();
        let me: Pid<Replier> = install::<Replier>(tx);
        let rp = RemotePid::from_local(me).expect("identity set");
        // Probe the outbound path: nothing must be handed to any connection.
        let (probe_tx, probe_rx) = channel::<Frame>();
        remote::bind_outbound_probe("me", Incarnation::new(7), probe_tx);

        send_to_remote(rp, Reply("hi".into())).unwrap();
        assert_eq!(rx.recv().unwrap(), Reply("hi".into()));
        assert!(
            matches!(probe_rx.try_recv(), Ok(None)),
            "no frame for a local collapse"
        );
    });
}

/// RFC v2 §3: a `RemotePid` whose incarnation is not the current one for its
/// node fails at the local send site with `DeadIncarnation`, and NO frame
/// is emitted — asserted on a probe sender bound as that node's outbound.
#[test]
fn stale_incarnation_rejected_locally_no_frame() {
    maybe_child(ROLES);
    run(|| {
        remote::set_local_identity("me", Incarnation::new(7));
        let (probe_tx, probe_rx) = channel::<Frame>();
        remote::bind_outbound_probe("peer", Incarnation::new(5), probe_tx);

        let stale = RemotePid::<Replier>::from_parts("peer", Incarnation::new(4), 9, 1);
        match send_to_remote(stale, Reply("late".into())) {
            Err(ToRemoteError::DeadIncarnation(Reply(s))) => assert_eq!(s, "late"),
            other => panic!("expected DeadIncarnation, got {other:?}"),
        }
        assert!(
            matches!(probe_rx.try_recv(), Ok(None)),
            "stale pid must emit no frame"
        );

        // The current incarnation goes through: a Send frame with the pid's
        // (index, generation) and Reply's hash lands on the probe.
        let live = RemotePid::<Replier>::from_parts("peer", Incarnation::new(5), 9, 1);
        send_to_remote(live, Reply("now".into())).unwrap();
        match probe_rx.recv().unwrap() {
            Frame::Send {
                index,
                generation,
                type_hash: h,
                payload,
            } => {
                assert_eq!((index, generation), (9, 1));
                assert_eq!(h, type_hash::<Reply>());
                let r: Reply = smarm::cluster::envelope::decode_payload(&payload).unwrap();
                assert_eq!(r, Reply("now".into()));
            }
            f => panic!("expected Send, got {f:?}"),
        }

        // Unknown node: NotConnected, no frame anywhere.
        let nowhere = RemotePid::<Replier>::from_parts("nowhere", Incarnation::new(1), 1, 1);
        assert!(matches!(
            send_to_remote(nowhere, Reply("x".into())),
            Err(ToRemoteError::NotConnected(_))
        ));
    });
}

// ================= cross-process gate ==================================

const ROLES: &[(&str, fn())] = &[
    ("server", role_server),
    ("client", role_client),
    ("relay", role_relay),
];

const ECHO: Name<Req> = Name::new("c10.echo");
const RELAY: Name<Req> = Name::new("c10.relay");

fn cfg(name: &str, seeds: Vec<(String, String)>) -> Config {
    Config {
        node_name: name.to_string(),
        meta: NodeMeta {
            role: "c10".into(),
            region: "local".into(),
        },
        listen_addr: std::env::var("SMARM_LISTEN_ADDR").unwrap_or_else(|_| "127.0.0.1:0".into()),
        strategy: Box::new(StaticSeeds::new(seeds)),
        timing: Timing::default(),
    }
}

fn wait_up(events: &smarm::cluster::membership::MembershipEvents, who: &str) {
    loop {
        match events.rx.recv() {
            Ok(NodeEvent::NodeUp(i)) if i.name == who => return,
            Ok(_) => continue,
            Err(_) => panic!("manager gone"),
        }
    }
}

/// Server: exposes ECHO; each Req is answered by `send_to_remote` to its
/// reply_to — the server never learns a name for the client. If the Req text
/// starts with "via-relay:", it forwards the whole Req (reply_to and all) to
/// the relay node instead, which sends it back here; the second arrival is
/// answered normally. That is the pid's third-node roundtrip.
fn role_server() {
    let relay_addr = std::env::var("SMARM_RELAY_ADDR").ok();
    smarm::run(move || {
        let seeds = relay_addr
            .map(|a| vec![("relay".to_string(), a)])
            .unwrap_or_default();
        let cluster = start(cfg("server", seeds)).expect("binds");
        println!("LISTENING {}", cluster.local_addr());
        let (tx, rx) = channel::<Req>();
        register(ECHO, tx).unwrap();
        expose(ECHO);
        println!("READY");
        loop {
            let req = rx.recv().unwrap();
            if let Some(rest) = req.text.strip_prefix("via-relay:") {
                let fwd = Req {
                    text: format!("relayed:{rest}"),
                    reply_to: req.reply_to,
                };
                remote::send(RemoteName::new("relay", RELAY), fwd).unwrap();
                println!("FORWARDED");
                continue;
            }
            println!("REQ {}", req.text);
            send_to_remote(req.reply_to, Reply(format!("echo:{}", req.text))).unwrap();
        }
    });
}

/// Relay: exposes RELAY; bounces every Req straight back to the server's
/// ECHO, untouched. The client's pid inside it now crosses relay→server.
fn role_relay() {
    let server_addr = std::env::var("SMARM_SERVER_ADDR").expect("SMARM_SERVER_ADDR");
    smarm::run(move || {
        let cluster = start(cfg("relay", vec![("server".into(), server_addr)])).expect("binds");
        println!("LISTENING {}", cluster.local_addr());
        let (tx, rx) = channel::<Req>();
        register(RELAY, tx).unwrap();
        expose(RELAY);
        let ev = subscribe().unwrap();
        wait_up(&ev, "server");
        println!("READY");
        loop {
            let req = rx.recv().unwrap();
            println!("RELAYING {}", req.text);
            remote::send(RemoteName::new("server", ECHO), req).unwrap();
        }
    });
}

/// Client: connects to server, installs a Reply inbox on its own pid,
/// declares it accepts `Reply` (`expose_type` — the RFC's one kept piece of
/// ceremony: nothing is remotely deliverable by default), sends a Req with
/// `reply_to = my pid` (auto-serialized), awaits the reply.
fn role_client() {
    let server_addr = std::env::var("SMARM_SERVER_ADDR").expect("SMARM_SERVER_ADDR");
    let via_relay = std::env::var("SMARM_VIA_RELAY").is_ok();
    smarm::run(move || {
        let _cluster = start(cfg("client", vec![("server".into(), server_addr)])).expect("binds");
        let ev = subscribe().unwrap();
        wait_up(&ev, "server");
        println!("MEMBER-UP server");

        let (tx, rx) = channel::<Reply>();
        let me: Pid<Replier> = install::<Replier>(tx);
        // The one deliberate line: a pid-targeted inbound is deliverable only
        // for types this node has said it accepts (RFC §4, the safety).
        smarm::cluster::expose::expose_type::<Reply>();
        let text = if via_relay { "via-relay:ping" } else { "ping" };
        remote::send(
            RemoteName::new("server", ECHO),
            Req {
                text: text.into(),
                reply_to: RemotePid::from_local(me).expect("identity set"),
            },
        )
        .unwrap();
        println!("SENT");
        let Reply(s) = rx.recv().unwrap();
        println!("REPLY {s}");
        loop {
            smarm::sleep(Duration::from_secs(3600));
        }
    });
}

/// The gate: cross-node call/reply with no ceremony.
#[test]
fn cross_node_call_reply_no_ceremony() {
    maybe_child(ROLES);
    let mut server = spawn_node("server", &[]);
    let saddr = server.wait_listening();
    server.wait_line("READY", |l| l == "READY");
    let mut client = spawn_node("client", &[("SMARM_SERVER_ADDR", &saddr)]);
    client.wait_line("SENT", |l| l == "SENT");
    server.wait_line("REQ ping", |l| l == "REQ ping");
    client.wait_line("REPLY echo:ping", |l| l == "REPLY echo:ping");
}

/// The client's pid, round-tripped through a third node, still delivers.
#[test]
fn pid_roundtrips_through_third_node() {
    maybe_child(ROLES);
    // Relay needs the server address; server needs the relay address —
    // pre-reserve the relay port (same accepted micro-window as cluster_mesh).
    let relay_addr = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        l.local_addr().unwrap().to_string()
    };
    let mut server = spawn_node("server", &[("SMARM_RELAY_ADDR", &relay_addr)]);
    let saddr = server.wait_listening();
    server.wait_line("READY", |l| l == "READY");
    let mut relay = spawn_node(
        "relay",
        &[
            ("SMARM_SERVER_ADDR", &saddr),
            ("SMARM_LISTEN_ADDR", &relay_addr),
        ],
    );
    let _ = relay.wait_listening();
    relay.wait_line("READY", |l| l == "READY");
    let mut client = spawn_node(
        "client",
        &[("SMARM_SERVER_ADDR", &saddr), ("SMARM_VIA_RELAY", "1")],
    );
    client.wait_line("SENT", |l| l == "SENT");
    server.wait_line("FORWARDED", |l| l == "FORWARDED");
    relay.wait_line("RELAYING", |l| l.starts_with("RELAYING"));
    server.wait_line("REQ relayed:ping", |l| l == "REQ relayed:ping");
    client.wait_line("REPLY echo:relayed:ping", |l| {
        l == "REPLY echo:relayed:ping"
    });
}