//! Functions to download or load directory objects, using the

//! state machines in the `states` module.

use std::num::NonZeroUsize;

use std::ops::Deref;

use std::result::Result as StdResult;

use std::{

    collections::HashMap,

    sync::{Arc, Weak},

    time::{Duration, SystemTime},

};

use crate::DirMgrConfig;

use crate::DocSource;

use crate::err::BootstrapAction;

use crate::state::{DirState, PoisonedState};

use crate::{

    DirMgr, DocId, DocQuery, DocumentText, Error, Readiness, Result,

    docid::{self, ClientRequest},

    upgrade_weak_ref,

};

use futures::FutureExt;

use futures::StreamExt;

use oneshot_fused_workaround as oneshot;

use tor_dirclient::DirResponse;

use tor_error::{info_report, warn_report};

use tor_rtcompat::Runtime;

use tor_rtcompat::scheduler::TaskSchedule;

use tracing::{debug, info, instrument, trace, warn};

use crate::storage::Store;

#[cfg(test)]

use std::sync::LazyLock;

#[cfg(test)]

use std::sync::Mutex;

use tor_circmgr::{CircMgr, DirInfo};

use tor_netdir::{NetDir, NetDirProvider as _};

use tor_netdoc::doc::netstatus::ConsensusFlavor;

/// Given a Result<()>, exit the current function if it is anything other than

/// Ok(), or a nonfatal error.

macro_rules! propagate_fatal_errors {

    ( $e:expr ) => {

        let v: Result<()> = $e;

        if let Err(e) = v {

            match e.bootstrap_action() {

                BootstrapAction::Nonfatal => {}

                _ => return Err(e),

            }

        }

    };

}

/// Identifier for an attempt to bootstrap a directory.

///

/// Every time that we decide to download a new directory, _despite already

/// having one_, counts as a new attempt.

///

/// These are used to track the progress of each attempt independently.

#[derive(Copy, Clone, Debug, derive_more::Display, Eq, PartialEq, Ord, PartialOrd)]

#[display("{0}", id)]

pub(crate) struct AttemptId {

    /// Which attempt at downloading a directory is this?

    id: NonZeroUsize,

}

impl AttemptId {

    /// Return a new unused AtomicUsize that will be greater than any previous

    /// one.

    ///

    /// # Panics

    ///

    /// Panics if we have exhausted the possible space of AtomicIds.

        use std::sync::atomic::{AtomicUsize, Ordering};

        /// atomic used to generate the next attempt.

        static NEXT: AtomicUsize = AtomicUsize::new(1);

}

/// If there were errors from a peer in `outcome`, record those errors by

/// marking the circuit (if any) as needing retirement, and noting the peer

/// (if any) as having failed.

    use tor_dirclient::{Error::RequestFailed, RequestFailedError};

    // Extract an error and a source from this outcome, if there is one.

    //

    // This is complicated because DirResponse can encapsulate the notion of

    // a response that failed part way through a download: in the case, it

    // has some data, and also an error.

            } else {

            }

        }

        Err(

                ..

            }),

    };

/// Record that a problem has occurred because of a failure in an answer from `source`.

    use tor_circmgr::ExternalActivity;

    // Does the error here tell us whom to really blame?  If so, blame them

    // instead.

    //

    // (This can happen if we notice a problem while downloading a certificate,

    // but the real problem is that the consensus was no good.)

        Error::NetDocError {

            ..

    };

/// Record that `source` has successfully given us some directory info.

    use tor_circmgr::ExternalActivity;

/// Load every document in `missing` and try to apply it to `state`.

    /// How many documents will we try to load at once?  We try to keep this from being too large,

    /// to avoid excessive RAM usage.

    ///

    /// TODO: we may well want to tune this.

    const CHUNK_SIZE: usize = 256;

        };

    }

/// Load a set of documents from a `Store`, returning all documents found in the store.

/// Note that this may be less than the number of documents in `missing`.

    }

