//! The identity layer: between the node and its transports, whose identities are
//! addresses (Mycelium's public keys), so that the rest of the node deals in node IDs
//! (`crate::identity`) only, whatever transport reaches a peer. A node may have several
//! transports; a peer is reached on one both have, so nodes on several transports
//! carry the traffic between the others.
//!
//! A node proves who it is to an address with WHO (which asks back) or ME (the answer):
//! its public node key, and a signature over its own address and the recipient's
//! (`signed`). The transport authenticates the sender's address, so a proof is good for
//! that pair of addresses only: replayed from another address, to another recipient or
//! on another transport, it fails. Once a node proved itself at an address, messages
//! from there are its.
//!
//! * A message from an address no node proved itself at is dropped, and the address is
//!   asked WHO (at most every `ASK_INTERVAL`): after a restart, the first message from
//!   a peer brings its proof back, and the peer's node retries what was lost.
//! * A seed (`resolve_seed`) is an address, or something one of the transports resolves
//!   to one; it is asked WHO, and its ME says which node is there.
//! * Where other nodes are comes from the member list and HELLO_REPLYs
//!   (`learn_addresses`): hints, not proofs. A message to a node goes to an address it
//!   proved itself at (on the first transport in the configured order that has one);
//!   with only hints, those are asked WHO, and the send is refused until one answers
//!   (the node retries refused sends).
//!
//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`, or
//! `<scheme>:<location>` where a transport needs more than an identity to reach a node
//! (`Transport::locate`, `Transport::locations`); the books keep the identities.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use tokio::sync::{mpsc, oneshot};
use tokio::time::Instant;
use tracing::{debug, info};

use crate::identity::{self, NodeKey};
use crate::locks::MutexExt;
use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
use crate::transport::traffic::Traffic;
use crate::transport::{
    Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind,
    ParsedMessage, Transport,
};

pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1";
pub(crate) const ME_LABEL: &[u8] = b"continuum/me/1";
/// An address is asked WHO at most this often, unless a seed is being resolved.
const ASK_INTERVAL: Duration = Duration::from_secs(5);
/// How long resolving a seed waits for its ME.
const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
/// Addresses kept per node from hints: about one per transport.
const MAX_HINTS: usize = 8;
/// Addresses with a proven node, nodes with hints, and addresses asked, each at most:
/// anyone can send us proofs (keys are free), so the books are bounded.
const MAX_ENTRIES: usize = 4096;

/// A send refused because the peer hasn't proven where it is yet (`send_encoded`):
/// expected when a peer known only from hints is first contacted, and retried.
#[derive(Debug, thiserror::Error)]
#[error("{0} hasn't proven where it is yet; asked where it is said to be")]
pub(crate) struct NotProven(String);

/// Whether `error` is (or was caused by) a [`NotProven`].
pub(crate) fn is_not_proven(error: &anyhow::Error) -> bool {
    error.chain().any(|cause| cause.is::<NotProven>())
}

/// What a node under the identity layer needs (`node::NodeConfig::identity`).
#[derive(Clone)]
pub(crate) struct IdentityConfig {
    pub(crate) key: Arc<NodeKey>,
    /// The address scheme of the node's transport (`NodeConfig::transport`), e.g.
    /// `mycelium`.
    pub(crate) scheme: &'static str,
    /// Further transports, by scheme, after the first in preference.
    pub(crate) more: Vec<(&'static str, Arc<dyn Transport>)>,
}

/// The bytes a WHO (`label`: `WHO_LABEL`) or ME proof signs: the label, then the
/// sender's and the recipient's address, each with its length.
pub(crate) fn signed(label: &[u8], from: &str, to: &str) -> Vec<u8> {
    let mut bytes = label.to_vec();
    for address in [from, to] {
        bytes.extend_from_slice(&(address.len() as u32).to_be_bytes());
        bytes.extend_from_slice(address.as_bytes());
    }
    bytes
}

pub(crate) struct IdentityTransport {
    shared: Arc<Shared>,
}

/// One of the node's transports.
struct Link {
    scheme: &'static str,
    inner: Arc<dyn Transport>,
}

impl Link {
    fn address(&self, target: &str) -> String {
        format!("{}:{target}", self.scheme)
    }

