//! 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::{

    collections::HashMap,

    sync::{Arc, Weak},

    time::{Duration, SystemTime},

};

use crate::err::BootstrapAction;

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

use crate::DirMgrConfig;

use crate::DocSource;

use crate::{

    docid::{self, ClientRequest},

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

};

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::scheduler::TaskSchedule;

use tor_rtcompat::Runtime;

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

use crate::storage::Store;

#[cfg(test)]

use once_cell::sync::Lazy;

#[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`.

    }

/// 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.

        }

    }

/// 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.

    };

/// 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.

    };

    #[cfg(test)]

    {

    }

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

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

    // could stream the responses back or something.

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

                    );

                }

            }

        }

    }

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

    } else {

        );

    };

    // 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?

            );

        }

    }

/// 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.

        {

            }

        };

                    } else {

                    }

            }

            }

        }

    }

/// 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 `on_usable`.

    'next_state: loop {

        // 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.

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

                // cache, do so.

        };

        // Apply any netdir changes that the state gives us.

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

        {

        // 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.

            // 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.

            now = {

                            // TODO: get warn_report! to support `attempt=%attempt_id`?

                        } 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.

                    },

                };

            };

            // Apply any netdir changes that the state gives us.

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

            {

            }

            // Exit if there is nothing more to download.

                // Unwrap should be safe due to parent `.is_some()` check

                #[allow(clippy::unwrap_used)]

                // We have enough info to advance to another state.

        }

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

    }

/// 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_duration_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

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

    use super::*;

    use crate::storage::DynStore;

    use crate::test::new_mgr;

    use crate::DownloadSchedule;

    use std::sync::Mutex;

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

    use tor_rtcompat::SleepProvider;

    #[test]

    fn week() {

        let now = SystemTime::now();

        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(Arc::clone(&mgr), state, attempt_id)

                .await

                .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 on_usable = None;

            super::download(

                Arc::downgrade(&mgr),

                &mut state,

                &mut schedule,

                attempt_id,

                &mut on_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 on_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 on_usable,

            )

            .await

            .unwrap();

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

        });

    }

}