//! Keeps the VRPs and ASPAs in step with a validator, over the
//! RPKI-to-Router protocol (RFC 8210 and draft-ietf-sidrops-8210bis).
//!
//! The client connects to a cache (the validator's RTR server), asks for
//! everything once, and from then on for what has changed: when the cache
//! says there is something new, and otherwise every so often. Each time the
//! data changes an [`Update`] is published for whoever validates routes.
//!
//! The newest version of the protocol is asked for first, and an older one
//! if that is what the cache speaks. Only version 2 carries ASPAs: with an
//! older cache there are none, and paths are not verified.
//!
//! A cache that cannot be reached is tried again, and the other caches in
//! turn. Meanwhile the data last received is used, but not for ever: a list
//! that has not been confirmed for the expire interval is dropped, and
//! routes are then neither valid nor invalid. Using stale data would be
//! worse: it would go on refusing routes whose holders have since
//! authorised them.

use std::collections::HashSet;
use std::io;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, SystemTime};

use bgp_primitives::constants::IpVersion;
use bytes::BytesMut;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use tokio::time::{Instant, sleep_until, timeout};
use tracing::{info, warn};

use crate::aspa::AspaTable;
use crate::rtr::{Pdu, RtrError, Timers, VERSION_0, VERSION_2, error_code};
use crate::vrp::{Vrp, VrpTable};

const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// How long a cache may take to send the next part of an answer.
const RESPONSE_TIMEOUT: Duration = Duration::from_secs(60);

/// How long a whole answer may take. Everything a validator has takes it
/// seconds to send.
const ANSWER_TIMEOUT: Duration = Duration::from_secs(600);

/// The most records taken in one answer: several times what the whole
/// RPKI holds. A cache that sends more is not sending the RPKI, and
/// would otherwise be listened to until there is no memory left.
const MAX_RECORDS: usize = 8_000_000;

/// How long to wait before trying the next cache after one has failed.
const NEXT_CACHE_DELAY: Duration = Duration::from_secs(1);

/// How long a cache that is being left is given to take the reason.
const ERROR_REPORT_TIMEOUT: Duration = Duration::from_secs(1);

/// Where the VRPs come from, and how patiently.
#[derive(Debug, Clone, Default)]
pub struct RpkiConfig {
    /// The caches, in order of preference. One is used at a time; the next
    /// is tried when it fails.
    pub caches: Vec<SocketAddr>,
    /// If set, used in place of what the cache says (or of the defaults of
    /// RFC 8210 §6, with a cache that says nothing).
    pub refresh: Option<Duration>,
    pub retry: Option<Duration>,
    pub expire: Option<Duration>,
}

/// The VRPs and ASPAs as they are after a change.
#[derive(Debug, Clone)]
pub struct Update {
    /// Goes up by one with every update.
    pub version: u64,
    pub vrps: Arc<VrpTable>,
    pub aspas: Arc<AspaTable>,
    /// What was added or removed to get here from the version before.
    /// `None` if any route's validity may have changed.
    pub changed: Option<Changed>,
}

/// What differs between one [`Update`] and the next.
#[derive(Debug, Clone)]
pub struct Changed {
    /// The VRPs that came or went: only routes that these cover can have
    /// another origin validity than before.
    pub vrps: Arc<VrpTable>,
    /// The customer ASes whose ASPA came, went or lists other providers:
    /// only paths with one of these on them can verify otherwise than
    /// before.
    pub aspa_customers: Arc<HashSet<u32>>,
}

/// How things stand with the caches.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CacheState {
    /// No cache has been reached yet, or the one in use was lost.
    Connecting,
    /// Connected and waiting for the data.
    Synchronising,
    /// The data is complete as of the cache's serial.
    Synchronised,
}