/// Try to update `state` by loading cached information, on success returns the

/// current wall-clock time.

        // If the load failed but the error can be blamed on a directory

        // cache, do so.

/// Applies changes from `state` to the manager's netdir.

///

/// Updates and broadcasts download progress based on the current `state`.

/// Panics if manager's store lock is poisoned.

// TODO(eta): Consider deprecating state.is_ready().

/// Errors returned by `advance_state`.

enum AdvanceStateError {

    /// The directory has already been downloaded completely.

    AlreadyComplete,

    /// The state can't currently be advanced.

    CantAdvanceYet,

}

/// Advances the progress of the directory download.

///

/// Returns `Ok(())` if the state was successfully advanced.

///

/// # Errors

///

/// - Returns `AdvanceStateError::AlreadyComplete` if the directory has already been

///   fully downloaded.

/// - Returns `AdvanceStateError::CantAdvanceYet` if the directory is incomplete and

///   the download cannot advance yet.

/// Return value of `perform_download`.

///

/// Describes the outcome of the attempt to download the current directory's state.

enum DownloadOutcome {

    /// the download failed with a non-fatal error.

    DownloadFailed,

    /// The directory being downloaded is outdated.

    DirectoryOutdated,

    /// The download attempt was successful, and the result has been applied to

    /// the state.

    Applied,

}

/// Attempts to download the current directory's state.

///

/// On success, it reports the outcome of the download using a

/// `DownloadOutcome` instance and updates `now` to the current

/// wallclock time.

///

/// The possible outcomes are:

/// - `DownloadOutcome::DownloadFailed`: the download failed with a non-fatal

///   error; the caller should proceed to the next state.

/// - `DownloadOutcome::DirectoryOutdated`: The directory being downloaded is

///   outdated. The caller should attempt a new download. As a side effect,

///   the directory state has been reset.

/// - `DownloadOutcome::Applied`: The download attempt was successful, and the

///   result has been applied to the state.

///

/// Returns an error only if a fatal error occurred.

    *now = {

                } else {

                }

            }

                // We need to reset. This can happen if (for

                // example) we're downloading the last few

                // microdescriptors on a consensus that now

                // we're ready to replace.

            },

        };

    };

/// Construct an appropriate ClientRequest to download a consensus

/// of the given flavor.

            // If we don't have a consensus, then request one that's

            // "reasonably new".  That way, our clock is set far in the

            // future, we won't download stuff we can't use.

        }

    }

        // If we are _fast_ by at least this much, then any valid directory will

        // seem to be at least this far in the past.

        // If we are _slow_ by this much, then any valid directory will seem to

        // be at least this far in the future.

    );

/// Construct a set of `ClientRequest`s in order to fetch the documents in `docs`.

    {

            }

            #[cfg(feature = "routerdesc")]

        }

    }

/// Launch a single client request and get an associated response.

#[instrument(level = "trace", skip_all)]

        Some(netdir) => netdir.into(),

        None => tor_circmgr::DirInfo::Nothing,

    };

    let outcome =

        tor_dirclient::get_resource(request.as_requestable(), dirinfo, rt, circmgr.clone()).await;

    note_request_outcome(&circmgr, &outcome);

    let resource = outcome?;

    Ok((request, resource))

/// Testing helper: if this is Some, then we return it in place of any

/// response to fetch_multiple.

///

/// Note that only one test uses this: otherwise there would be a race

/// condition. :p

#[cfg(test)]

/// Launch a set of download requests for a set of missing objects in

/// `missing`, and return each request along with the response it received.

///

/// Don't launch more than `parallelism` requests at once.

