//! Simple implementation for the internal map state of a ChanMgr.

use std::time::Duration;

use super::AbstractChannelFactory;

use super::{select, AbstractChannel, Pending, Sending};

use crate::{ChannelConfig, Dormancy, Error, Result};

use futures::FutureExt;

use std::result::Result as StdResult;

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

use std::sync::Arc;

use tor_async_utils::oneshot;

use tor_basic_utils::RngExt as _;

use tor_cell::chancell::msg::PaddingNegotiate;

use tor_config::PaddingLevel;

use tor_error::{error_report, internal, into_internal};

use tor_linkspec::{HasRelayIds, ListByRelayIds, RelayIds};

use tor_netdir::{params::NetParameters, params::CHANNEL_PADDING_TIMEOUT_UPPER_BOUND};

use tor_proto::channel::kist::{KistMode, KistParams};

use tor_proto::channel::padding::Parameters as PaddingParameters;

use tor_proto::channel::padding::ParametersBuilder as PaddingParametersBuilder;

use tor_proto::channel::ChannelPaddingInstructionsUpdates;

use tor_proto::ChannelPaddingInstructions;

use tor_units::{BoundedInt32, IntegerMilliseconds};

use tracing::info;

use void::{ResultVoidExt as _, Void};

#[cfg(test)]

mod padding_test;

/// All mutable state held by an `AbstractChannelMgr`.

///

/// One reason that this is an isolated type is that we want to

/// to limit the amount of code that can see and

/// lock the Mutex here.  (We're using a blocking mutex close to async

/// code, so we need to be careful.)

pub(crate) struct MgrState<C: AbstractChannelFactory> {

    /// The data, within a lock

    ///

    /// (Danger: this uses a blocking mutex close to async code.  This mutex

    /// must never be held while an await is happening.)

    inner: std::sync::Mutex<Inner<C>>,

}

/// Parameters for channels that we create, and that all existing channels are using

struct ChannelParams {

    /// Channel padding instructions

    padding: ChannelPaddingInstructions,

    /// KIST parameters

    kist: KistParams,

}

/// A map from channel id to channel state, plus necessary auxiliary state - inside lock

struct Inner<C: AbstractChannelFactory> {

    /// The channel factory type that we store.

    ///

    /// In this module we never use this _as_ an AbstractChannelFactory: we just

    /// hand out clones of it when asked.

    builder: C,

    /// A map from identity to channels, or to pending channel statuses.

    channels: ListByRelayIds<ChannelState<C::Channel>>,

    /// Parameters for channels that we create, and that all existing channels are using

    ///

    /// Will be updated by a background task, which also notifies all existing

    /// `Open` channels via `channels`.

    ///

    /// (Must be protected by the same lock as `channels`, or a channel might be

    /// created using being-replaced parameters, but not get an update.)

    channels_params: ChannelParams,

    /// The configuration (from the config file or API caller)

    config: ChannelConfig,

    /// Dormancy

    ///

    /// The last dormancy information we have been told about and passed on to our channels.

    /// Updated via `MgrState::set_dormancy` and hence `MgrState::reconfigure_general`,

    /// which then uses it to calculate how to reconfigure the channels.

    dormancy: Dormancy,

}

/// The state of a channel (or channel build attempt) within a map.

///

/// A ChannelState can be Open (representing a fully negotiated channel) or

/// Building (representing a pending attempt to build a channel). Both states

/// have a set of RelayIds, but these RelayIds represent slightly different

/// things:

///  * On a Building channel, the set of RelayIds is all the identities that we

///    require the peer to have. (The peer may turn out to have _more_

///    identities than this.)

///  * On an Open channel, the set of RelayIds is all the identities that

///    we were able to successfully authenticate for the peer.

pub(crate) enum ChannelState<C> {

    /// An open channel.

    ///

    /// This channel might not be usable: it might be closing or