    fn own_address(&self) -> String {
        self.address(self.inner.local_id())
    }
}

struct Shared {
    /// In order of preference.
    links: Vec<Link>,
    key: Arc<NodeKey>,
    node_id: String,
    /// WHO and ME are counted here, under the address (`traffic`).
    traffic: Arc<Traffic>,
    // A std mutex: every access is short and never spans an await.
    book: Mutex<Book>,
}

/// Who is where, by address.
#[derive(Default)]
struct Book {
    /// The node that proved itself at an address, and when.
    at: HashMap<String, (String, Instant)>,
    /// The addresses each node last proved itself at, one per scheme.
    proven: HashMap<String, Vec<String>>,
    /// Where others say a node is.
    hints: HashMap<String, Vec<String>>,
    /// When each address was last asked WHO.
    asked: HashMap<String, Instant>,
    /// Seeds being resolved: who waits for a ME from where.
    waiting: HashMap<String, Vec<oneshot::Sender<String>>>,
    /// Since start (`stats`).
    counts: IdentityCounts,
}

/// The scheme of `address`.
fn scheme_of(address: &str) -> &str {
    address.split_once(':').map_or("", |(scheme, _)| scheme)
}

impl Book {
    /// `node` proved itself at `address`. Returns whether that is news.
    fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool {
        let news = self.at.get(address).is_none_or(|(known, _)| known != node);
        // One address per node and scheme, and one node per address: a node that
        // moved, or an address taken over by another node, leaves no stale entry.
        let scheme = scheme_of(address);
        let addresses = self.proven.entry(node.to_owned()).or_default();
        if let Some(slot) = addresses.iter_mut().find(|at| scheme_of(at) == scheme) {
            if *slot != address {
                let previous = std::mem::replace(slot, address.to_owned());
                self.at.remove(&previous);
                self.counts.address_changes += 1;
            }
        } else {
            addresses.push(address.to_owned());
        }
        if let Some((previous, _)) = self.at.insert(address.to_owned(), (node.to_owned(), now))
            && previous != node
        {
            self.forget_proof(&previous, address);
        }
        if self.at.len() > MAX_ENTRIES
            && let Some(oldest) = self
                .at
                .iter()
                .min_by_key(|(_, (_, at))| *at)
                .map(|(address, _)| address.clone())
            && let Some((node, _)) = self.at.remove(&oldest)
        {
            self.forget_proof(&node, &oldest);
        }
        for waiter in self.waiting.remove(address).into_iter().flatten() {
            let _ = waiter.send(node.to_owned());
        }
        news
    }

    fn forget_proof(&mut self, node: &str, address: &str) {
        if let Some(addresses) = self.proven.get_mut(node) {
            addresses.retain(|at| at != address);
            if addresses.is_empty() {
                self.proven.remove(node);
            }
        }
    }

    /// What the books hold now, and the counts.
    fn stats(&self) -> IdentityStats {
        let mut proven = std::collections::BTreeMap::new();
        for addresses in self.proven.values() {
            for address in addresses {
                *proven.entry(scheme_of(address).to_owned()).or_insert(0) += 1;
            }
        }
        IdentityStats {
            proven,
            hinted: self
                .hints
                .keys()
                .filter(|node| !self.proven.contains_key(*node))
                .count(),
            counts: self.counts,
        }
    }

    /// Whether to ask `address` WHO now (always, if `force`), noting that it was asked.
    fn ask(&mut self, address: &str, now: Instant, force: bool) -> bool {
        if !force
            && self
                .asked
                .get(address)
                .is_some_and(|at| now.duration_since(*at) < ASK_INTERVAL)
        {
            return false;
        }
        if self.asked.len() >= MAX_ENTRIES {
            self.asked
                .retain(|_, at| now.duration_since(*at) < ASK_INTERVAL);
        }
        self.asked.insert(address.to_owned(), now);
        true
    }
}

impl IdentityTransport {
    /// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
    pub(crate) fn new(
        transport: Arc<dyn Transport>,
        config: IdentityConfig,
        traffic: Arc<Traffic>,
    ) -> Self {
        let node_id = config.key.id();
        let links = std::iter::once((config.scheme, transport))
            .chain(config.more)
            .map(|(scheme, inner)| Link { scheme, inner })
            .collect();
        Self {
            shared: Arc::new(Shared {
                links,
                key: config.key,
                node_id,
                traffic,
                book: Mutex::default(),
            }),
        }
    }
}

impl Shared {
    fn lock(&self) -> MutexGuard<'_, Book> {
        self.book.lock_or_recover()
    }

