Transport Lanes

//! Example: lane a (pheromone directed batches) end-to-end over loopback QUIC.
//!
//! Operator A dials operator B on the pheromone lane and delivers one
//! `PheromoneGossipBatch` using the crate's real client, `deliver_batch_over_iroh`.
//! Operator B resolves the authenticated sender from the cryptographically
//! authenticated `EndpointId` through the `DirectoryGate` (the ONE hop that, on
//! the shipped HTTP path, comes from HTTP-signature verification), then feeds that
//! resolved `kernel_id` as `authenticated_sender_kernel_id` into the REAL,
//! unchanged `chio-federation` per-frame verifier (`verify_pheromone_gossip_batch`,
//! pheromone_gossip.rs:236/244). The verifier accepts only because the transport
//! resolved the right sender; a spoofed sender fails closed.
//!
//! Why the acceptor here is a small in-example handler, not the crate's
//! `PheromoneBatchHandler`: that handler takes an `Arc<dyn RelayBatchReceiver>`
//! whose `receive_batch` returns `PheromoneReceiveReport`, a type that lives in
//! `chio-pheromone-runtime` (not a dependency of this crate and not re-exported).
//! The crate's own lane test hits the same wall and stands up a `CannedReportHandler`
//! double for exactly this reason. This example mirrors that double using only
//! public APIs (the real gate, the real wire client, the real per-frame verifier),
//! so the transport path and the load-bearing sender binding are genuine.
//!
//! Run: `cargo run -p chio-federation-transport-iroh --example pheromone_exchange`

use std::error::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;

use chio_core_types::canonical_json_bytes;
use chio_core_types::sha256_hex;
use chio_core_types::Keypair;
use chio_federation::pheromone_gossip::verify_pheromone_gossip_batch;
use chio_federation::pheromone_gossip::PheromoneDepositGossip;
use chio_federation::pheromone_gossip::PheromoneGossipBatch;
use chio_federation::pheromone_gossip::PheromoneGossipBatchVerificationContext;
use chio_federation::pheromone_gossip::PheromoneTransitPolicy;
use chio_federation::pheromone_gossip::PHEROMONE_GOSSIP_BATCH_SCHEMA;
use chio_federation::pheromone_gossip::PHEROMONE_GOSSIP_SCHEMA;
use chio_federation::pheromone_gossip::PHEROMONE_TRANSIT_POLICY_SCHEMA;
use chio_federation_transport_iroh::admission::DirectoryGate;
use chio_federation_transport_iroh::identity::transport_endorsement_preimage;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleBody;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleTrust;
use chio_federation_transport_iroh::identity::TransportDirectoryDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryEntry;
use chio_federation_transport_iroh::identity::TrustedTransportDirectoryIssuer;
use chio_federation_transport_iroh::identity::TRANSPORT_DIRECTORY_BUNDLE_SCHEMA;
use chio_federation_transport_iroh::lanes::pheromone::deliver_batch_over_iroh;
use chio_federation_transport_iroh::lanes::pheromone::ALPN_PHEROMONE_BATCH;
use chio_federation_transport_iroh::lanes::pheromone::MAX_PHEROMONE_BATCH_BYTES;
use iroh::endpoint::presets;
use iroh::endpoint::Connection;
use iroh::protocol::AcceptError;
use iroh::protocol::ProtocolHandler;
use iroh::protocol::Router;
use iroh::Endpoint;
use iroh::EndpointAddr;
use iroh::EndpointId;
use iroh::RelayMode;
use iroh::SecretKey;
use tokio::io::AsyncRead;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWrite;
use tokio::io::AsyncWriteExt;

const NOW: u64 = 1_766_000_000_500;
const OPERATOR_A: &str = "did:chio:operator-a";
const OPERATOR_B: &str = "did:chio:operator-b";
const TREATY: &str = "treaty:operator-a-operator-b:support-ops";
const NAMESPACE: &str = "dev.chio.support";

const OPERATOR_B_SEED: u8 = 31;
const OPERATOR_A_SEED: u8 = 30;
const MALLORY_SEED: u8 = 98;

fn endpoint_id(seed: u8) -> EndpointId {
    SecretKey::from_bytes(&[seed; 32]).public()
}

/// Build a load-time-verified directory (mirrors the crate's test fixtures) and
/// wrap it in a gate. Each entry binds `kernel_id` to the transport `EndpointId`
/// derived from `transport_seed`.
fn build_gate(entries: &[(&str, u8, u8)]) -> Result<DirectoryGate, Box<dyn Error>> {
    let issuer = Keypair::from_seed(&[240u8; 32]);
    let peers = entries
        .iter()
        .map(|(kernel_id, passport_seed, transport_seed)| {
            let passport = Keypair::from_seed(&[*passport_seed; 32]);
            let transport = endpoint_id(*transport_seed);
            TransportDirectoryEntry {
                kernel_id: (*kernel_id).to_string(),
                passport_public_key: passport.public_key(),
                transport_endpoint_id: transport,
                passport_endorsement: passport
                    .sign(&transport_endorsement_preimage(kernel_id, &transport)),
                revocation_signers: Vec::new(),
                removed: false,
            }
        })
        .collect::<Vec<_>>();
    let directory = TransportDirectoryDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        local_kernel_id: OPERATOR_B.to_string(),
        peers,
        treaties: Vec::new(),
    };
    let directory_sha256 = sha256_hex(&canonical_json_bytes(&directory)?);
    let body = TransportDirectoryBundleBody {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        issuer: "did:chio:issuer".to_string(),
        key_id: "issuer-key-1".to_string(),
        directory_sha256,
        version: 1,
        previous_version_sha256: None,
        issued_at_unix_ms: NOW - 1,
        expires_at_unix_ms: NOW + 1,
    };
    let (signature, _) = issuer.sign_canonical(&body)?;
    let bundle = TransportDirectoryBundleDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        body,
        directory,
        signature,
    };
    let trust = TransportDirectoryBundleTrust {
        issuers: vec![TrustedTransportDirectoryIssuer {
            issuer: "did:chio:issuer".to_string(),
            key_id: "issuer-key-1".to_string(),
            public_key: issuer.public_key(),
        }],
        version_floor: 0,
        expected_previous_version_sha256: None,
        now_unix_ms: NOW,
    };
    Ok(DirectoryGate::new(Arc::new(bundle.verify_bundle(&trust)?)))
}

async fn bind_endpoint(seed: u8, gate: Option<DirectoryGate>) -> Result<Endpoint, Box<dyn Error>> {
    let mut builder = Endpoint::builder(presets::Minimal)
        .secret_key(SecretKey::from_bytes(&[seed; 32]))
        .relay_mode(RelayMode::Disabled)
        .bind_addr((Ipv4Addr::LOCALHOST, 0))
        .map_err(|error| error.to_string())?;
    if let Some(gate) = gate {
        builder = builder.hooks(gate);
    }
    Ok(builder.bind().await.map_err(|error| error.to_string())?)
}

fn direct_addr(endpoint: &Endpoint) -> Result<EndpointAddr, Box<dyn Error>> {
    let socket = endpoint
        .bound_sockets()
        .into_iter()
        .next()
        .ok_or("endpoint bound no socket")?;
    Ok(EndpointAddr::new(endpoint.id()).with_ip_addr(socket))
}

/// A well-formed `chio_core_types::Signature` in its JSON form. The direct-frame
/// verifier checks sender equality, not the deposit signature, so any parseable
/// signature exercises the sender-binding path (mirrors the crate's lane test).
fn signature_value() -> Result<serde_json::Value, Box<dyn Error>> {
    let sig = Keypair::from_seed(&[9u8; 32]).sign(b"pheromone-example-fixture");
    Ok(serde_json::to_value(sig)?)
}