/// What there is to say about the client, for showing.
#[derive(Debug, Clone)]
pub struct Status {
    /// The cache in use, or last tried.
    pub cache: Option<SocketAddr>,
    pub state: CacheState,
    /// The cache's session and the serial the data is at.
    pub session_id: Option<u16>,
    pub serial: Option<u32>,
    pub protocol_version: u8,
    pub vrps_ipv4: usize,
    pub vrps_ipv6: usize,
    /// How many ASes have an ASPA. None do if the cache speaks a version
    /// of the protocol before 2.
    pub aspas: usize,
    /// When the data was last confirmed by a cache.
    pub updated_at: Option<SystemTime>,
    /// When it will be dropped unless confirmed again.
    pub expires_at: Option<SystemTime>,
    /// Why the last attempt failed, if it did.
    pub last_error: Option<String>,
}

struct Shared {
    updates: watch::Sender<Update>,
    /// The version of the latest update, for a look that costs nothing.
    version: AtomicU64,
    status: Mutex<Status>,
}

/// Handle to the VRPs and to the client that keeps them. Cheap to clone.
#[derive(Clone)]
pub struct Rpki {
    shared: Arc<Shared>,
}

impl Rpki {
    /// Starts a client for `config` and returns the handle to its data,
    /// with the task that runs it. Until the first cache has answered
    /// there are no VRPs and no ASPAs: every route's origin is
    /// [`crate::Validity::NotFound`] and every path
    /// [`crate::PathValidity::Unknown`].
    pub fn spawn(config: RpkiConfig) -> (Self, JoinHandle<()>) {
        let (updates, _) = watch::channel(Update {
            version: 0,
            vrps: Arc::default(),
            aspas: Arc::default(),
            changed: None,
        });
        let rpki = Self {
            shared: Arc::new(Shared {
                updates,
                version: AtomicU64::new(0),
                status: Mutex::new(Status {
                    cache: None,
                    state: CacheState::Connecting,
                    session_id: None,
                    serial: None,
                    protocol_version: VERSION_2,
                    vrps_ipv4: 0,
                    vrps_ipv6: 0,
                    aspas: 0,
                    updated_at: None,
                    expires_at: None,
                    last_error: None,
                }),
            }),
        };
        let client = Client {
            shared: Arc::clone(&rpki.shared),
            config,
            data: Data::default(),
            place: None,
            protocol_version: VERSION_2,
            timers: Timers::default(),
            expires: None,
        };
        (rpki, tokio::spawn(client.run()))
    }

    /// The VRPs and ASPAs as they are now.
    pub fn current(&self) -> Update {
        self.shared.updates.borrow().clone()
    }

    /// The version of the latest [`Update`]. Costs a load of one number,
    /// for those who ask before every use whether theirs is still current.
    pub fn version(&self) -> u64 {
        self.shared.version.load(Ordering::Acquire)
    }

    /// Follows the updates. A follower that is busy sees the latest only;
    /// [`Update::version`] tells it whether it skipped any.
    pub fn subscribe(&self) -> watch::Receiver<Update> {
        self.shared.updates.subscribe()
    }

    pub fn status(&self) -> Status {
        self.shared
            .status
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }
}

/// Why a session with a cache ended.
#[derive(Debug, thiserror::Error)]
enum SessionError {
    #[error("{0}")]
    Io(#[from] io::Error),
    #[error("the cache sent something that is not RTR: {0}")]
    Malformed(#[from] RtrError),
    #[error("the cache closed the connection")]
    Closed,
    #[error("the cache did not answer within {RESPONSE_TIMEOUT:?}")]
    Timeout,
    #[error("the cache reported an error: {text} (code {code})")]
    Reported { code: u16, text: String },
    #[error("the cache sent {0} where it has no place")]
    Unexpected(&'static str),
    #[error("the cache {0}")]
    Inconsistent(Inconsistency),
    /// Not a failure of the cache: it speaks an older version, and is to
    /// be asked again in that.
    #[error("the cache speaks protocol version {0}")]
    Downgrade(u8),
}

impl SessionError {
    /// The error code that tells the cache why it is being left, if that
    /// is something it did and can be told (RFC 8210 §10). An Error Report
    /// is never answered with another.
    fn report_code(&self) -> Option<u16> {
        match self {
            Self::Malformed(e) => Some(e.code()),
            Self::Unexpected(_) => Some(error_code::CORRUPT_DATA),
            Self::Inconsistent(problem) => Some(problem.code),
            Self::Io(_) | Self::Closed | Self::Timeout | Self::Reported { .. } => None,
            Self::Downgrade(_) => None,
        }
    }
}

/// Something a cache said that does not fit with what it said before.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Inconsistency {
    what: &'static str,
    /// The error code it is reported to the cache with.
    code: u16,
}

impl std::fmt::Display for Inconsistency {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.what)
    }
}