    /// The transport `address` is on, and the transport identity at it
    /// (`Transport::locate`, which notes where that is if the address says); None if no
    /// transport of ours has its scheme, or it isn't one of that transport's.
    fn split(&self, address: &str) -> Option<(&Link, String)> {
        let (scheme, target) = address.split_once(':')?;
        let link = self.links.iter().find(|link| link.scheme == scheme)?;
        let target = link.inner.locate(target)?;
        Some((link, target))
    }

    /// `address` as the books keep it: `<scheme>:<transport identity>`.
    fn canonical(&self, address: &str) -> Option<String> {
        self.split(address)
            .map(|(link, target)| link.address(&target))
    }

    /// Where the node at `address` (as the books keep it) is, as others can use it
    /// (`Transport::locations`).
    fn locations(&self, address: &str) -> Vec<String> {
        let Some((link, target)) = self.split(address) else {
            return Vec::new();
        };
        link.inner
            .locations(&target)
            .iter()
            .map(|location| link.address(location))
            .collect()
    }

    /// Where `address`'s transport is in the order of preference (none of ours: last).
    fn preference(&self, address: &str) -> usize {
        let scheme = scheme_of(address);
        self.links
            .iter()
            .position(|link| link.scheme == scheme)
            .unwrap_or(usize::MAX)
    }

    fn is_own(&self, address: &str) -> bool {
        self.links.iter().any(|link| link.own_address() == address)
    }

    fn proof(&self, label: &[u8], link: &Link, target: &str) -> IdentityProof {
        let message = signed(label, &link.own_address(), &link.address(target));
        IdentityProof {
            key: self.key.public(),
            signature: self.key.sign(&message),
        }
    }

    /// Notes the node `proof` proves to be at `address` (on `link`); None if it proves
    /// nothing.
    fn prove(
        &self,
        link: &Link,
        address: &str,
        label: &[u8],
        proof: &IdentityProof,
    ) -> Option<String> {
        let message = signed(label, address, &link.own_address());
        if !identity::verify(&proof.key, &message, &proof.signature) {
            self.lock().counts.proofs_rejected += 1;
            debug!(from = %address, "Ignoring an identity proof that doesn't hold");
            return None;
        }
        let node = identity::node_id_of(&proof.key);
        if node == self.node_id {
            debug!(from = %address, "Ignoring a proof of our own node key");
            return None;
        }
        if self.lock().bind(address, &node, Instant::now()) {
            debug!(node = %node, address = %address, "Node proved its address");
        }
        Some(node)
    }

    /// Asks `address` WHO, unless it was asked recently (`force`: anyway).
    fn ask(self: &Arc<Self>, address: &str, force: bool) {
        if !self.lock().ask(address, Instant::now(), force) {
            return;
        }
        let shared = Arc::clone(self);
        let address = address.to_owned();
        tokio::spawn(async move { shared.send_proof(MessageKind::Who, &address).await });
    }

    /// Sends a WHO or ME (`kind`) to `address`.
    async fn send_proof(&self, kind: MessageKind, address: &str) {
        let Some((link, target)) = self.split(address) else {
            return;
        };
        let message = if kind == MessageKind::Who {
            ParsedMessage::Who(self.proof(WHO_LABEL, link, &target))
        } else {
            ParsedMessage::Me(self.proof(ME_LABEL, link, &target))
        };
        let encoded = match link.inner.encode(&target, &message) {
            Ok(encoded) => encoded,
            Err(error) => {
                debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof");
                return;
            }
        };
        match link.inner.send_encoded(&encoded).await {
            Ok(()) => self
                .traffic
                .sent(address, encoded.kind, encoded.payload_len),
            Err(error) => {
                self.traffic
                    .refused(address, encoded.kind, encoded.payload_len);
                debug!(to = %address, error = %format!("{error:#}"), "Couldn't send an identity proof");
            }
        }
    }