/// A single-frame direct batch authored by `author` (both `origin` and
/// `gossiping_peer`), scoped to `TREATY`. Built by deserialization so the example
/// need not name `chio-pheromone`'s deposit type.
fn direct_batch(author: &str) -> Result<PheromoneGossipBatch, Box<dyn Error>> {
    let frame = serde_json::json!({
        "schema": PHEROMONE_GOSSIP_SCHEMA,
        "deposit": {
            "schema": "chio.pheromone-deposit.v1",
            "kernel_id": author,
            "agent_passport_key_hash": "a".repeat(64),
            "agent_passport_jwk_thumbprint": "b".repeat(43),
            "subject_class": "support.prompt_injection",
            "subject_class_namespace": NAMESPACE,
            "indicator": {"digest": "c".repeat(64)},
            "severity": "high",
            "confidence": 0.8,
            "timestamp_unix_ms": NOW,
            "decay_half_life_secs": 3_600.0,
            "nonce": "nonce-operator-a-001",
            "treaty_scope": [TREATY],
            "signature": signature_value()?,
        },
        "origin_kernel_id": author,
        "gossiping_peer_kernel_id": author,
        "treaty_id": TREATY,
        "ts_unix_ms": NOW,
    });
    let frame: PheromoneDepositGossip = serde_json::from_value(frame)?;
    Ok(PheromoneGossipBatch {
        schema: PHEROMONE_GOSSIP_BATCH_SCHEMA.to_string(),
        recipient_kernel_id: OPERATOR_B.to_string(),
        treaty_id: TREATY.to_string(),
        frames: vec![frame],
        flushed_at_unix_ms: NOW,
    })
}

fn live_policy() -> PheromoneTransitPolicy {
    PheromoneTransitPolicy {
        schema: PHEROMONE_TRANSIT_POLICY_SCHEMA.to_string(),
        accepted_hubs: Vec::new(),
        allowed_ingress_treaties: vec![TREATY.to_string()],
        allowed_egress_treaties: vec![TREATY.to_string()],
        allowed_subject_class_namespaces: vec![NAMESPACE.to_string()],
        valid_from_unix_ms: NOW - 1_000,
        valid_until_unix_ms: NOW + 1_000,
        max_hops: 4,
        required_action_class_id: "action:demo".to_string(),
        pinned_ladder_refs: Vec::new(),
    }
}

async fn read_u32_frame<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Vec<u8>, Box<dyn Error>> {
    let len = reader.read_u32().await? as usize;
    if len > MAX_PHEROMONE_BATCH_BYTES {
        return Err("pheromone frame exceeds the transport cap".into());
    }
    let mut buf = vec![0u8; len];
    reader.read_exact(&mut buf).await?;
    Ok(buf)
}

async fn write_u32_frame<W: AsyncWrite + Unpin>(
    writer: &mut W,
    bytes: &[u8],
) -> Result<(), Box<dyn Error>> {
    let len = u32::try_from(bytes.len())?;
    writer.write_u32(len).await?;
    writer.write_all(bytes).await?;
    writer.flush().await?;
    Ok(())
}

/// Operator B's acceptor. It resolves the authenticated sender via the real gate,
/// then drives the real `chio-federation` verifier with that resolved kernel id.
#[derive(Clone)]
struct ReceiverHandler {
    gate: DirectoryGate,
    policy: Arc<PheromoneTransitPolicy>,
}

impl std::fmt::Debug for ReceiverHandler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ReceiverHandler").finish_non_exhaustive()
    }
}

impl ReceiverHandler {
    async fn handle(&self, conn: &Connection) -> Result<(), Box<dyn Error>> {
        // The ONE hop that replaces the shipped HTTP-signed sender: the resolved
        // kernel id bound to the authenticated EndpointId.
        let sender = self
            .gate
            .resolve(&conn.remote_id())
            .ok_or("unadmitted endpoint reached the pheromone handler")?;
        println!(
            "  [B] authenticated EndpointId {} -> kernel_id {sender:?}",
            conn.remote_id().fmt_short()
        );

        let (mut send, mut recv) = conn.accept_bi().await.map_err(|error| error.to_string())?;
        let raw = read_u32_frame(&mut recv).await?;
        let batch: PheromoneGossipBatch = serde_json::from_slice(&raw)?;

        // Drive the REAL, unchanged per-frame verifier with the transport-resolved
        // sender as authenticated_sender_kernel_id (pheromone_gossip.rs:236/244).
        let context = PheromoneGossipBatchVerificationContext {
            now_unix_ms: NOW,
            recipient_kernel_id: batch.recipient_kernel_id.clone(),
            authenticated_sender_kernel_id: sender.clone(),
        };
        let verdict = verify_pheromone_gossip_batch(&batch, &self.policy, &context);
        let accepted = verdict.is_ok();
        println!("  [B] verify_pheromone_gossip_batch -> accepted={accepted}");

        // A COMPLETE receive report, mirroring the runtime `PheromoneReceiveReport`.
        // The dial side rejects a partial report before marking a batch delivered, so
        // every required field is present (plus a diagnostic `verifierOutcome`, which
        // the forward-compatible validation ignores).
        let frame_count = batch.frames.len() as u64;
        let report = serde_json::json!({
            "schema": "chio.pheromone-receive-report.v1",
            "accepted": accepted,
            "batchOutcome": if accepted { "accepted" } else { "rejected" },
            "acceptedFrameCount": if accepted { frame_count } else { 0 },
            "rejectedFrameCount": if accepted { 0 } else { frame_count },
            "batchSha256": "0".repeat(64),
            "recipientKernelId": batch.recipient_kernel_id,
            "authenticatedSenderKernelId": sender,
            "receivedAtUnixMs": NOW,
            "frames": [],
            "verifierOutcome": match &verdict {
                Ok(()) => "accepted".to_string(),
                Err(error) => error.to_string(),
            },
        });
        let bytes = serde_json::to_vec(&report)?;
        write_u32_frame(&mut send, &bytes).await?;
        send.finish().map_err(|error| error.to_string())?;
        conn.closed().await;
        Ok(())
    }
}

impl ProtocolHandler for ReceiverHandler {
    async fn accept(&self, conn: Connection) -> Result<(), AcceptError> {
        self.handle(&conn)
            .await
            .map_err(|error| AcceptError::from_err(std::io::Error::other(error.to_string())))
    }
}