const fn inconsistent(what: &'static str, code: u16) -> Inconsistency {
    Inconsistency { what, code }
}

const DUPLICATE_VRP: Inconsistency =
    inconsistent("announced a VRP twice", error_code::DUPLICATE_ANNOUNCEMENT);
const UNKNOWN_VRP: Inconsistency = inconsistent(
    "withdrew a VRP it had not announced",
    error_code::WITHDRAWAL_OF_UNKNOWN_RECORD,
);
const UNKNOWN_ASPA: Inconsistency = inconsistent(
    "withdrew an ASPA it had not announced",
    error_code::WITHDRAWAL_OF_UNKNOWN_RECORD,
);
const OTHER_SESSION: Inconsistency = inconsistent(
    "ended an answer in another session than it began it",
    error_code::CORRUPT_DATA,
);
const OTHER_VERSION: Inconsistency = inconsistent(
    "changed protocol version in the middle of a session",
    error_code::UNEXPECTED_VERSION,
);
const INTERNAL: Inconsistency = inconsistent(
    "could not be followed (a bug here)",
    error_code::INTERNAL_ERROR,
);

/// Where the data is at: which of the cache's sessions, and which serial.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Place {
    cache: SocketAddr,
    /// The version of the protocol the data came in. What another version
    /// carries is not the same data.
    protocol_version: u8,
    session_id: u16,
    serial: u32,
}

/// What a cache has given: the data validation rests on.
#[derive(Debug, Clone, Default)]
struct Data {
    vrps: HashSet<Vrp>,
    aspas: AspaTable,
}

/// One thing an answer says: a VRP or an ASPA, come or gone.
#[derive(Debug, Clone)]
enum Record {
    Vrp {
        announce: bool,
        vrp: Vrp,
    },
    /// With providers, the ASPA that is now `customer`'s; without, that
    /// it has none any more.
    Aspa {
        customer: u32,
        providers: Option<Vec<u32>>,
    },
}

struct Client {
    shared: Arc<Shared>,
    config: RpkiConfig,
    /// The data as last received.
    data: Data,
    /// Where `data` is at, if it came from a cache that can be asked for
    /// what has changed since.
    place: Option<Place>,
    protocol_version: u8,
    /// As the cache last gave them, or the defaults.
    timers: Timers,
    /// When `data` is to be dropped unless a cache confirms it first.
    expires: Option<Instant>,
}

/// A connection to a cache, read a PDU at a time.
struct Connection {
    stream: TcpStream,
    buf: BytesMut,
}

impl Connection {
    async fn send(&mut self, version: u8, pdu: &Pdu) -> io::Result<()> {
        let mut out = BytesMut::new();
        pdu.encode(version, &mut out);
        self.stream.write_all(&out).await
    }

    /// The next PDU, and the protocol version it was sent with.
    /// Cancel-safe: what has arrived of a PDU stays in the buffer.
    async fn next(&mut self) -> Result<(u8, Pdu), SessionError> {
        loop {
            if let Some((version, pdu, used)) = Pdu::decode(&self.buf)? {
                let _ = self.buf.split_to(used);
                return Ok((version, pdu));
            }
            if self.stream.read_buf(&mut self.buf).await? == 0 {
                return Err(SessionError::Closed);
            }
        }
    }
}

impl Client {
    fn status(&self, change: impl FnOnce(&mut Status)) {
        change(
            &mut self
                .shared
                .status
                .lock()
                .unwrap_or_else(PoisonError::into_inner),
        );
    }

    fn refresh(&self) -> Duration {
        self.config.refresh.unwrap_or(Duration::from_secs(
            self.timers.refresh.clamp(1, 86400).into(),
        ))
    }