    /// broken.  We need to check its is_usable() method before

    /// yielding it to the user.

    Open(OpenEntry<C>),

    /// A channel that's getting built.

    Building(PendingEntry),

}

/// An open channel entry.

#[derive(Clone)]

pub(crate) struct OpenEntry<C> {

    /// The underlying open channel.

    pub(crate) channel: Arc<C>,

    /// The maximum unused duration allowed for this channel.

    pub(crate) max_unused_duration: Duration,

}

/// A unique ID for a pending ([`PendingEntry`]) channel.

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]

pub(crate) struct UniqPendingChanId(u64);

impl UniqPendingChanId {

    /// Construct a new `UniqPendingChanId`.

        /// The next unique ID.

        static NEXT_ID: AtomicU64 = AtomicU64::new(0);

        // Relaxed ordering is fine; we don't care about how this

        // is instantiated with respect to other channels.

}

impl std::fmt::Display for UniqPendingChanId {

}

/// An entry for a not-yet-build channel

#[derive(Clone)]

pub(crate) struct PendingEntry {

    /// The keys of the relay to which we're trying to open a channel.

    pub(crate) ids: RelayIds,

    /// A future we can clone and listen on to learn when this channel attempt

    /// is successful or failed.

    ///

    /// This entry will be removed from the map (and possibly replaced with an

    /// `OpenEntry`) _before_ this future becomes ready.

    pub(crate) pending: Pending,

    /// A unique ID that allows us to find this exact pending entry later.

    pub(crate) unique_id: UniqPendingChanId,

}

impl<C> HasRelayIds for ChannelState<C>

where

    C: HasRelayIds,

{

        }

}

impl<C: Clone> ChannelState<C> {

    /// For testing: either give the Open channel inside this state,

    /// or panic if there is none.

    #[cfg(test)]

        }

}

/// Type of the `nf_ito_*` netdir parameters, convenience alias

type NfIto = IntegerMilliseconds<BoundedInt32<0, CHANNEL_PADDING_TIMEOUT_UPPER_BOUND>>;

/// Extract from a `NetParameters` which we need, conveniently organized for our processing

///

/// This type serves two functions at once:

///

///  1. Being a subset of the parameters, we can copy it out of

///     the netdir, before we do more complex processing - and, in particular,

///     before we obtain the lock on `inner` (which we need to actually handle the update,

///     because we need to combine information from the config with that from the netdir).

///

///  2. Rather than four separate named fields for the padding options,

///     it has arrays, so that it is easy to

///     select the values without error-prone recapitulation of field names.

#[derive(Debug, Clone)]

struct NetParamsExtract {

    /// `nf_ito_*`, the padding timeout parameters from the netdir consensus

    ///

    /// `nf_ito[ 0=normal, 1=reduced ][ 0=low, 1=high ]`

    /// are `nf_ito_{low,high}{,_reduced` from `NetParameters`.

    // TODO we could use some enum or IndexVec or something to make this less `0` and `1`

    nf_ito: [[NfIto; 2]; 2],

    /// The KIST parameters.

    kist: KistParams,

}

impl From<&NetParameters> for NetParamsExtract {

}

/// Build a `KistMode` from [`NetParameters`].

///

/// Used for converting [`kist_enabled`](NetParameters::kist_enabled)

/// to a corresponding `KistMode`.

    }

impl NetParamsExtract {

    /// Return the padding timer parameter low end, for reduced-ness `reduced`, as a `u32`

    /// Return the padding timer parameter high end, for reduced-ness `reduced`, as a `u32`

    /// Return and converts one padding parameter timer

    ///

    /// Internal function.

}

impl<C: AbstractChannel> ChannelState<C> {

    /// Return true if a channel is ready to expire.

    /// Update `expire_after` if a smaller duration than

    /// the given value is required to expire this channel.

        };

            // still in use

        };

            // no time remaining; drop now.

        };

            // Ignoring this edge case would result in a fairly benign race

            // condition outside of Shadow, but deadlock in Shadow.

}