async fn deliver_with_timeout(
    endpoint: &Endpoint,
    addr: EndpointAddr,
    batch: &PheromoneGossipBatch,
) -> Result<bool, String> {
    match tokio::time::timeout(
        Duration::from_secs(15),
        deliver_batch_over_iroh(endpoint, addr, batch),
    )
    .await
    {
        Ok(Ok(outcome)) => Ok(outcome.accepted),
        Ok(Err(error)) => Err(error.to_string()),
        Err(_elapsed) => Err("delivery timed out".to_string()),
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    println!("== lane a: pheromone directed batch (operator A -> operator B) ==\n");

    // B's directory admits operator A (bound to A's transport key). Mallory is
    // bound to nothing.
    let gate = build_gate(&[(OPERATOR_A, 1, OPERATOR_A_SEED)])?;

    let acceptor = bind_endpoint(OPERATOR_B_SEED, Some(gate.clone())).await?;
    let router = Router::builder(acceptor)
        .accept(
            ALPN_PHEROMONE_BATCH,
            ReceiverHandler {
                gate,
                policy: Arc::new(live_policy()),
            },
        )
        .spawn();
    let acceptor_addr = direct_addr(router.endpoint())?;

    // Operator A: admitted. Its batch is authored by OPERATOR_A, so the resolved
    // sender and the frame author agree and the verifier accepts.
    let operator_a = bind_endpoint(OPERATOR_A_SEED, None).await?;
    let batch = direct_batch(OPERATOR_A)?;
    println!(
        "operator A delivers a 1-frame batch on {:?}:",
        String::from_utf8_lossy(ALPN_PHEROMONE_BATCH)
    );
    let admitted_ok = match deliver_with_timeout(&operator_a, acceptor_addr.clone(), &batch).await {
        Ok(accepted) => {
            println!("  [A] peer report: accepted={accepted}\n");
            accepted
        }
        Err(error) => {
            println!("  [A] delivery failed: {error}\n");
            false
        }
    };

    // Mallory: unadmitted. The gate rejects at handshake before the handler runs.
    let mallory = bind_endpoint(MALLORY_SEED, None).await?;
    println!("an unadmitted operator (mallory) attempts the same delivery:");
    let mallory_rejected = match deliver_with_timeout(&mallory, acceptor_addr, &batch).await {
        Ok(accepted) => {
            println!("  [mallory] UNEXPECTEDLY delivered, accepted={accepted}");
            false
        }
        Err(error) => {
            println!("  [mallory] rejected before any handler ran: {error}");
            true
        }
    };

    router.shutdown().await.ok();

    println!("\nfail-closed summary:");
    println!("  admitted sender resolved + batch accepted: {admitted_ok}");
    println!("  unadmitted sender rejected at the gate:     {mallory_rejected}");
    if admitted_ok && mallory_rejected {
        println!("\nOK: the transport-resolved kernel_id fed the real verifier and it accepted.");
        Ok(())
    } else {
        Err("pheromone lane invariant violated".into())
    }
}

//! Example: lane c (cross-operator fan-out) end-to-end over a loopback
//! iroh-gossip swarm.
//!
//! Three admitted operators join one per-treaty gossip topic in a chain
//! (A seed; B bootstraps from A; C from B). A authors a self-signed pheromone
//! deposit and broadcasts it; C receives it RELAYED through B. In gossip,
//! `Message::delivered_from` is the forwarding NEIGHBOR (here B), NOT the author
//! (A), so authorship must be established FROM THE PAYLOAD ALONE: the
//! transport-independent frame verifier plus the embedded deposit's OWN
//! self-signature, checked against the origin operator's passport key resolved by
//! `origin_kernel_id` (never from `delivered_from`).
//!
//! The load-bearing property: a frame relayed by an ADMITTED neighbor but whose
//! deposit self-signature is invalid is REJECTED. Being forwarded by a trusted,
//! directory-admitted swarm member does not launder a bad payload, because
//! `delivered_from` is never an input to verification.
//!
//! Run: `cargo run -p chio-federation-transport-iroh --example fanout_gossip`

use std::error::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;

use chio_core_types::canonical_json_bytes;
use chio_core_types::sha256_hex;
use chio_core_types::Keypair;
use chio_federation::pheromone_gossip::PheromoneDepositGossip;
use chio_federation::pheromone_gossip::PheromoneTransitPolicy;
use chio_federation::pheromone_gossip::PHEROMONE_GOSSIP_SCHEMA;
use chio_federation::pheromone_gossip::PHEROMONE_TRANSIT_POLICY_SCHEMA;
use chio_federation_transport_iroh::admission::DirectoryGate;
use chio_federation_transport_iroh::identity::transport_endorsement_preimage;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleBody;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleTrust;
use chio_federation_transport_iroh::identity::TransportDirectoryDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryEntry;
use chio_federation_transport_iroh::identity::TrustedTransportDirectoryIssuer;
use chio_federation_transport_iroh::identity::VerifiedDirectory;
use chio_federation_transport_iroh::identity::TRANSPORT_DIRECTORY_BUNDLE_SCHEMA;
use chio_federation_transport_iroh::lanes::fanout::decode_and_verify_fanout_frame;
use chio_federation_transport_iroh::lanes::fanout::FanoutError;
use chio_federation_transport_iroh::lanes::fanout::FanoutLane;
use chio_federation_transport_iroh::lanes::fanout::FanoutTopic;
use chio_federation_transport_iroh::lanes::fanout::StaticOriginKeys;
use chio_federation_transport_iroh::lanes::fanout::StaticTreatyMembership;
use chio_pheromone::sign_deposit;
use chio_pheromone::PheromoneDepositBody;
use chio_pheromone::Severity;
use chio_pheromone::PHEROMONE_DEPOSIT_SCHEMA;
use iroh::address_lookup::memory::MemoryLookup;
use iroh::endpoint::presets;
use iroh::protocol::Router;
use iroh::Endpoint;
use iroh::EndpointAddr;
use iroh::EndpointId;
use iroh::RelayMode;
use iroh::SecretKey;
use iroh_gossip::Gossip;

const NOW: u64 = 1_700_000_000_000;
const NAMESPACE: &str = "chio/agents";
const TREATY: &str = "treaty:reputation-clearing";
const AUTHOR: &str = "did:chio:author";

const NODE_A_SEED: u8 = 61;
const NODE_B_SEED: u8 = 62;
const NODE_C_SEED: u8 = 63;

fn endpoint_id(seed: u8) -> EndpointId {
    SecretKey::from_bytes(&[seed; 32]).public()
}

/// Build a load-time-verified directory admitting every node's transport
/// endpoint and wrap it in a gate. All three swarm members are admitted, so every
/// gossip connection passes the accept-time gate (the "admitted neighbor" story).
fn build_gate(entries: &[(&str, u8, u8)]) -> Result<DirectoryGate, Box<dyn Error>> {
    let issuer = Keypair::from_seed(&[240u8; 32]);
    let peers = entries
        .iter()
        .map(|(kernel_id, passport_seed, transport_seed)| {
            let passport = Keypair::from_seed(&[*passport_seed; 32]);
            let transport = endpoint_id(*transport_seed);
            TransportDirectoryEntry {
                kernel_id: (*kernel_id).to_string(),
                passport_public_key: passport.public_key(),
                transport_endpoint_id: transport,
                passport_endorsement: passport
                    .sign(&transport_endorsement_preimage(kernel_id, &transport)),
                revocation_signers: Vec::new(),
                removed: false,
            }
        })
        .collect::<Vec<_>>();
    let directory = TransportDirectoryDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        local_kernel_id: "did:chio:swarm".to_string(),
        peers,
        treaties: Vec::new(),
    };
    let directory_sha256 = sha256_hex(&canonical_json_bytes(&directory)?);
    let body = TransportDirectoryBundleBody {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        issuer: "did:chio:issuer".to_string(),
        key_id: "issuer-key-1".to_string(),
        directory_sha256,
        version: 1,
        previous_version_sha256: None,
        issued_at_unix_ms: NOW - 1,
        expires_at_unix_ms: NOW + 1,
    };
    let (signature, _) = issuer.sign_canonical(&body)?;
    let bundle = TransportDirectoryBundleDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        body,
        directory,
        signature,
    };
    let trust = TransportDirectoryBundleTrust {
        issuers: vec![TrustedTransportDirectoryIssuer {
            issuer: "did:chio:issuer".to_string(),
            key_id: "issuer-key-1".to_string(),
            public_key: issuer.public_key(),
        }],
        version_floor: 0,
        expected_previous_version_sha256: None,
        now_unix_ms: NOW,
    };
    let verified: VerifiedDirectory = bundle.verify_bundle(&trust)?;
    Ok(DirectoryGate::new(Arc::new(verified)))
}