    /// A message from `sender` on `links[index]`.
    async fn receive(
        self: Arc<Self>,
        index: usize,
        sender: String,
        message: ParsedMessage,
        payload_len: usize,
        handler: &HandlerFn,
    ) -> Result<()> {
        let link = &self.links[index];
        let address = link.address(&sender);
        match message {
            ParsedMessage::Who(proof) => {
                self.traffic
                    .received(&address, MessageKind::Who, payload_len);
                if self.prove(link, &address, WHO_LABEL, &proof).is_some() {
                    let shared = Arc::clone(&self);
                    tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await });
                }
                Ok(())
            }
            ParsedMessage::Me(proof) => {
                self.traffic
                    .received(&address, MessageKind::Me, payload_len);
                self.prove(link, &address, ME_LABEL, &proof);
                Ok(())
            }
            message => {
                let node = self.lock().at.get(&address).map(|(node, _)| node.clone());
                match node {
                    Some(node) => handler(node, message, payload_len).await,
                    None => {
                        self.lock().counts.dropped_unproven += 1;
                        debug!(
                            from = %address,
                            kind = ?message.kind(),
                            "Dropping a message from an address no node proved itself at; asking who is there"
                        );
                        self.ask(&address, false);
                        Ok(())
                    }
                }
            }
        }
    }

    /// Where to send to `peer`: an address it proved itself at, else one it is said to
    /// be at; the most preferred transport's first.
    fn destination(&self, peer: &str) -> Option<String> {
        let book = self.lock();
        let pick = |addresses: Option<&Vec<String>>| {
            addresses?
                .iter()
                .filter(|address| self.split(address).is_some())
                .min_by_key(|address| self.preference(address))
                .cloned()
        };
        pick(book.proven.get(peer)).or_else(|| pick(book.hints.get(peer)))
    }
}

/// The first of the `results` to arrive, from tasks that run until they end.
async fn first<T: Send + 'static>(
    tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>,
) -> Option<T> {
    let (sender, mut receiver) = mpsc::channel(1);
    for task in tasks {
        let sender = sender.clone();
        tokio::spawn(async move {
            let _ = sender.send(task.await).await;
        });
    }
    drop(sender);
    receiver.recv().await
}

#[async_trait]
impl Transport for IdentityTransport {
    fn local_id(&self) -> &str {
        &self.shared.node_id
    }

    fn is_peer_id(&self, id: &str) -> bool {
        identity::is_node_id(id)
    }

    /// The longest of the transports': until then, something may still be on its way.
    fn delivery_limit(&self) -> Duration {
        self.shared
            .links
            .iter()
            .map(|link| link.inner.delivery_limit())
            .max()
            .unwrap_or_default()
    }

    /// Each transport's (`Transport::local_addresses`), with its scheme.
    fn local_addresses(&self) -> Vec<String> {
        self.shared
            .links
            .iter()
            .flat_map(|link| {
                link.inner
                    .local_addresses()
                    .into_iter()
                    .map(|target| link.address(&target))
            })
            .collect()
    }

    /// Where `peer` proved itself, else where it is said to be, as others can use it.
    fn addresses_of(&self, peer: &str) -> Vec<String> {
        let addresses = {
            let book = self.shared.lock();
            book.proven
                .get(peer)
                .or_else(|| book.hints.get(peer))
                .cloned()
                .unwrap_or_default()
        };
        let mut locations = Vec::new();
        for address in addresses {
            for location in self.shared.locations(&address) {
                if !locations.contains(&location) {
                    locations.push(location);
                }
            }
        }
        locations
    }

    fn learn_addresses(&self, peer: &str, addresses: &[String]) {
        if peer == self.shared.node_id || !identity::is_node_id(peer) {
            return;
        }
        let mut known: Vec<String> = Vec::new();
        for address in addresses {
            if let Some(address) = self.shared.canonical(address)
                && !self.shared.is_own(&address)
                && !known.contains(&address)
            {
                known.push(address);
            }
        }
        known.truncate(MAX_HINTS);
        if known.is_empty() {
            return;
        }
        let mut book = self.shared.lock();
        if book.hints.len() < MAX_ENTRIES || book.hints.contains_key(peer) {
            book.hints.insert(peer.to_owned(), known);
        }
    }

    fn can_reach(&self, peer: &str) -> bool {
        self.shared.destination(peer).is_some()
    }