impl<C: AbstractChannelFactory> MgrState<C> {

    /// Create a new empty `MgrState`.

    /// Run a function on the [`ListByRelayIds`] that implements the map in this `MgrState`.

    ///

    /// This function grabs a mutex: do not provide a slow function.

    ///

    /// We provide this function rather than exposing the channels set directly,

    /// to make sure that the calling code doesn't await while holding the lock.

    ///

    /// This is only `cfg(test)` since it can deadlock.

    ///

    /// # Deadlock

    ///

    /// Calling a method on [`MgrState`] from within `func` may cause a deadlock.

    #[cfg(test)]

    /// Return a copy of the builder stored in this state.

    /// Run a function to modify the builder stored in this state.

    ///

    /// # Deadlock

    ///

    /// Calling a method on [`MgrState`] from within `func` may cause a deadlock.

    #[allow(dead_code)]

    /// Remove every unusable state from the map in this state.

    #[cfg(test)]

    /// Request an open or pending channel to `target`. If `add_new_entry_if_not_found` is true and

    /// an open or pending channel isn't found, a new pending entry will be added and

    /// [`ChannelForTarget::NewEntry`] will be returned. This is all done as part of the same method

    /// so that all operations are performed under the same lock acquisition.

        use ChannelState::*;

        // The idea here is to choose the channel in two steps:

        //

        // - Eligibility: Get channels from the channel map and filter them down to only channels

        //   which are eligible to be returned.

        // - Ranking: From the eligible channels, choose the best channel.

        //

        // Another way to choose the channel could be something like: first try all canonical open

        // channels, then all non-canonical open channels, then all pending channels with all

        // matching relay ids, then remaining pending channels, etc. But this ends up being hard to

        // follow and inflexible (what if you want to prioritize pending channels over non-canonical

        // open channels?).

        // Open channels which are allowed for requests to `target`.

            }

            }

        {

            // At least one *open, usable* channel has been negotiated that overlaps only

            // partially with our target: it has proven itself to have _one_ of our target

            // identities, but not all.

            //

            // Because this channel exists, we know that our target cannot succeed, since relays

            // are not allowed to share _any_ identities.

            //return Ok(Some(Action::Return(Err(Error::IdentityConflict))));

        // Great, nothing interfered at all.

    /// Remove the pending channel identified by its `handle`.

    /// Upgrade the pending channel identified by its `handle` by replacing it with a new open

    /// `channel`.

        // Do all operations under the same lock acquisition.

    /// Reconfigure all channels as necessary

    ///

    /// (By reparameterizing channels as needed)

    /// This function will handle

    ///   - netdir update

    ///   - a reconfiguration

    ///   - dormancy

    ///

    /// For `new_config` and `new_dormancy`, `None` means "no change to previous info".

        use ChannelState as CS;

        // TODO when we support operation as a relay, inter-relay channels ought

        // not to get padding.

        };

            // The KIST params have changed: remember their value,

            // and reparameterize_kist()

        } else {

            // If the new KIST params are identical to the previous ones,

            // we don't need to call reparameterize_kist()

        };

            // Return early, nothing to reconfigure

            };

        }

    /// Expire all channels that have been unused for too long.

    ///

    /// Return a Duration until the next time at which

    /// a channel _could_ expire.

}

/// A channel for a given target relay.

pub(crate) enum ChannelForTarget<CF: AbstractChannelFactory> {

    /// A channel that is open.

    Open(Arc<CF::Channel>),

    /// A channel that is building.

    Pending(Pending),

    /// Information about a new pending channel entry.

    NewEntry((PendingChannelHandle, Sending)),

}

/// A handle for a pending channel.

///

/// WARNING: This handle should never be dropped, and should always be passed back into