#[instrument(level = "trace", skip_all)]

    let requests = {

        let store = dirmgr.store.lock().expect("store lock poisoned");

        make_requests_for_documents(&dirmgr.runtime, missing, &**store, &dirmgr.config.get())?

    };

    trace!(attempt=%attempt_id, "Launching {} requests for {} documents",

           requests.len(), missing.len());

    #[cfg(test)]

    {

        let m = CANNED_RESPONSE.lock().expect("Poisoned mutex");

        if !m.is_empty() {

            return Ok(requests

                .into_iter()

                .zip(m.iter().map(DirResponse::from_get_body))

                .collect());

        }

    }

    let circmgr = dirmgr.circmgr()?;

    // Only use timely directories for bootstrapping directories; otherwise, we'll try fallbacks.

    let netdir = dirmgr.netdir(tor_netdir::Timeliness::Timely).ok();

    // TODO: instead of waiting for all the queries to finish, we

    // could stream the responses back or something.

    let responses: Vec<Result<(ClientRequest, DirResponse)>> = futures::stream::iter(requests)

        .buffer_unordered(parallelism)

        .collect()

        .await;

    let mut useful_responses = Vec::new();

    for r in responses {

        // TODO: on some error cases we might want to stop using this source.

        match r {

            Ok((request, response)) => {

                if response.status_code() == 200 {

                    useful_responses.push((request, response));

                } else {

                    trace!(

                        "cache declined request; reported status {:?}",

                        response.status_code()

                    );

                }

            }

            Err(e) => warn_report!(e, "error while downloading"),

        }

    }

    trace!(attempt=%attempt_id, "received {} useful responses from our requests.", useful_responses.len());

    Ok(useful_responses)

/// Try to update `state` by loading cached information from `dirmgr`.

    } else {

            "Found {} missing documents; trying to load them",

        );

    };

    // We have to update the status here regardless of the outcome, if we got

    // any information: even if there was an error, we might have received

    // partial information that changed our status.

/// Try to load as much state as possible for a provided `state` from the

/// cache in `dirmgr`, advancing the state to the extent possible.

///

/// No downloads are performed; the provided state will not be reset.

    loop {

                BootstrapAction::Nonfatal => {

                }

                BootstrapAction::Fatal | BootstrapAction::Reset => {

                }

            }

        } else {

                // TODO: Are there more nonfatal errors that mean we should

                // break?

                "Spent 100 iterations in the same state: this is a bug"

            );

        }

    }

/// Helper: Make a set of download attempts for the current directory state,

/// and on success feed their results into the state object.

///

/// This can launch one or more download requests, but will not launch more

/// than `parallelism` requests at a time.

#[instrument(level = "trace", skip_all)]

    let missing = state.missing_docs();

    let fetched = fetch_multiple(Arc::clone(dirmgr), attempt_id, &missing, parallelism).await?;

    let mut n_errors = 0;

    for (client_req, dir_response) in fetched {

        let source = dir_response.source().cloned();

        let text = match String::from_utf8(dir_response.into_output_unchecked())

            .map_err(Error::BadUtf8FromDirectory)

        {

            Ok(t) => t,

            Err(e) => {

                if let Some(source) = source {

                    n_errors += 1;

                    note_cache_error(dirmgr.circmgr()?.deref(), &source, &e);

                }

                continue;

            }

        };

        match dirmgr.expand_response_text(&client_req, text) {

            Ok(text) => {

                let doc_source = DocSource::DirServer {

                    source: source.clone(),

                };

                let mut changed = false;

                let outcome = state.add_from_download(

                    &text,

                    &client_req,

                    doc_source,

                    Some(&dirmgr.store),

                    &mut changed,

                );

                if !changed {

                    debug_assert!(outcome.is_err());

                }

                if let Some(source) = source {

                    if let Err(e) = &outcome {

                        n_errors += 1;

                        note_cache_error(dirmgr.circmgr()?.deref(), &source, e);

                    } else {

                        note_cache_success(dirmgr.circmgr()?.deref(), &source);

                    }

                }

                if let Err(e) = &outcome {

                    dirmgr.note_errors(attempt_id, 1);

                    warn_report!(e, "error while adding directory info");

                }

                propagate_fatal_errors!(outcome);

            }

            Err(e) => {

                warn_report!(e, "Error when expanding directory text");

                if let Some(source) = source {

                    n_errors += 1;

                    note_cache_error(dirmgr.circmgr()?.deref(), &source, &e);

                }

                propagate_fatal_errors!(Err(e));

            }

        }

    }

    if n_errors != 0 {

        dirmgr.note_errors(attempt_id, n_errors);

    }

    dirmgr.update_progress(attempt_id, state.bootstrap_progress());

    Ok(())