    fn retry(&self) -> Duration {
        self.config
            .retry
            .unwrap_or(Duration::from_secs(self.timers.retry.clamp(1, 7200).into()))
    }

    /// Never sooner than a refresh that failed and was tried once more
    /// would take, whatever the cache or the configuration says: data that
    /// ran out between two refreshes would leave nothing validated for a
    /// part of every round (RFC 8210 §6 wants it longer than either).
    fn expire(&self) -> Duration {
        self.config
            .expire
            .unwrap_or(Duration::from_secs(
                self.timers.expire.clamp(600, 172_800).into(),
            ))
            .max(self.refresh() + self.retry())
    }

    async fn run(mut self) {
        if self.config.caches.is_empty() {
            return std::future::pending().await;
        }
        let mut attempt = 0usize;
        loop {
            let cache = self.config.caches[attempt % self.config.caches.len()];
            self.status(|status| {
                status.cache = Some(cache);
                status.state = CacheState::Connecting;
            });
            let ended = self.session(cache).await;
            if let SessionError::Downgrade(version) = ended {
                info!(%cache, version, "RPKI cache speaks an older RTR version; asking again in that");
                self.protocol_version = version;
                continue;
            }
            warn!(%cache, err = %ended, "RPKI cache session ended");
            self.status(|status| {
                status.state = CacheState::Connecting;
                status.last_error = Some(ended.to_string());
            });

            // The next cache at once; when all have been tried, a rest.
            attempt += 1;
            // The next cache, or this one when it is back, may speak a
            // newer version.
            self.protocol_version = VERSION_2;
            let wait = if attempt.is_multiple_of(self.config.caches.len()) {
                self.retry()
            } else {
                NEXT_CACHE_DELAY
            };
            self.wait(wait).await;
        }
    }

    /// Waits for `duration`, dropping the data on the way if its time runs
    /// out.
    async fn wait(&mut self, duration: Duration) {
        let until = Instant::now() + duration;
        if let Some(expires) = self.expires
            && expires <= until
        {
            sleep_until(expires).await;
            self.expired();
        }
        sleep_until(until).await;
    }

    /// The data has gone unconfirmed for too long to be used.
    fn expired(&mut self) {
        warn!(
            vrps = self.data.vrps.len(),
            aspas = self.data.aspas.len(),
            "RPKI data has expired without a cache to confirm it; dropping it"
        );
        self.data = Data::default();
        self.place = None;
        self.expires = None;
        self.publish(Arc::default(), None);
        self.status(|status| {
            status.session_id = None;
            status.serial = None;
            status.expires_at = None;
        });
    }

    /// Runs a session with `cache` until it ends, which it does only by
    /// failing.
    async fn session(&mut self, cache: SocketAddr) -> SessionError {
        let stream = match timeout(CONNECT_TIMEOUT, TcpStream::connect(cache)).await {
            Ok(Ok(stream)) => stream,
            Ok(Err(e)) => return e.into(),
            Err(_) => return SessionError::Timeout,
        };
        let mut connection = Connection {
            stream,
            buf: BytesMut::with_capacity(64 * 1024),
        };
        info!(%cache, "Connected to RPKI cache");
        self.status(|status| {
            status.state = CacheState::Synchronising;
            status.protocol_version = self.protocol_version;
        });

        loop {
            if let Err(e) = self.synchronise(cache, &mut connection).await {
                return self.leave(&mut connection, e).await;
            }
            // Until the cache says there is something new, or it is time
            // to ask anyway. If the data runs out first, it is the cache
            // that has stopped confirming it.
            let ask = Instant::now() + self.refresh();
            let expires = self.expires.unwrap_or(ask);
            tokio::select! {
                pdu = connection.next() => {
                    let problem = match pdu {
                        Ok((_, Pdu::ErrorReport { code, text })) => {
                            SessionError::Reported { code, text }
                        }
                        Ok((version, _)) if version != self.protocol_version => {
                            SessionError::Inconsistent(OTHER_VERSION)
                        }
                        Ok((_, Pdu::SerialNotify { .. })) => continue,
                        Ok(_) => SessionError::Unexpected("an answer nobody asked for"),
                        Err(e) => e,
                    };
                    return self.leave(&mut connection, problem).await;
                }
                () = sleep_until(ask) => {}
                () = sleep_until(expires), if expires < ask => {
                    self.expired();
                    return SessionError::Timeout;
                }
            }
        }
    }