/// [`MgrState::remove_pending_channel`] or [`MgrState::upgrade_pending_channel_to_open`], otherwise

/// the pending channel may be left in the channel map forever.

///

/// This handle must only be used with the `MgrState` from which it was given.

pub(crate) struct PendingChannelHandle {

    /// Any relay ID for this pending channel.

    relay_id: tor_linkspec::RelayId,

    /// The unique ID for this pending channel.

    unique_id: UniqPendingChanId,

    /// The pending channel has been removed from the channel map.

    chan_has_been_removed: bool,

}

impl PendingChannelHandle {

    /// Create a new [`PendingChannelHandle`].

    /// This should be called when the pending channel has been removed from the pending channel

    /// map. Not calling this will result in an error log message (and panic in debug builds) when

    /// this handle is dropped.

}

impl std::ops::Drop for PendingChannelHandle {

                "'PendingChannelHandle' dropped unexpectedly",

            );

}

/// Helper: return the objects used to inform pending tasks about a newly open or failed channel.

/// Helper: remove the pending channel identified by `handle` from `channel_map`.

        };

/// Converts config, dormancy, and netdir, into parameter updates

///

/// Calculates new parameters, updating `channels_params` as appropriate.

/// If anything changed, the corresponding update instruction is returned.

///

/// `channels_params` is updated with the new parameters,

/// and the update message, if one is needed, is returned.

///

/// This is called in two places:

///

///  1. During chanmgr creation, it is called once to analyze the initial state

///     and construct a corresponding ChannelPaddingInstructions.

///

///  2. During reconfiguration.

    };

    };

    // Whether the inbound padding approach we are to use, is the same as the default

    // derived from the netdir (disregarding our config and dormancy).

    //

    // Ie, whether the parameters we want are precisely those that a peer would

    // use by default (assuming they have the same view of the netdir as us).

        // Our padding approach is the same as peers' defaults.  So the PADDING_NEGOTIATE

        // message we need to send is the START(0,0).  (The channel frontend elides an

        // initial message of this form, - see crates/tor-proto/src/channel.rs::note_usage.)

        //

        // If the netdir default is no padding, and we previously negotiated

        // padding being enabled, and now want to disable it, we would send

        // START(0,0) rather than STOP.  That is OK (even, arguably, right).

    } else {

        }

    };

/// Given a `NetDirExtract` and whether we're reducing padding, return a `PaddingParameters`

///

/// With `PaddingLevel::None`, or the consensus specifies no padding, will return `None`;

/// but does not account for other reasons why padding might be enabled/disabled.

    };

            );

/// Given a `NetDirExtract` and whether we're reducing padding,

/// return a `PaddingParametersBuilder`

///

/// If the consensus specifies no padding, will return `None`;

/// but does not account for other reasons why padding might be enabled/disabled.

///

/// If `Err`, the string is a description of what is wrong with the parameters;

/// the caller should use `PaddingParameters::Default`.

        // Zeroes for both channel padding consensus parameters means "don't send padding".

        // padding-spec.txt s2.6, see description of `nf_ito_high`.