    fn transports(&self) -> Vec<String> {
        self.shared
            .links
            .iter()
            .map(|link| link.scheme.to_owned())
            .collect()
    }

    /// The address `encode` picks, and the transport's connection there.
    fn destination(&self, peer: &str) -> Option<Destination> {
        let address = self.shared.destination(peer)?;
        let proven = self
            .shared
            .lock()
            .proven
            .get(peer)
            .is_some_and(|addresses| addresses.contains(&address));
        let connection = self
            .shared
            .split(&address)
            .and_then(|(link, target)| link.inner.connection(&target));
        Some(Destination {
            address,
            proven,
            connection,
        })
    }

    /// The identity layer's, and each transport's.
    fn stats(&self) -> TransportStats {
        let mut stats = TransportStats {
            identity: Some(self.shared.lock().stats()),
            ..TransportStats::default()
        };
        for link in &self.shared.links {
            stats.merge(link.inner.stats());
        }
        stats
    }

    /// Each transport's, with its scheme.
    fn discovered_seeds(&self) -> Vec<String> {
        self.shared
            .links
            .iter()
            .flat_map(|link| {
                link.inner
                    .discovered_seeds()
                    .into_iter()
                    .map(|seed| link.address(&seed))
            })
            .collect()
    }

    /// A node ID resolves to itself if this node knows where it is; anything else is
    /// an address (`<scheme>:...`) or what one of the transports resolves to one (the
    /// first that does), asked WHO.
    async fn resolve_seed(&self, seed: &str) -> Result<String> {
        if identity::is_node_id(seed) {
            if self.shared.destination(seed).is_some() {
                return Ok(seed.to_owned());
            }
            bail!("a node ID alone doesn't say where that node is; give its address");
        }
        // With a scheme of ours, only that transport resolves the rest (an address the
        // transport can't read as an identity, e.g. a libp2p multiaddr without its peer
        // ID, still resolves); without one, the first transport that can.
        let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
            let link = self
                .shared
                .links
                .iter()
                .find(|link| link.scheme == scheme)?;
            Some((link, rest))
        });
        let address = match self.shared.canonical(seed) {
            Some(address) => address,
            None => match scheme_and_rest {
                Some((link, rest)) => {
                    let target = link
                        .inner
                        .resolve_seed(rest)
                        .await
                        .with_context(|| format!("{}: {seed}", link.scheme))?;
                    link.address(&target)
                }
                None => {
                    let mut failures = Vec::new();
                    let mut resolved = None;
                    for link in &self.shared.links {
                        match link.inner.resolve_seed(seed).await {
                            Ok(target) => {
                                resolved = Some(link.address(&target));
                                break;
                            }
                            Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
                        }
                    }
                    resolved.ok_or_else(|| anyhow!("{}", failures.join("; ")))?
                }
            },
        };
        if self.shared.is_own(&address) {
            bail!("{seed} is this node");
        }
        let receiver = {
            let mut book = self.shared.lock();
            if let Some((node, _)) = book.at.get(&address) {
                return Ok(node.clone());
            }
            let (sender, receiver) = oneshot::channel();
            let waiting = book.waiting.entry(address.clone()).or_default();
            waiting.retain(|waiter| !waiter.is_closed());
            waiting.push(sender);
            receiver
        };
        self.shared.ask(&address, true);
        match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await {
            Ok(Ok(node)) => {
                info!(seed = %seed, node = %node, "Seed is node");
                Ok(node)
            }
            _ => {
                self.shared.lock().counts.seeds_unanswered += 1;
                Err(anyhow!(
                    "no node answered at {address} within {RESOLVE_TIMEOUT:?}"
                ))
            }
        }
    }

    /// For the address `destination` picks (a send there is refused until `peer`
    /// proved itself there, `send_encoded`).
    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
        let Some(address) = self.shared.destination(peer) else {
            bail!("no address known for {peer} on this node's transports");
        };
        let Some((link, target)) = self.shared.split(&address) else {
            bail!("no transport for {address}");
        };
        let mut encoded = link.inner.encode(&target, message)?;
        encoded.peer = peer.to_owned();
        encoded.address = address;
        Ok(encoded)
    }

    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
        let hints = {
            let mut book = self.shared.lock();
            let proven = book
                .proven
                .get(&message.peer)
                .is_some_and(|addresses| addresses.contains(&message.address));
            if proven {
                None
            } else {
                book.counts.refused_unproven += 1;
                Some(book.hints.get(&message.peer).cloned().unwrap_or_default())
            }
        };
        if let Some(hints) = hints {
            for address in hints {
                self.shared.ask(&address, false);
            }
            return Err(NotProven(message.peer.clone()).into());
        }
        let Some((link, target)) = self.shared.split(&message.address) else {
            bail!("no transport for {}", message.address);
        };
        let mut inner = message.clone();
        inner.peer = target.clone();
        inner.address = target;
        link.inner.send_encoded(&inner).await
    }

    /// Every transport's consumer; returns when the first one does.
    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
        let handler: Arc<HandlerFn> = Arc::from(handler);
        let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
            Vec::new();
        for (index, link) in self.shared.links.iter().enumerate() {
            let inner = Arc::clone(&link.inner);
            let shared = Arc::clone(&self.shared);
            let handler = Arc::clone(&handler);
            consumers.push(Box::pin(async move {
                inner
                    .start_consumer(Box::new(move |sender, message, payload_len| {
                        let shared = Arc::clone(&shared);
                        let handler = Arc::clone(&handler);
                        Box::pin(async move {
                            shared
                                .receive(index, sender, message, payload_len, &*handler)
                                .await
                        })
                    }))
                    .await
            }));
        }
        first(consumers)
            .await
            .unwrap_or_else(|| Err(anyhow!("no transports")))
    }

    /// When the first transport exits.
    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
        let waits = self
            .shared
            .links
            .iter()
            .map(|link| {
                let inner = Arc::clone(&link.inner);
                Box::pin(async move { inner.wait_for_exit().await })
                    as std::pin::Pin<Box<dyn Future<Output = _> + Send>>
            })
            .collect();
        first(waits)
            .await
            .unwrap_or_else(|| Err(std::io::Error::other("no transports")))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn signed_bytes_say_both_addresses() {
        // A length before each address: "a:bc" to "d" isn't "a:b" to "cd".
        assert_ne!(
            signed(WHO_LABEL, "a:bc", "d"),
            signed(WHO_LABEL, "a:b", "cd")
        );
        assert_ne!(signed(WHO_LABEL, "a", "b"), signed(ME_LABEL, "a", "b"));
        assert_ne!(signed(WHO_LABEL, "a", "b"), signed(WHO_LABEL, "b", "a"));
    }

    #[test]
    fn one_address_per_node_and_scheme_and_one_node_per_address() {
        let now = Instant::now();
        let mut book = Book::default();
        assert!(book.bind("s:x", "n1", now));
        assert!(!book.bind("s:x", "n1", now));
        // n1 moved to s:y: s:x is no longer n1's.
        assert!(book.bind("s:y", "n1", now));
        assert!(!book.at.contains_key("s:x"));
        assert_eq!(book.proven["n1"], ["s:y"]);
        // On another transport too.
        assert!(book.bind("t:x", "n1", now));
        assert_eq!(book.proven["n1"], ["s:y", "t:x"]);
        // n2 took over s:y: n1 is proven on t only.
        assert!(book.bind("s:y", "n2", now));
        assert_eq!(book.proven["n1"], ["t:x"]);
        assert_eq!(book.at["s:y"].0, "n2");
        // Only n1's move from s:x to s:y changed an address.
        assert_eq!(book.counts.address_changes, 1);
        let stats = book.stats();
        assert_eq!(stats.proven["s"], 1);
        assert_eq!(stats.proven["t"], 1);
    }

    #[test]
    fn asking_is_rate_limited() {
        let now = Instant::now();
        let mut book = Book::default();
        assert!(book.ask("s:x", now, false));
        assert!(!book.ask("s:x", now, false));
        assert!(book.ask("s:x", now, true));
        assert!(book.ask("s:x", now + ASK_INTERVAL, false));
    }

    #[tokio::test]
    async fn a_waiting_seed_learns_the_node() {
        let mut book = Book::default();
        let (sender, receiver) = oneshot::channel();
        book.waiting.insert("s:x".into(), vec![sender]);
        book.bind("s:x", "n1", Instant::now());
        assert_eq!(receiver.await.unwrap(), "n1");
    }
}