    /// Tells the cache why the session is ending, if it is for something
    /// the cache did, and gives the reason back.
    async fn leave(&self, connection: &mut Connection, why: SessionError) -> SessionError {
        if let Some(code) = why.report_code() {
            let report = Pdu::ErrorReport {
                code,
                text: why.to_string(),
            };
            // If it cannot be told, it cannot.
            let _ = timeout(
                ERROR_REPORT_TIMEOUT,
                connection.send(self.protocol_version, &report),
            )
            .await;
        }
        why
    }

    /// Asks the cache for what has changed, or for everything, and takes
    /// the answer.
    async fn synchronise(
        &mut self,
        cache: SocketAddr,
        connection: &mut Connection,
    ) -> Result<(), SessionError> {
        // Changes can be asked for only of the cache the data came from,
        // in the version of the protocol it came in.
        let mut since = self.place.filter(|place| {
            place.cache == cache && place.protocol_version == self.protocol_version
        });
        loop {
            let query = match since {
                Some(place) => Pdu::SerialQuery {
                    session_id: place.session_id,
                    serial: place.serial,
                },
                None => Pdu::ResetQuery,
            };
            connection.send(self.protocol_version, &query).await?;
            match self.answer(cache, connection, since).await? {
                Answer::Complete => return Ok(()),
                // The cache cannot say what changed: ask for everything.
                Answer::Reset => since = None,
            }
        }
    }