/// One swarm node: a gated endpoint, an iroh-gossip protocol handler mounted on a
/// router, and a `FanoutLane` over the gossip handle.
struct Node {
    id: EndpointId,
    addr: EndpointAddr,
    lookup: MemoryLookup,
    lane: FanoutLane,
    router: Router,
}

async fn spawn_node(seed: u8, gate: DirectoryGate) -> Result<Node, Box<dyn Error>> {
    let lookup = MemoryLookup::new();
    let endpoint = Endpoint::builder(presets::Minimal)
        .secret_key(SecretKey::from_bytes(&[seed; 32]))
        .relay_mode(RelayMode::Disabled)
        .bind_addr((Ipv4Addr::LOCALHOST, 0))
        .map_err(|error| error.to_string())?
        .hooks(gate)
        .address_lookup(lookup.clone())
        .bind()
        .await
        .map_err(|error| error.to_string())?;
    let socket = endpoint
        .bound_sockets()
        .into_iter()
        .next()
        .ok_or("node bound no socket")?;
    let id = endpoint.id();
    let addr = EndpointAddr::new(id).with_ip_addr(socket);

    // Spawn gossip and mount it on a router under its own ALPN (gossip owns the
    // ALPN; this lane does not invent one). Membership is gated upstream.
    let gossip = Gossip::builder().spawn(endpoint.clone());
    // Build the lane before the router moves `endpoint`: the lane derives its
    // authenticated local id from this endpoint, so it must see the real one.
    let lane = FanoutLane::new(gossip.clone(), &endpoint);
    let router = Router::builder(endpoint)
        .accept(iroh_gossip::ALPN, gossip)
        .spawn();
    Ok(Node {
        id,
        addr,
        lookup,
        lane,
        router,
    })
}

fn live_policy() -> PheromoneTransitPolicy {
    PheromoneTransitPolicy {
        schema: PHEROMONE_TRANSIT_POLICY_SCHEMA.to_string(),
        accepted_hubs: Vec::new(),
        allowed_ingress_treaties: Vec::new(),
        allowed_egress_treaties: Vec::new(),
        allowed_subject_class_namespaces: vec![NAMESPACE.to_string()],
        valid_from_unix_ms: NOW - 1,
        valid_until_unix_ms: NOW + 1_000_000,
        max_hops: 4,
        required_action_class_id: "action".to_string(),
        pinned_ladder_refs: Vec::new(),
    }
}

/// Build a self-signed deposit frame authored by `author`. When `tamper` is set,
/// a signed-but-frame-irrelevant field is mutated AFTER signing: the frame-level
/// checks still pass, but the deposit self-signature no longer matches.
fn build_frame(
    author: &Keypair,
    nonce: &str,
    tamper: bool,
) -> Result<PheromoneDepositGossip, Box<dyn Error>> {
    let body = PheromoneDepositBody {
        schema: PHEROMONE_DEPOSIT_SCHEMA.to_string(),
        kernel_id: AUTHOR.to_string(),
        agent_passport_key_hash: "passport-key-hash".to_string(),
        agent_passport_jwk_thumbprint: "passport-thumbprint".to_string(),
        subject_class: "malicious-tool".to_string(),
        subject_class_namespace: NAMESPACE.to_string(),
        indicator: serde_json::json!({ "kind": "observation" }),
        severity: Severity::High,
        confidence: 0.8,
        timestamp_unix_ms: NOW,
        decay_half_life_secs: 3600.0,
        evaporation_floor: None,
        nonce: nonce.to_string(),
        treaty_scope: vec![TREATY.to_string()],
        cost_commitment: None,
        workflow_context: None,
    };
    let deposit = sign_deposit(body, author).map_err(|error| error.to_string())?;
    let mut frame = PheromoneDepositGossip {
        schema: PHEROMONE_GOSSIP_SCHEMA.to_string(),
        deposit,
        origin_kernel_id: AUTHOR.to_string(),
        // The re-gossiping hub, deliberately NOT the author: fan-out does not
        // require gossiping_peer == author (unlike the direct lanes).
        gossiping_peer_kernel_id: "did:chio:relay-hub".to_string(),
        treaty_id: TREATY.to_string(),
        ts_unix_ms: NOW,
        transit_chain: None,
    };
    if tamper {
        frame.deposit.body.confidence = 0.123_456;
    }
    Ok(frame)
}

/// The receive-side verification context (payload-only; never `delivered_from`).
struct Verifier<'a> {
    origin_keys: &'a StaticOriginKeys,
    membership: &'a StaticTreatyMembership,
    policy: &'a PheromoneTransitPolicy,
}