/// Download information into a DirState state machine until it is

/// ["complete"](Readiness::Complete), or until we hit a non-recoverable error.

///

/// Use `dirmgr` to load from the cache or to launch downloads.

///

/// Keep resetting the state as needed.

///

/// The first time that the state becomes ["usable"](Readiness::Usable), notify

/// the sender in `tx_usable`.

#[instrument(level = "trace", skip_all)]

    let runtime = upgrade_weak_ref(&dirmgr)?.runtime.clone();

    trace!(attempt=%attempt_id, state=%state.describe(), "Trying to download directory material.");

    'next_state: loop {

        let retry_config = state.dl_config();

        let parallelism = retry_config.parallelism();

        // In theory this could be inside the loop below maybe?  If we

        // want to drop the restriction that the missing() members of a

        // state must never grow, then we'll need to move it inside.

        let mut now = update_state(&upgrade_weak_ref(&dirmgr)?, attempt_id, state)?;

        apply_state(&upgrade_weak_ref(&dirmgr)?, state, attempt_id)?;

        // Skip the downloads if we can...

        match advance_state(state, attempt_id) {

            Ok(_) => continue 'next_state,

            Err(AdvanceStateError::AlreadyComplete) => return Ok(()),

            Err(AdvanceStateError::CantAdvanceYet) => {}

        }

        let reset_time = no_more_than_a_week_from(runtime.wallclock(), state.reset_time());

        let mut retry = retry_config.schedule();

        let mut delay = None;

        // Make several attempts to fetch whatever we're missing,

        // until either we can advance, or we've got a complete

        // document, or we run out of tries, or we run out of time.

        'next_attempt: for attempt in retry_config.attempts() {

            // We wait at the start of this loop, on all attempts but the first.

            // This ensures that we always wait between attempts, but not after

            // the final attempt.

            let next_delay = retry.next_delay(&mut rand::rng());

            if let Some(delay) = delay.replace(next_delay) {

                let time_until_reset = reset_time

                    .duration_since(now)

                    .unwrap_or(Duration::from_secs(0));

                let real_delay = delay.min(time_until_reset);

                debug!(attempt=%attempt_id, "Waiting {:?} for next download attempt...", real_delay);

                schedule.sleep(real_delay).await?;

                now = upgrade_weak_ref(&dirmgr)?.runtime.wallclock();

                if now >= reset_time {

                    info!(attempt=%attempt_id, "Directory being fetched is now outdated; resetting download state.");

                    reset(state);

                    continue 'next_state;

                }

            }

            info!(attempt=%attempt_id, "{}: {}", attempt + 1, state.describe());

            match perform_download(attempt_id, &dirmgr, &mut now, parallelism, schedule, state)

                .await?

            {

                DownloadOutcome::DownloadFailed => continue 'next_state,

                DownloadOutcome::DirectoryOutdated => continue 'next_attempt,

                DownloadOutcome::Applied => {}

            }

            propagate_fatal_errors!(apply_state(&upgrade_weak_ref(&dirmgr)?, state, attempt_id));

            // Exit if the download is complete. Report usable-ness if appropriate.

            match advance_state(state, attempt_id) {

                Err(AdvanceStateError::AlreadyComplete) => return Ok(()),

                Ok(()) => continue 'next_state,

                Err(AdvanceStateError::CantAdvanceYet) => {

                    if state.is_ready(Readiness::Usable) {

                        if let Some(tx) = tx_usable.take() {

                            trace!(attempt=%attempt_id, state=%state.describe(), "directory is now usable.");

                            let _ = tx.send(());

                        }

                    }

                }

            }

        }

        // We didn't advance the state, after all the retries.

        warn!(n_attempts=retry_config.n_attempts(),

              state=%state.describe(),

              "Unable to advance downloading state");

        return Err(Error::CantAdvanceState);

    }