    /// Reads the answer to a query. `since` is the place a Serial Query
    /// asked about; `None` after a Reset Query.
    async fn answer(
        &mut self,
        cache: SocketAddr,
        connection: &mut Connection,
        since: Option<Place>,
    ) -> Result<Answer, SessionError> {
        // Each PDU in good time, and all of them too: a cache that sends
        // one a minute would keep the old data in use for ever.
        let deadline = Instant::now() + ANSWER_TIMEOUT;
        let mut next = async || {
            let wait = RESPONSE_TIMEOUT.min(deadline.saturating_duration_since(Instant::now()));
            match timeout(wait, connection.next()).await {
                Ok(pdu) => pdu,
                Err(_) => Err(SessionError::Timeout),
            }
        };

        // The answer begins with a Cache Response, or is a Cache Reset.
        let session_id = loop {
            let (version, pdu) = next().await?;
            if version < self.protocol_version {
                // An older cache answers in its own version, with an error
                // or as if it had been asked in that; the query has to be
                // put again in that version (RFC 8210 §7).
                return Err(SessionError::Downgrade(version));
            }
            if version > self.protocol_version {
                return Err(SessionError::Inconsistent(OTHER_VERSION));
            }
            match pdu {
                Pdu::CacheResponse { session_id } => break session_id,
                Pdu::CacheReset if since.is_some() => return Ok(Answer::Reset),
                // News of what is being asked for anyway.
                Pdu::SerialNotify { .. } => {}
                // A cache that refuses the version in the version it
                // refuses: the next one down, then.
                Pdu::ErrorReport { code, .. }
                    if code == error_code::UNSUPPORTED_VERSION
                        && self.protocol_version != VERSION_0 =>
                {
                    return Err(SessionError::Downgrade(self.protocol_version - 1));
                }
                Pdu::ErrorReport { code, text } => {
                    return Err(SessionError::Reported { code, text });
                }
                _ => return Err(SessionError::Unexpected("data before a Cache Response")),
            }
        };
        if since.is_some_and(|place| place.session_id != session_id) {
            // The cache has started over; its serials mean something else
            // now.
            return Ok(Answer::Reset);
        }

        // Then the VRPs and ASPAs, and End of Data. Nothing takes effect
        // until the answer is complete (RFC 8210 §8.1), so what it says is
        // kept aside in the order it came.
        let mut records: Vec<Record> = Vec::new();
        let (serial, timers) = loop {
            let (version, pdu) = next().await?;
            if version != self.protocol_version && !matches!(pdu, Pdu::ErrorReport { .. }) {
                return Err(SessionError::Inconsistent(OTHER_VERSION));
            }
            if records.len() >= MAX_RECORDS {
                return Err(SessionError::Unexpected("more data than any cache has"));
            }
            match pdu {
                Pdu::Prefix { announce, vrp } => records.push(Record::Vrp { announce, vrp }),
                Pdu::Aspa {
                    announce,
                    customer,
                    providers,
                } => records.push(Record::Aspa {
                    customer,
                    providers: announce.then_some(providers),
                }),
                Pdu::RouterKey => {}
                Pdu::EndOfData {
                    session_id: ended,
                    serial,
                    timers,
                } => {
                    if ended != session_id {
                        return Err(SessionError::Inconsistent(OTHER_SESSION));
                    }
                    break (serial, timers);
                }
                Pdu::ErrorReport { code, text } => {
                    return Err(SessionError::Reported { code, text });
                }
                _ => {
                    return Err(SessionError::Unexpected(
                        "a PDU that is not part of an answer",
                    ));
                }
            }
        };

        // The first data there is can change any route's validity; after
        // that, only where something came or went.
        let first = self.place.is_none() && self.expires.is_none();
        let (before, replaced) = match since {
            // Changes, to what there is.
            Some(_) => (std::mem::take(&mut self.data), None),
            // Everything anew.
            None => (Data::default(), Some(std::mem::take(&mut self.data))),
        };
        let vrps = Arc::clone(&self.shared.updates.borrow().vrps);
        // Building the tables is work for a thread that may take its time.
        let built = tokio::task::spawn_blocking(move || apply(before, replaced, &records, vrps))
            .await
            .map_err(|_| SessionError::Inconsistent(INTERNAL))?;
        let built = match built {
            Ok(built) => built,
            Err((data, problem)) => {
                // What there was stays as it was.
                self.data = data;
                return Err(SessionError::Inconsistent(problem));
            }
        };

        self.data = built.data;
        self.place = Some(Place {
            cache,
            protocol_version: self.protocol_version,
            session_id,
            serial,
        });
        self.timers = timers.unwrap_or_default();
        self.expires = Some(Instant::now() + self.expire());
        let changes = built.changed.vrps.len() + built.changed.aspa_customers.len();
        if first || changes > 0 {
            self.publish(built.vrps, (!first).then_some(built.changed));
        }
        info!(
            %cache,
            serial,
            vrps = self.data.vrps.len(),
            aspas = self.data.aspas.len(),
            changes,
            "RPKI data is up to date"
        );
        let expires_at = SystemTime::now() + self.expire();
        self.status(|status| {
            status.state = CacheState::Synchronised;
            status.session_id = Some(session_id);
            status.serial = Some(serial);
            status.updated_at = Some(SystemTime::now());
            status.expires_at = Some(expires_at);
            status.last_error = None;
        });
        Ok(Answer::Complete)
    }

    /// Publishes the data as it now is. `vrps` holds its VRPs.
    fn publish(&self, vrps: Arc<VrpTable>, changed: Option<Changed>) {
        let count = |version| {
            self.data
                .vrps
                .iter()
                .filter(|vrp| vrp.prefix.address_family() == version)
                .count()
        };
        let (ipv4, ipv6) = (count(IpVersion::Ipv4), count(IpVersion::Ipv6));
        self.status(|status| {
            status.vrps_ipv4 = ipv4;
            status.vrps_ipv6 = ipv6;
            status.aspas = self.data.aspas.len();
        });
        let aspas = Arc::new(self.data.aspas.clone());
        self.shared.updates.send_modify(|update| {
            update.version += 1;
            update.vrps = vrps;
            update.aspas = aspas;
            update.changed = changed;
            self.shared.version.store(update.version, Ordering::Release);
        });
    }
}

/// The data after an answer, ready to be used.
struct Built {
    data: Data,
    /// The VRPs of `data`, arranged for validating.
    vrps: Arc<VrpTable>,
    changed: Changed,
}