/// Rebroadcast fresh frames (from `make_frame`) on `author_topic` until
/// `receiver_topic` yields a payload the predicate accepts, or the deadline
/// lapses. Fresh nonces avoid gossip dedup between attempts.
async fn drive<M, F>(
    author_topic: &FanoutTopic,
    receiver_topic: &mut FanoutTopic,
    verifier: &Verifier<'_>,
    mut make_frame: M,
    mut accept: F,
) -> Result<bool, Box<dyn Error>>
where
    M: FnMut(u64) -> Result<PheromoneDepositGossip, Box<dyn Error>>,
    F: FnMut(EndpointId, &Result<PheromoneDepositGossip, FanoutError>) -> bool,
{
    let deadline = Instant::now() + Duration::from_secs(20);
    // Bind the raw decode to the receiver's own treaty so a foreign-treaty frame
    // could never be accepted on this swarm (captured before the receive borrow).
    let receiver_treaty = receiver_topic.treaty_id().to_string();
    let mut attempt = 0u64;
    while Instant::now() < deadline {
        let frame = make_frame(attempt)?;
        attempt += 1;
        author_topic.broadcast(&frame).await?;
        match tokio::time::timeout(Duration::from_millis(400), receiver_topic.next_payload()).await
        {
            Ok(Some(Ok(message))) => {
                let verdict = decode_and_verify_fanout_frame(
                    &message.content,
                    &receiver_treaty,
                    verifier.origin_keys,
                    verifier.membership,
                    verifier.policy,
                    NOW,
                );
                if accept(message.delivered_from, &verdict) {
                    return Ok(true);
                }
            }
            Ok(Some(Err(error))) => return Err(error.to_string().into()),
            Ok(None) => return Err("gossip topic closed unexpectedly".into()),
            Err(_elapsed) => continue,
        }
    }
    Ok(false)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    println!("== lane c: cross-operator fan-out over an iroh-gossip swarm ==\n");

    let gate = build_gate(&[
        ("did:chio:node-a", 1, NODE_A_SEED),
        ("did:chio:node-b", 2, NODE_B_SEED),
        ("did:chio:node-c", 3, NODE_C_SEED),
    ])?;

    // Chain topology: A (seed) <- B <- C. B must know A's address; C must know
    // B's; the seed needs nobody (relay + discovery are disabled on loopback).
    let node_a = spawn_node(NODE_A_SEED, gate.clone()).await?;
    let node_b = spawn_node(NODE_B_SEED, gate.clone()).await?;
    let node_c = spawn_node(NODE_C_SEED, gate).await?;
    node_b.lookup.add_endpoint_info(node_a.addr.clone());
    node_c.lookup.add_endpoint_info(node_b.addr.clone());

    // The issuer-signed per-treaty party set the fan-out gate consults. Every node
    // (and the frame author) is a party to TREATY, so join and receive both admit;
    // a non-party would be rejected with FanoutError::TreatyMembershipDenied.
    // The join gate resolves each node's authenticated EndpointId to its kernel id
    // through this same membership, so bind every node's endpoint here.
    let membership = StaticTreatyMembership::new()
        .with(
            TREATY,
            [
                "did:chio:node-a",
                "did:chio:node-b",
                "did:chio:node-c",
                AUTHOR,
            ],
        )
        .with_endpoint(node_a.id, "did:chio:node-a")
        .with_endpoint(node_b.id, "did:chio:node-b")
        .with_endpoint(node_c.id, "did:chio:node-c");

    println!("joining topic for {TREATY:?} (A seed; B<-A; C<-B) ...");
    let joined = tokio::time::timeout(Duration::from_secs(30), async {
        tokio::join!(
            node_a
                .lane
                .subscribe_treaty(TREATY, "did:chio:node-a", &membership, vec![]),
            node_b
                .lane
                .subscribe_treaty(TREATY, "did:chio:node-b", &membership, vec![node_a.id]),
            node_c
                .lane
                .subscribe_treaty(TREATY, "did:chio:node-c", &membership, vec![node_b.id]),
        )
    })
    .await
    .map_err(|_elapsed| "gossip join timed out")?;
    let a_topic = joined.0?;
    // B's topic is the relay hop: hold it so B keeps forwarding A -> C.
    let _b_topic = joined.1?;
    let mut c_topic = joined.2?;
    println!(
        "  joined; topic id {}\n",
        hex_short(a_topic.topic_id().as_bytes())
    );

    // The author's passport key, trusted by C's origin resolver. This is resolved
    // from the PAYLOAD origin, never from delivered_from.
    let author = Keypair::from_seed(&[9u8; 32]);
    let origin_keys = StaticOriginKeys::new().with(AUTHOR, author.public_key());
    let policy = live_policy();
    let verifier = Verifier {
        origin_keys: &origin_keys,
        membership: &membership,
        policy: &policy,
    };

    // ---- 1. Valid self-signed frame: accepted; delivered_from is the relay ---
    println!("A broadcasts a VALID self-signed deposit authored by {AUTHOR:?}:");
    let valid_ok = drive(
        &a_topic,
        &mut c_topic,
        &verifier,
        |attempt| build_frame(&author, &format!("valid-{attempt}"), false),
        |delivered_from, verdict| match verdict {
            Ok(frame) => {
                let relayed = delivered_from == node_b.id;
                println!(
                    "  [C] received: delivered_from={} (relay B={}), payload origin_kernel_id={:?}",
                    delivered_from.fmt_short(),
                    node_b.id.fmt_short(),
                    frame.origin_kernel_id
                );
                println!("  [C] origin-verified from the payload -> ACCEPTED (delivered_from ignored, relayed={relayed})");
                true
            }
            Err(_) => false,
        },
    )
    .await?;

    // ---- 2. Tampered frame from the same admitted relay: rejected ------------
    println!("\nA broadcasts a frame with an INVALID deposit signature (relayed by admitted B):");
    let tampered_rejected = drive(
        &a_topic,
        &mut c_topic,
        &verifier,
        |attempt| build_frame(&author, &format!("tampered-{attempt}"), true),
        |delivered_from, verdict| match verdict {
            Err(FanoutError::DepositSignatureInvalid) => {
                println!(
                    "  [C] received via admitted neighbor delivered_from={} -> REJECTED (DepositSignatureInvalid)",
                    delivered_from.fmt_short()
                );
                true
            }
            // Skip any leftover valid frames still in flight from phase 1.
            _ => false,
        },
    )
    .await?;

    // Graceful teardown: drop the topic subscriptions, then shut the routers down
    // so the gossip actors stop cleanly (an abrupt runtime drop can otherwise
    // cancel a gossip forwarder task mid-flight and log a teardown panic).
    drop(a_topic);
    drop(_b_topic);
    drop(c_topic);
    node_a.router.shutdown().await.ok();
    node_b.router.shutdown().await.ok();
    node_c.router.shutdown().await.ok();

    println!("\nfail-closed summary:");
    println!("  valid relayed frame accepted from payload:        {valid_ok}");
    println!("  tampered frame from admitted neighbor rejected:   {tampered_rejected}");
    if valid_ok && tampered_rejected {
        println!("\nOK: authorship comes from the signed payload; delivered_from never launders a frame.");
        Ok(())
    } else {
        Err("fan-out lane invariant violated".into())
    }
}

/// Short hex prefix of a byte slice, for a readable topic id in the log.
fn hex_short(bytes: &[u8]) -> String {
    bytes
        .iter()
        .take(5)
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

//! Example: lane d (bilateral DSSE co-sign) end-to-end over loopback QUIC.
//!
//! Org B requests a DSSE co-signature from Org A over a dedicated-ALPN
//! bidirectional QUIC RPC (categorically NOT gossip: an in-flight statement must
//! not leak to non-parties). Org A's accept-time gate 403-rejects any unadmitted
//! Org B; past the gate, Org A verifies `org_b_signature` over the exact
//! `pae_bytes` against Org B's pinned passport key AND binds the authenticated
//! `EndpointId` to the claimed `org_b_kernel_id`. Only then does Org A co-sign the
//! SAME opaque bytes (never re-derived). On any failure it answers with a typed
//! error and NEVER signs.
//!
//! This example drives the crate's real client (`IrohBilateralCoSigner`) against
//! the real server handler (`BilateralCoSignHandler`): a full co-sign whose
//! response verifies over `pae_bytes`, and an Org B that claims a different kernel
//! id than it authenticated as, refused without a signature.
//!
//! Run: `cargo run -p chio-federation-transport-iroh --example bilateral_cosign`

use std::collections::HashMap;
use std::error::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;

use chio_core_types::canonical_json_bytes;
use chio_core_types::sha256_hex;
use chio_core_types::Keypair;
use chio_core_types::PublicKey;
use chio_federation::bilateral::BilateralCoSigningError;
use chio_federation::bilateral::DsseCoSigningRequest;
use chio_federation::bilateral::DsseCoSigningResponse;
use chio_federation::bilateral::BILATERAL_DSSE_COSIGNING_SCHEMA;
use chio_federation_transport_iroh::admission::DirectoryGate;
use chio_federation_transport_iroh::identity::transport_endorsement_preimage;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleBody;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleTrust;
use chio_federation_transport_iroh::identity::TransportDirectoryDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryEntry;
use chio_federation_transport_iroh::identity::TrustedTransportDirectoryIssuer;
use chio_federation_transport_iroh::identity::VerifiedDirectory;
use chio_federation_transport_iroh::identity::TRANSPORT_DIRECTORY_BUNDLE_SCHEMA;
use chio_federation_transport_iroh::lanes::bilateral::BilateralCoSignHandler;
use chio_federation_transport_iroh::lanes::bilateral::IrohBilateralCoSigner;
use chio_federation_transport_iroh::lanes::bilateral::PinnedPassportKeys;
use chio_federation_transport_iroh::lanes::bilateral::ALPN_BILATERAL;
use iroh::endpoint::presets;
use iroh::protocol::Router;
use iroh::Endpoint;
use iroh::EndpointAddr;
use iroh::EndpointId;
use iroh::RelayMode;
use iroh::SecretKey;

const NOW: u64 = 2_000_000;
const ORG_A: &str = "did:chio:org-a";
const ORG_B: &str = "did:chio:org-b";

/// A federation participant: an ed25519 transport identity plus a long-term
/// passport keypair (the algorithm-agnostic co-signing key material).
struct Peer {
    kernel_id: String,
    transport_secret: SecretKey,
    transport_id: EndpointId,
    passport: Keypair,
}

impl Peer {
    fn new(kernel_id: &str, transport_seed: u8, passport_seed: u8) -> Self {
        let transport_secret = SecretKey::from_bytes(&[transport_seed; 32]);
        let transport_id = transport_secret.public();
        Self {
            kernel_id: kernel_id.to_string(),
            transport_secret,
            transport_id,
            passport: Keypair::from_seed(&[passport_seed; 32]),
        }
    }

    fn entry(&self) -> TransportDirectoryEntry {
        TransportDirectoryEntry {
            kernel_id: self.kernel_id.clone(),
            passport_public_key: self.passport.public_key(),
            transport_endpoint_id: self.transport_id,
            passport_endorsement: self.passport.sign(&transport_endorsement_preimage(
                &self.kernel_id,
                &self.transport_id,
            )),
            revocation_signers: Vec::new(),
            removed: false,
        }
    }
}

/// Build a load-time-verified directory admitting the given peers.
fn verified_directory(peers: &[&Peer]) -> Result<Arc<VerifiedDirectory>, Box<dyn Error>> {
    let issuer = Keypair::from_seed(&[240u8; 32]);
    let directory = TransportDirectoryDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        local_kernel_id: ORG_A.to_string(),
        peers: peers.iter().map(|peer| peer.entry()).collect(),
        treaties: Vec::new(),
    };
    let directory_sha256 = sha256_hex(&canonical_json_bytes(&directory)?);
    let body = TransportDirectoryBundleBody {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        issuer: "did:chio:issuer".to_string(),
        key_id: "issuer-key-1".to_string(),
        directory_sha256,
        version: 1,
        previous_version_sha256: None,
        issued_at_unix_ms: NOW - 1,
        expires_at_unix_ms: NOW + 1,
    };
    let (signature, _) = issuer.sign_canonical(&body)?;
    let bundle = TransportDirectoryBundleDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        body,
        directory,
        signature,
    };
    let trust = TransportDirectoryBundleTrust {
        issuers: vec![TrustedTransportDirectoryIssuer {
            issuer: "did:chio:issuer".to_string(),
            key_id: "issuer-key-1".to_string(),
            public_key: issuer.public_key(),
        }],
        version_floor: 0,
        expected_previous_version_sha256: None,
        now_unix_ms: NOW,
    };
    Ok(Arc::new(bundle.verify_bundle(&trust)?))
}