/// Replace `state` with `state.reset()`.

/// Replace `state` with `state.advance()`.

/// Helper: Clamp `v` so that it is no more than one week from `now`.

///

/// If `v` is absent, return the time that's one week from now.

///

/// We use this to determine a reset time when no reset time is

/// available, or when it is too far in the future.

    }

#[cfg(test)]

mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_time_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    #![allow(clippy::string_slice)] // See arti#2571

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use super::*;

    use crate::storage::DynStore;

    use crate::test::new_mgr;

    use std::sync::Mutex;

    use tor_dircommon::retry::DownloadSchedule;

    use tor_netdoc::doc::microdesc::MdDigest;

    use tor_rtcompat::SleepProvider;

    use web_time_compat::SystemTimeExt;

    #[test]

    fn week() {

        let now = SystemTime::get();

        let one_day = Duration::new(86400, 0);

        assert_eq!(no_more_than_a_week_from(now, None), now + one_day * 7);

        assert_eq!(

            no_more_than_a_week_from(now, Some(now + one_day)),

            now + one_day

        );

        assert_eq!(

            no_more_than_a_week_from(now, Some(now - one_day)),

            now - one_day

        );

        assert_eq!(

            no_more_than_a_week_from(now, Some(now + 30 * one_day)),

            now + one_day * 7

        );

    }

    /// A fake implementation of DirState that just wants a fixed set

    /// of microdescriptors.  It doesn't care if it gets them: it just

    /// wants to be told that the IDs exist.

    #[derive(Debug, Clone)]

    struct DemoState {

        second_time_around: bool,

        got_items: HashMap<MdDigest, bool>,

    }

    // Constants from Lou Reed

    const H1: MdDigest = *b"satellite's gone up to the skies";

    const H2: MdDigest = *b"things like that drive me out of";

    const H3: MdDigest = *b"my mind i watched it for a littl";

    const H4: MdDigest = *b"while i like to watch things on ";

    const H5: MdDigest = *b"TV Satellite of love Satellite--";

    impl DemoState {

        fn new1() -> Self {

            DemoState {

                second_time_around: false,

                got_items: vec![(H1, false), (H2, false)].into_iter().collect(),

            }

        }

        fn new2() -> Self {

            DemoState {

                second_time_around: true,

                got_items: vec![(H3, false), (H4, false), (H5, false)]

                    .into_iter()

                    .collect(),

            }

        }

        fn n_ready(&self) -> usize {

            self.got_items.values().filter(|x| **x).count()

        }

    }

    impl DirState for DemoState {

        fn describe(&self) -> String {

            format!("{:?}", self)

        }

        fn bootstrap_progress(&self) -> crate::event::DirProgress {

            crate::event::DirProgress::default()

        }

        fn is_ready(&self, ready: Readiness) -> bool {

            match (ready, self.second_time_around) {

                (_, false) => false,

                (Readiness::Complete, true) => self.n_ready() == self.got_items.len(),

                (Readiness::Usable, true) => self.n_ready() >= self.got_items.len() - 1,

            }

        }

        fn can_advance(&self) -> bool {

            if self.second_time_around {

                false

            } else {

                self.n_ready() == self.got_items.len()

            }

        }

        fn missing_docs(&self) -> Vec<DocId> {

            self.got_items

                .iter()

                .filter_map(|(id, have)| {

                    if *have {

                        None

                    } else {

                        Some(DocId::Microdesc(*id))

                    }

                })

                .collect()

        }

        fn add_from_cache(

            &mut self,

            docs: HashMap<DocId, DocumentText>,

            changed: &mut bool,

        ) -> Result<()> {

            for id in docs.keys() {

                if let DocId::Microdesc(id) = id {

                    if self.got_items.get(id) == Some(&false) {

                        self.got_items.insert(*id, true);

                        *changed = true;

                    }

                }

            }

            Ok(())

        }

        fn add_from_download(

            &mut self,

            text: &str,

            _request: &ClientRequest,

            _source: DocSource,

            _storage: Option<&Mutex<DynStore>>,

            changed: &mut bool,

        ) -> Result<()> {

            for token in text.split_ascii_whitespace() {

                if let Ok(v) = hex::decode(token) {

                    if let Ok(id) = v.try_into() {

                        if self.got_items.get(&id) == Some(&false) {

                            self.got_items.insert(id, true);

                            *changed = true;

                        }

                    }

                }

            }

            Ok(())

        }

        fn dl_config(&self) -> DownloadSchedule {

            DownloadSchedule::default()

        }

        fn advance(self: Box<Self>) -> Box<dyn DirState> {

            if self.can_advance() {

                Box::new(Self::new2())

            } else {

                self

            }

        }

        fn reset_time(&self) -> Option<SystemTime> {

            None

        }

        fn reset(self: Box<Self>) -> Box<dyn DirState> {

            Box::new(Self::new1())

        }

    }

    #[test]

    fn all_in_cache() {

        // Let's try bootstrapping when everything is in the cache.

        tor_rtcompat::test_with_one_runtime!(|rt| async {

            let now = rt.wallclock();

            let (_tempdir, mgr) = new_mgr(rt.clone());

            let (mut schedule, _handle) = TaskSchedule::new(rt);

            {

                let mut store = mgr.store_if_rw().unwrap().lock().unwrap();

                for h in [H1, H2, H3, H4, H5] {

                    store.store_microdescs(&[("ignore", &h)], now).unwrap();

                }

            }

            let mgr = Arc::new(mgr);

            let attempt_id = AttemptId::next();

            // Try just a load.

            let state = Box::new(DemoState::new1());

            let result = super::load(&mgr, state, attempt_id).unwrap();

            assert!(result.is_ready(Readiness::Complete));

            // Try a bootstrap that could (but won't!) download.

            let mut state: Box<dyn DirState> = Box::new(DemoState::new1());

            let mut tx_usable = None;

            super::download(

                Arc::downgrade(&mgr),

                &mut state,

                &mut schedule,

                attempt_id,

                &mut tx_usable,

            )

            .await

            .unwrap();

            assert!(state.is_ready(Readiness::Complete));

        });

    }

    #[test]

    fn partly_in_cache() {

        // Let's try bootstrapping with all of phase1 and part of

        // phase 2 in cache.

        tor_rtcompat::test_with_one_runtime!(|rt| async {

            let now = rt.wallclock();

            let (_tempdir, mgr) = new_mgr(rt.clone());

            let (mut schedule, _handle) = TaskSchedule::new(rt);

            {

                let mut store = mgr.store_if_rw().unwrap().lock().unwrap();

                for h in [H1, H2, H3] {

                    store.store_microdescs(&[("ignore", &h)], now).unwrap();

                }

            }

            {

                let mut resp = CANNED_RESPONSE.lock().unwrap();

                // H4 and H5.

                *resp = vec![

                    "7768696c652069206c696b6520746f207761746368207468696e6773206f6e20

                     545620536174656c6c697465206f66206c6f766520536174656c6c6974652d2d"

                        .to_owned(),

                ];

            }

            let mgr = Arc::new(mgr);

            let mut tx_usable = None;

            let attempt_id = AttemptId::next();

            let mut state: Box<dyn DirState> = Box::new(DemoState::new1());

            super::download(

                Arc::downgrade(&mgr),

                &mut state,

                &mut schedule,

                attempt_id,

                &mut tx_usable,

            )

            .await

            .unwrap();

            assert!(state.is_ready(Readiness::Complete));

        });

    }

}