/// Applies the `records` of a complete answer, in order, to `data`.
/// `replaced` is the data that an answer to a Reset Query takes the place
/// of, and `vrps` the table of the VRPs as they were before the answer,
/// which is kept if the answer changes none.
///
/// If the answer contradicts itself or what there was, what there was is
/// given back with the reason.
fn apply(
    mut data: Data,
    replaced: Option<Data>,
    records: &[Record],
    vrps: Arc<VrpTable>,
) -> Result<Built, (Data, Inconsistency)> {
    // What each record displaced, to put back if a later one cannot be
    // applied.
    let mut displaced: Vec<Option<Arc<[u32]>>> = Vec::new();
    let mut problem = None;
    for record in records {
        let ok = match record {
            Record::Vrp {
                announce: true,
                vrp,
            } => data.vrps.insert(*vrp).then_some(None).ok_or(DUPLICATE_VRP),
            Record::Vrp {
                announce: false,
                vrp,
            } => data.vrps.remove(vrp).then_some(None).ok_or(UNKNOWN_VRP),
            // An ASPA that is announced takes the place of the one there
            // was (draft-ietf-sidrops-8210bis §5.12).
            Record::Aspa {
                customer,
                providers: Some(providers),
            } => Ok(data.aspas.set(*customer, providers.iter().copied())),
            Record::Aspa {
                customer,
                providers: None,
            } => data.aspas.remove(*customer).map(Some).ok_or(UNKNOWN_ASPA),
        };
        match ok {
            Ok(was) => displaced.push(was),
            Err(e) => {
                problem = Some(e);
                break;
            }
        }
    }
    if let Some(problem) = problem {
        return Err(match replaced {
            Some(replaced) => (replaced, problem),
            None => {
                // Undo, last first.
                for (record, was) in records.iter().zip(displaced).rev() {
                    match record {
                        Record::Vrp {
                            announce: true,
                            vrp,
                        } => {
                            data.vrps.remove(vrp);
                        }
                        Record::Vrp {
                            announce: false,
                            vrp,
                        } => {
                            data.vrps.insert(*vrp);
                        }
                        Record::Aspa { customer, .. } => data.aspas.restore(*customer, was),
                    }
                }
                (data, problem)
            }
        });
    }

    // What came or went: against the data replaced, or else whatever the
    // answer touched.
    let (changed_vrps, aspa_customers): (Vec<Vrp>, HashSet<u32>) = match &replaced {
        Some(replaced) => (
            data.vrps
                .symmetric_difference(&replaced.vrps)
                .copied()
                .collect(),
            differing_customers(&data.aspas, &replaced.aspas),
        ),
        None => (
            records
                .iter()
                .filter_map(|record| match record {
                    Record::Vrp { vrp, .. } => Some(*vrp),
                    Record::Aspa { .. } => None,
                })
                .collect(),
            records
                .iter()
                .filter_map(|record| match record {
                    Record::Aspa { customer, .. } => Some(*customer),
                    Record::Vrp { .. } => None,
                })
                .collect(),
        ),
    };
    Ok(Built {
        vrps: match changed_vrps.is_empty() {
            true => vrps,
            false => Arc::new(VrpTable::new(data.vrps.iter().copied())),
        },
        changed: Changed {
            vrps: Arc::new(VrpTable::new(changed_vrps)),
            aspa_customers: Arc::new(aspa_customers),
        },
        data,
    })
}

/// The customers whose ASPA is not the same in `a` as in `b`.
fn differing_customers(a: &AspaTable, b: &AspaTable) -> HashSet<u32> {
    let differs = |(customer, providers): (u32, &[u32]), other: &AspaTable| {
        (other.providers(customer) != Some(providers)).then_some(customer)
    };
    a.iter()
        .filter_map(|aspa| differs(aspa, b))
        .chain(b.iter().filter_map(|aspa| differs(aspa, a)))
        .collect()
}

/// How a query was answered.
enum Answer {
    /// With data, which is now in use.
    Complete,
    /// With a request to ask for everything instead.
    Reset,
}