/// Stand up Org A: a gated loopback endpoint with the co-sign handler mounted on
/// the bilateral ALPN. Returns Org A's dialable address and the live router.
async fn spawn_org_a(
    org_a: &Peer,
    gate: DirectoryGate,
    passport_keys: Arc<dyn PinnedPassportKeys>,
) -> Result<(EndpointAddr, Router), Box<dyn Error>> {
    let endpoint = Endpoint::builder(presets::Minimal)
        .secret_key(org_a.transport_secret.clone())
        .relay_mode(RelayMode::Disabled)
        .bind_addr((Ipv4Addr::LOCALHOST, 0))
        .map_err(|error| error.to_string())?
        .hooks(gate.clone())
        .bind()
        .await
        .map_err(|error| error.to_string())?;
    let socket = endpoint
        .bound_sockets()
        .into_iter()
        .next()
        .ok_or("org a bound no socket")?;
    let addr = EndpointAddr::new(org_a.transport_id).with_ip_addr(socket);

    let handler = BilateralCoSignHandler::new(
        gate,
        org_a.kernel_id.clone(),
        org_a.passport.clone(),
        passport_keys,
    );
    let router = Router::builder(endpoint)
        .accept(ALPN_BILATERAL, handler)
        .spawn();
    Ok((addr, router))
}

/// Build Org B: a loopback client endpoint plus a co-signer that dials `addr` for
/// `org_a_kernel_id`.
async fn spawn_org_b(
    org_b: &Peer,
    org_a_kernel_id: &str,
    addr: EndpointAddr,
) -> Result<IrohBilateralCoSigner, Box<dyn Error>> {
    let endpoint = Endpoint::builder(presets::Minimal)
        .secret_key(org_b.transport_secret.clone())
        .relay_mode(RelayMode::Disabled)
        .bind_addr((Ipv4Addr::LOCALHOST, 0))
        .map_err(|error| error.to_string())?
        .bind()
        .await
        .map_err(|error| error.to_string())?;
    let mut book: HashMap<String, EndpointAddr> = HashMap::new();
    book.insert(org_a_kernel_id.to_string(), addr);
    Ok(IrohBilateralCoSigner::new(endpoint, Arc::new(book)))
}

/// Org A's pinned Org B passport keys (the algorithm-agnostic verify set).
fn pinned_org_b(org_b: &Peer) -> Arc<dyn PinnedPassportKeys> {
    let mut keys: HashMap<String, PublicKey> = HashMap::new();
    keys.insert(org_b.kernel_id.clone(), org_b.passport.public_key());
    Arc::new(keys)
}

async fn cosign_with_timeout(
    cosigner: &IrohBilateralCoSigner,
    request: &DsseCoSigningRequest,
) -> Result<Result<DsseCoSigningResponse, BilateralCoSigningError>, String> {
    tokio::time::timeout(
        Duration::from_secs(15),
        cosigner.request_dsse_cosignature_over_iroh(request),
    )
    .await
    .map_err(|_elapsed| "co-sign timed out".to_string())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    println!("== lane d: bilateral DSSE co-sign (Org B -> Org A) ==\n");

    let org_a = Peer::new(ORG_A, 51, 1);
    let org_b = Peer::new(ORG_B, 52, 2);
    let gate = DirectoryGate::new(verified_directory(&[&org_a, &org_b])?);

    let org_a_public = org_a.passport.public_key();
    let (addr, router) = spawn_org_a(&org_a, gate, pinned_org_b(&org_b)).await?;
    let cosigner = spawn_org_b(&org_b, ORG_A, addr).await?;

    // ---- 1. A full co-sign: the response verifies over the exact pae_bytes ---
    let pae_bytes = b"DSSEv1 opaque bilateral pae preimage".to_vec();
    let request = DsseCoSigningRequest::new(
        ORG_A.to_string(),
        org_b.kernel_id.clone(),
        pae_bytes.clone(),
        // Org B signs the PAE bytes with its passport key.
        org_b.passport.sign(&pae_bytes),
    );
    println!(
        "Org B requests a co-signature over {} pae bytes:",
        pae_bytes.len()
    );
    let cosign_ok = match cosign_with_timeout(&cosigner, &request).await {
        Ok(Ok(response)) => {
            let verifies = org_a_public.verify(&pae_bytes, &response.org_a_signature);
            let rejects_other = !org_a_public.verify(b"other bytes", &response.org_a_signature);
            println!("  -> Org A co-signed (schema={:?})", response.schema);
            println!("     org_a_signature verifies over pae_bytes:      {verifies}");
            println!("     org_a_signature rejects a different message:  {rejects_other}");
            verifies && rejects_other && response.schema == BILATERAL_DSSE_COSIGNING_SCHEMA
        }
        Ok(Err(error)) => {
            println!("  -> unexpected co-sign failure: {error}");
            false
        }
        Err(error) => {
            println!("  -> {error}");
            false
        }
    };

    // ---- 2. Org B claims a different kernel id than it authenticated as ------
    let spoof_pae = b"pae for a spoofed org_b".to_vec();
    let spoof_request = DsseCoSigningRequest::new(
        ORG_A.to_string(),
        "did:chio:evil-impersonator".to_string(),
        spoof_pae.clone(),
        org_b.passport.sign(&spoof_pae),
    );
    println!("\nOrg B (authenticated as {ORG_B}) claims to be did:chio:evil-impersonator:");
    let spoof_rejected = match cosign_with_timeout(&cosigner, &spoof_request).await {
        Ok(Err(BilateralCoSigningError::UnknownPeer(peer))) => {
            println!("  -> refused without signing: UnknownPeer({peer:?})");
            peer == "did:chio:evil-impersonator"
        }
        Ok(Err(other)) => {
            println!("  -> refused without signing: {other}");
            true
        }
        Ok(Ok(_response)) => {
            println!("  -> UNEXPECTEDLY co-signed for a spoofed identity");
            false
        }
        Err(error) => {
            println!("  -> {error}");
            false
        }
    };

    router.shutdown().await.ok();

    println!("\nfail-closed summary:");
    println!("  co-signature verifies over pae_bytes:        {cosign_ok}");
    println!("  spoofed org_b refused without a signature:   {spoof_rejected}");
    if cosign_ok && spoof_rejected {
        println!("\nOK: Org A co-signs the exact bytes only for the authenticated counterparty.");
        Ok(())
    } else {
        Err("bilateral co-sign invariant violated".into())
    }
}