#[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::factory::BootstrapReporter;

    use async_trait::async_trait;

    use std::sync::{Arc, Mutex};

    use tor_llcrypto::pk::ed25519::Ed25519Identity;

    use tor_proto::channel::params::ChannelPaddingInstructionsUpdates;

    use tor_proto::memquota::ChannelAccount;

    fn new_test_state() -> MgrState<FakeChannelFactory> {

        MgrState::new(

            FakeChannelFactory::default(),

            ChannelConfig::default(),

            Default::default(),

            &Default::default(),

        )

    }

    #[derive(Clone, Debug, Default)]

    struct FakeChannelFactory {}

    #[allow(clippy::diverging_sub_expression)] // for unimplemented!() + async_trait

    #[async_trait]

    impl AbstractChannelFactory for FakeChannelFactory {

        type Channel = FakeChannel;

        type BuildSpec = tor_linkspec::OwnedChanTarget;

        type Stream = ();

        async fn build_channel(

            &self,

            _target: &Self::BuildSpec,

            _reporter: BootstrapReporter,

            _memquota: ChannelAccount,

        ) -> Result<Arc<FakeChannel>> {

            unimplemented!()

        }

        #[cfg(feature = "relay")]

        async fn build_channel_using_incoming(

            &self,

            _peer: std::net::SocketAddr,

            _stream: Self::Stream,

            _memquota: ChannelAccount,

        ) -> Result<Arc<Self::Channel>> {

            unimplemented!()

        }

    }

    #[derive(Clone, Debug)]

    struct FakeChannel {

        ed_ident: Ed25519Identity,

        usable: bool,

        unused_duration: Option<u64>,

        params_update: Arc<Mutex<Option<Arc<ChannelPaddingInstructionsUpdates>>>>,

    }

    impl AbstractChannel for FakeChannel {

        fn is_usable(&self) -> bool {

            self.usable

        }

        fn duration_unused(&self) -> Option<Duration> {

            self.unused_duration.map(Duration::from_secs)

        }

        fn reparameterize(

            &self,

            update: Arc<ChannelPaddingInstructionsUpdates>,

        ) -> tor_proto::Result<()> {

            *self.params_update.lock().unwrap() = Some(update);

            Ok(())

        }

        fn reparameterize_kist(&self, _kist_params: KistParams) -> tor_proto::Result<()> {

            Ok(())

        }

        fn engage_padding_activities(&self) {}

    }

    impl tor_linkspec::HasRelayIds for FakeChannel {

        fn identity(

            &self,

            key_type: tor_linkspec::RelayIdType,

        ) -> Option<tor_linkspec::RelayIdRef<'_>> {

            match key_type {

                tor_linkspec::RelayIdType::Ed25519 => Some((&self.ed_ident).into()),

                _ => None,

            }

        }

    }

    /// Get a fake ed25519 identity from the first byte of a string.

    fn str_to_ed(s: &str) -> Ed25519Identity {

        let byte = s.as_bytes()[0];

        [byte; 32].into()

    }

    fn ch(ident: &'static str) -> ChannelState<FakeChannel> {

        let channel = FakeChannel {

            ed_ident: str_to_ed(ident),

            usable: true,

            unused_duration: None,

            params_update: Arc::new(Mutex::new(None)),

        };

        ChannelState::Open(OpenEntry {

            channel: Arc::new(channel),

            max_unused_duration: Duration::from_secs(180),

        })

    }

    fn ch_with_details(

        ident: &'static str,

        max_unused_duration: Duration,

        unused_duration: Option<u64>,

    ) -> ChannelState<FakeChannel> {

        let channel = FakeChannel {

            ed_ident: str_to_ed(ident),

            usable: true,

            unused_duration,

            params_update: Arc::new(Mutex::new(None)),

        };

        ChannelState::Open(OpenEntry {

            channel: Arc::new(channel),

            max_unused_duration,

        })

    }

    fn closed(ident: &'static str) -> ChannelState<FakeChannel> {

        let channel = FakeChannel {

            ed_ident: str_to_ed(ident),

            usable: false,

            unused_duration: None,

            params_update: Arc::new(Mutex::new(None)),

        };

        ChannelState::Open(OpenEntry {

            channel: Arc::new(channel),

            max_unused_duration: Duration::from_secs(180),

        })

    }

    #[test]

    fn rmv_unusable() -> Result<()> {

        let map = new_test_state();

        map.with_channels(|map| {

            map.insert(closed("machen"));

            map.insert(closed("wir"));

            map.insert(ch("wir"));

            map.insert(ch("feinen"));

            map.insert(ch("Fug"));

            map.insert(ch("Fug"));

        })?;

        map.remove_unusable().unwrap();

        map.with_channels(|map| {

            assert_eq!(map.by_id(&str_to_ed("m")).len(), 0);

            assert_eq!(map.by_id(&str_to_ed("w")).len(), 1);

            assert_eq!(map.by_id(&str_to_ed("f")).len(), 1);

            assert_eq!(map.by_id(&str_to_ed("F")).len(), 2);

        })?;

        Ok(())

    }

    #[test]

    fn reparameterize_via_netdir() -> Result<()> {

        let map = new_test_state();

        // Set some non-default parameters so that we can tell when an update happens

        let _ = map

            .inner

            .lock()

            .unwrap()

            .channels_params

            .padding

            .start_update()

            .padding_parameters(

                PaddingParametersBuilder::default()

                    .low(1234.into())

                    .build()

                    .unwrap(),

            )

            .finish();

        map.with_channels(|map| {

            map.insert(ch("track"));

        })?;

        let netdir = tor_netdir::testnet::construct_netdir()

            .unwrap_if_sufficient()

            .unwrap();

        let netdir = Arc::new(netdir);

        let with_ch = |f: &dyn Fn(&FakeChannel)| {

            let inner = map.inner.lock().unwrap();

            let mut ch = inner.channels.by_ed25519(&str_to_ed("t"));

            let ch = ch.next().unwrap().unwrap_open();

            f(ch);

        };

        eprintln!("-- process a default netdir, which should send an update --");

        map.reconfigure_general(None, None, netdir.clone()).unwrap();

        with_ch(&|ch| {

            assert_eq!(

                format!("{:?}", ch.params_update.lock().unwrap().take().unwrap()),

                // evade field visibility by (ab)using Debug impl

                "ChannelPaddingInstructionsUpdates { padding_enable: None, \

                    padding_parameters: Some(Parameters { \

                        low: IntegerMilliseconds { value: 1500 }, \

                        high: IntegerMilliseconds { value: 9500 } }), \

                    padding_negotiate: None }"

            );

        });

        eprintln!();

        eprintln!("-- process a default netdir again, which should *not* send an update --");

        map.reconfigure_general(None, None, netdir).unwrap();

        with_ch(&|ch| assert!(ch.params_update.lock().unwrap().is_none()));

        Ok(())

    }

    #[test]

    fn expire_channels() -> Result<()> {

        let map = new_test_state();

        // Channel that has been unused beyond max duration allowed is expired

        map.with_channels(|map| {

            map.insert(ch_with_details(

                "wello",

                Duration::from_secs(180),

                Some(181),

            ));

        })?;

        // Minimum value of max unused duration is 180 seconds

        assert_eq!(180, map.expire_channels().as_secs());

        map.with_channels(|map| {

            assert_eq!(map.by_ed25519(&str_to_ed("w")).len(), 0);

        })?;

        let map = new_test_state();

        // Channel that has been unused for shorter than max unused duration

        map.with_channels(|map| {

            map.insert(ch_with_details(

                "wello",

                Duration::from_secs(180),

                Some(120),

            ));

            map.insert(ch_with_details(

                "yello",

                Duration::from_secs(180),

                Some(170),

            ));

            // Channel that has been unused beyond max duration allowed is expired

            map.insert(ch_with_details(

                "gello",

                Duration::from_secs(180),

                Some(181),

            ));

            // Closed channel should be retained

            map.insert(closed("hello"));

        })?;

        // Return duration until next channel expires

        assert_eq!(10, map.expire_channels().as_secs());

        map.with_channels(|map| {

            assert_eq!(map.by_ed25519(&str_to_ed("w")).len(), 1);

            assert_eq!(map.by_ed25519(&str_to_ed("y")).len(), 1);

            assert_eq!(map.by_ed25519(&str_to_ed("h")).len(), 1);

            assert_eq!(map.by_ed25519(&str_to_ed("g")).len(), 0);

        })?;

        Ok(())

    }

}