//! Example: lane e (revocation catch-up) end-to-end over loopback iroh-blobs.
//!
//! A revocation authority publishes signed epoch roots as content-addressed
//! blobs. A follower fetches a manifest range over iroh-blobs and verifies every
//! fetched root against a PINNED signer key. iroh-blobs gives BLAKE3 verified
//! streaming (integrity: the bytes match the requested hash) but NOT authenticity;
//! so each fetched blob is re-checked against the pinned verifying key before it
//! is trusted. A legitimately signed root is accepted; a FORGED root - one that
//! claims the same `signer_id` but is signed by a different key - is rejected
//! `BadSignature` and yields no history (all-or-nothing).
//!
//! This example demonstrates the iroh-blobs catch-up substrate specifically (the
//! bulk half of lane b). ADAPTER-SPEC pairs the two revocation transports: the
//! direct `RevocationHandler` push for real-time roots, and iroh-blobs
//! content-addressed range catch-up for backfilling a follower that has fallen
//! behind. Both share the SAME pinned-signer authenticity check, so this demo
//! exercises that check over the catch-up path; the direct-push path is covered by
//! the crate's `real_handler_*_over_quic` tests.
//!
//! Run: `cargo run -p chio-federation-transport-iroh --example revocation_catchup`

use std::error::Error;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;

use chio_core_types::canonical_json_bytes;
use chio_core_types::sha256_hex;
use chio_core_types::Keypair;
use chio_federation_transport_iroh::admission::DirectoryGate;
use chio_federation_transport_iroh::catchup::BlobCatchupClient;
use chio_federation_transport_iroh::catchup::CatchupError;
use chio_federation_transport_iroh::identity::revocation_signer_endorsement_preimage;
use chio_federation_transport_iroh::identity::transport_endorsement_preimage;
use chio_federation_transport_iroh::identity::RevocationSignerEntry;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleBody;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryBundleTrust;
use chio_federation_transport_iroh::identity::TransportDirectoryDocument;
use chio_federation_transport_iroh::identity::TransportDirectoryEntry;
use chio_federation_transport_iroh::identity::TrustedTransportDirectoryIssuer;
use chio_federation_transport_iroh::identity::VerifiedDirectory;
use chio_federation_transport_iroh::identity::TRANSPORT_DIRECTORY_BUNDLE_SCHEMA;
use chio_revocation_oracle::Ed25519RootSigner;
use chio_revocation_oracle::EpochRoot;
use chio_revocation_oracle::SignedEpochRoot;
use iroh::address_lookup::memory::MemoryLookup;
use iroh::endpoint::presets;
use iroh::protocol::Router;
use iroh::Endpoint;
use iroh::EndpointAddr;
use iroh::EndpointId;
use iroh::RelayMode;
use iroh::SecretKey;
use iroh_blobs::Hash;

const NOW: u64 = 2_000_000;
const SIGNER_ID: &str = "oracle-a";
// 64-hex ed25519 seeds: the pinned key vs. a forger's key.
const PINNED_SEED: &str = "0101010101010101010101010101010101010101010101010101010101010101";
const FORGER_SEED: &str = "0202020202020202020202020202020202020202020202020202020202020202";

const AUTHORITY_SEED: u8 = 41;
const FOLLOWER_SEED: u8 = 40;
const FOLLOWER_KERNEL: &str = "did:chio:follower";
const AUTHORITY_KERNEL: &str = "did:chio:revocation-authority";

fn endpoint_id(seed: u8) -> EndpointId {
    SecretKey::from_bytes(&[seed; 32]).public()
}

/// A plain admitted operator entry (no revocation signers), passport-endorsed
/// over its transport endpoint.
fn admitted_entry(
    kernel_id: &str,
    passport_seed: u8,
    transport_seed: u8,
) -> TransportDirectoryEntry {
    let passport = Keypair::from_seed(&[passport_seed; 32]);
    let transport = endpoint_id(transport_seed);
    TransportDirectoryEntry {
        kernel_id: kernel_id.to_string(),
        passport_public_key: passport.public_key(),
        transport_endpoint_id: transport,
        passport_endorsement: passport.sign(&transport_endorsement_preimage(kernel_id, &transport)),
        revocation_signers: Vec::new(),
        removed: false,
    }
}

/// Build the ONE issuer-signed, load-time-verified transport directory. It admits
/// the authority (declaring oracle signer `SIGNER_ID` bound - structurally - to the
/// authority's transport endpoint via the `pinned` key) AND the follower (so the
/// follower's blob fetch is not 403-rejected at the authority's admission gate).
///
/// Both the authority's gate and the follower's catch-up client are derived from
/// THIS directory, so the catch-up signer binding inherits the issuer signature,
/// body-hash pin, validity window, and anti-rollback anchoring (no free-standing
/// signer map: the `from_bindings` path is test-only).
fn build_directory(pinned: &Ed25519RootSigner) -> Result<Arc<VerifiedDirectory>, Box<dyn Error>> {
    let issuer = Keypair::from_seed(&[240u8; 32]);

    // Authority entry carrying the derived oracle-signer binding: its passport
    // endorses the pinned oracle key under the revocation-signer domain tag.
    let authority_passport = Keypair::from_seed(&[3u8; 32]);
    let authority_transport = endpoint_id(AUTHORITY_SEED);
    let oracle_public_key = pinned.public_key();
    let authority_entry = TransportDirectoryEntry {
        kernel_id: AUTHORITY_KERNEL.to_string(),
        passport_public_key: authority_passport.public_key(),
        transport_endpoint_id: authority_transport,
        passport_endorsement: authority_passport.sign(&transport_endorsement_preimage(
            AUTHORITY_KERNEL,
            &authority_transport,
        )),
        revocation_signers: vec![RevocationSignerEntry {
            signer_id: SIGNER_ID.to_string(),
            oracle_public_key: oracle_public_key.clone(),
            oracle_endorsement: authority_passport.sign(&revocation_signer_endorsement_preimage(
                AUTHORITY_KERNEL,
                SIGNER_ID,
                &oracle_public_key,
            )),
        }],
        removed: false,
    };

    let directory = TransportDirectoryDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        local_kernel_id: AUTHORITY_KERNEL.to_string(),
        peers: vec![
            authority_entry,
            admitted_entry(FOLLOWER_KERNEL, 1, FOLLOWER_SEED),
        ],
        treaties: Vec::new(),
    };
    let directory_sha256 = sha256_hex(&canonical_json_bytes(&directory)?);
    let body = TransportDirectoryBundleBody {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        issuer: "did:chio:issuer".to_string(),
        key_id: "issuer-key-1".to_string(),
        directory_sha256,
        version: 1,
        previous_version_sha256: None,
        issued_at_unix_ms: NOW - 1,
        expires_at_unix_ms: NOW + 1,
    };
    let (signature, _) = issuer.sign_canonical(&body)?;
    let bundle = TransportDirectoryBundleDocument {
        schema: TRANSPORT_DIRECTORY_BUNDLE_SCHEMA.to_string(),
        body,
        directory,
        signature,
    };
    let trust = TransportDirectoryBundleTrust {
        issuers: vec![TrustedTransportDirectoryIssuer {
            issuer: "did:chio:issuer".to_string(),
            key_id: "issuer-key-1".to_string(),
            public_key: issuer.public_key(),
        }],
        version_floor: 0,
        expected_previous_version_sha256: None,
        now_unix_ms: NOW,
    };
    Ok(Arc::new(bundle.verify_bundle(&trust)?))
}

async fn bind_endpoint(
    seed: u8,
    gate: Option<DirectoryGate>,
    lookup: Option<MemoryLookup>,
) -> Result<Endpoint, Box<dyn Error>> {
    let mut builder = Endpoint::builder(presets::Minimal)
        .secret_key(SecretKey::from_bytes(&[seed; 32]))
        .relay_mode(RelayMode::Disabled)
        .bind_addr((Ipv4Addr::LOCALHOST, 0))
        .map_err(|error| error.to_string())?;
    if let Some(gate) = gate {
        builder = builder.hooks(gate);
    }
    if let Some(lookup) = lookup {
        builder = builder.address_lookup(lookup);
    }
    Ok(builder.bind().await.map_err(|error| error.to_string())?)
}

fn direct_addr(endpoint: &Endpoint) -> Result<EndpointAddr, Box<dyn Error>> {
    let socket = endpoint
        .bound_sockets()
        .into_iter()
        .next()
        .ok_or("endpoint bound no socket")?;
    Ok(EndpointAddr::new(endpoint.id()).with_ip_addr(socket))
}

fn signed_root(signer: &Ed25519RootSigner, epoch: u64) -> Result<SignedEpochRoot, Box<dyn Error>> {
    let root = EpochRoot {
        epoch,
        root_hash: [u8::try_from(epoch % 256).unwrap_or(0); 32],
        leaf_count: usize::try_from(epoch).unwrap_or(0),
        issued_at_unix_ms: 1_700_000_000_000 + epoch,
    };
    Ok(SignedEpochRoot::sign(root, signer)?)
}

/// A unique scratch directory for a durable blob store (cleaned up on exit).
fn scratch_dir(label: &str) -> Result<std::path::PathBuf, Box<dyn Error>> {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let mut dir = std::env::temp_dir();
    dir.push(format!(
        "chio-revocation-catchup-{label}-{}-{nanos}",
        std::process::id()
    ));
    Ok(dir)
}

async fn fetch_with_timeout(
    client: &BlobCatchupClient,
    authority: EndpointId,
    manifest: &[(u64, Hash)],
) -> Result<Result<Vec<SignedEpochRoot>, CatchupError>, String> {
    tokio::time::timeout(
        Duration::from_secs(20),
        client.fetch_range(SIGNER_ID, authority, manifest),
    )
    .await
    .map_err(|_elapsed| "fetch timed out".to_string())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    println!("== lane e: revocation catch-up over iroh-blobs (pinned-signer authenticity) ==\n");

    let pinned = Ed25519RootSigner::from_signing_key(SIGNER_ID, PINNED_SEED)?;
    let forger = Ed25519RootSigner::from_signing_key(SIGNER_ID, FORGER_SEED)?;

    // The ONE issuer-signed transport directory. It binds signer "oracle-a" -
    // structurally - to the AUTHORITY endpoint (the origin the follower may pull
    // from) and to the PINNED verifying key, and admits the follower. Both the
    // authority's admission gate and the follower's catch-up client derive from
    // this same directory, so the signer binding inherits the issuer signature and
    // anti-rollback anchoring rather than being a free-standing `from_bindings` map.
    let directory = build_directory(&pinned)?;

    // ---- Authority: gated endpoint + iroh-blobs provider --------------------
    let gate = DirectoryGate::new(directory.clone());
    let authority_endpoint = bind_endpoint(AUTHORITY_SEED, Some(gate), None).await?;
    let authority_dir = scratch_dir("authority")?;
    let authority_client = BlobCatchupClient::load(
        &authority_dir,
        authority_endpoint.clone(),
        directory.clone(),
    )
    .await?;
    let authority_addr = direct_addr(&authority_endpoint)?;
    let authority_id = endpoint_id(AUTHORITY_SEED);
    let router = Router::builder(authority_endpoint)
        .accept(iroh_blobs::ALPN, authority_client.blobs_protocol())
        .spawn();

    // Publish an epoch-5 root under the pinned key, and a FORGED epoch-6 root.
    let good_hash = authority_client
        .publish_signed_root(&signed_root(&pinned, 5)?)
        .await?;
    let forged_hash = authority_client
        .publish_signed_root(&signed_root(&forger, 6)?)
        .await?;
    println!("authority published:");
    println!("  epoch 5 (pinned key)  -> blob {good_hash}");
    println!("  epoch 6 (forger key)  -> blob {forged_hash}");

    // ---- Follower: fetch + verify against the pinned signer -----------------
    let lookup = MemoryLookup::new();
    lookup.add_endpoint_info(authority_addr);
    let follower_endpoint = bind_endpoint(FOLLOWER_SEED, None, Some(lookup)).await?;
    let follower_dir = scratch_dir("follower")?;
    let follower_client =
        BlobCatchupClient::load(&follower_dir, follower_endpoint, directory).await?;

    println!("\nfollower fetches epoch 5 (pinned-signer manifest):");
    let good_ok = match fetch_with_timeout(&follower_client, authority_id, &[(5, good_hash)]).await
    {
        Ok(Ok(roots)) => {
            let epochs: Vec<u64> = roots.iter().map(|r| r.root.epoch).collect();
            println!("  -> verified + caught up epochs {epochs:?}");
            epochs == vec![5]
        }
        Ok(Err(error)) => {
            println!("  -> unexpected verification failure: {error}");
            false
        }
        Err(error) => {
            println!("  -> {error}");
            false
        }
    };

    println!("\nfollower fetches epoch 6 (FORGED root under the same signer_id):");
    let forged_rejected =
        match fetch_with_timeout(&follower_client, authority_id, &[(6, forged_hash)]).await {
            Ok(Ok(roots)) => {
                println!("  -> UNEXPECTEDLY accepted {} root(s)", roots.len());
                false
            }
            Ok(Err(CatchupError::BadSignature(signer))) => {
                println!("  -> rejected BadSignature(signer_id={signer:?}), nothing caught up");
                signer == SIGNER_ID
            }
            Ok(Err(other)) => {
                println!("  -> rejected ({other}) - fail-closed, but not the signature check");
                false
            }
            Err(error) => {
                println!("  -> {error}");
                false
            }
        };

    router.shutdown().await.ok();
    let _ = std::fs::remove_dir_all(&authority_dir);
    let _ = std::fs::remove_dir_all(&follower_dir);

    println!("\nfail-closed summary:");
    println!("  pinned-signer root fetched + verified: {good_ok}");
    println!("  forged root rejected on authenticity:  {forged_rejected}");
    if good_ok && forged_rejected {
        println!(
            "\nOK: BLAKE3 integrity is not authenticity; only the pinned-signer root is trusted."
        );
        Ok(())
    } else {
        Err("revocation catch-up invariant violated".into())
    }
}