use crate::common::*;

use crate::runtime::{actors::*, endpoint::*, errors::*, *};

impl Controller {

    //usable BETWEEN rounds

    fn consistent(&self) -> bool {

        self.inner.mono_n.is_some()

    }

    fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {

        log!(&mut self.inner.logger, "ENDING ROUND WITH DECISION! {:?}", &decision);

        // let mut table_row = HashMap::<Key, _>::default();

        let n_pair = {

            let poly_n = self.ephemeral.poly_n.take().unwrap();

            let mono_n = poly_n.choose_mono(&decision).unwrap();

            (mono_n, poly_n)

        };

        self.inner.mono_n = Some(n_pair.0.clone());

        let p_pairs = self

            .ephemeral

            .poly_ps

            .drain(..)

            .map(|poly_p| {

                let mono_p = poly_p.choose_mono(&decision).unwrap();

                (mono_p, poly_p)

            })

        self.inner.mono_ps.extend(p_pairs.iter().map(|p_pair| p_pair.0.clone()));

        let announcement =

            CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);

        for &child_ekey in self.inner.family.children_ekeys.iter() {

            log!(

                &mut self.inner.logger,

                "Forwarding {:?} to child with ekey {:?}",

                &announcement,

                child_ekey

            );

            self.inner

                .endpoint_exts

                .get_mut(child_ekey)

                .expect("eefef")

                .endpoint

                .send(announcement.clone())?;

        }

        self.inner.round_index += 1;

        self.ephemeral.clear();

        Ok(())

    }

    // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found

    fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {

        if let Some(parent_ekey) = self.inner.family.parent_ekey {

            // I have a parent -> I'm not the leader

            let parent_endpoint =

                &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;

            for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {

                let msg =

                    CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);

                log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_ekey);

                parent_endpoint.send(msg)?;

            }

            Ok(false)

        } else {

            // I have no parent -> I'm the leader

            assert!(self.inner.family.parent_ekey.is_none());

            let maybe_decision = self.ephemeral.solution_storage.iter_new_local_make_old().next();

            Ok(if let Some(decision) = maybe_decision {

                log!(&mut self.inner.logger, "DECIDE ON {:?} AS LEADER!", &decision);

                self.end_round_with_decision(decision)?;

                true

            } else {

                false

            })

        }

    }

    fn kick_off_native(

        &mut self,

        sync_batches: impl Iterator<Item = SyncBatch>,

    ) -> Result<PolyN, EndpointErr> {

        let MonoN { ekeys, .. } = self.inner.mono_n.take().unwrap();

        let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;

        let mut branches = HashMap::<_, _>::default();

        for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {

            let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;

            let all_ekeys = ekeys.iter().copied();

            let all_channel_ids = all_ekeys.map(ekey_to_channel_id);

            let mut predicate = Predicate::new_trivial();

            // assign TRUE for puts and gets

            let true_ekeys = puts.keys().chain(gets.iter()).copied();

            let true_channel_ids = true_ekeys.clone().map(ekey_to_channel_id);

            predicate.batch_assign_nones(true_channel_ids, true);

            // assign FALSE for all in interface not assigned true

            predicate.batch_assign_nones(all_channel_ids.clone(), false);

            if branches.contains_key(&predicate) {

                // TODO what do I do with redundant predicates?

                unimplemented!(

                    "Having multiple batches with the same

                    predicate requires the support of oracle boolean variables"

                )

            }

            let branch = BranchN { to_get: gets, gotten: Default::default(), sync_batch_index };

            for (ekey, payload) in puts {

                log!(

                    &mut self.inner.logger,

                    "... ... Initial native put msg {:?} pred {:?} batch {:?}",

                    &payload,

                    &predicate,

                    sync_batch_index,

                );

                let msg =

                    CommMsgContents::SendPayload { payload_predicate: predicate.clone(), payload }

                        .into_msg(*round_index);

                endpoint_exts.get_mut(ekey).unwrap().endpoint.send(msg)?;

            }

            log!(

                &mut self.inner.logger,

                "... Initial native branch batch index={} with pred {:?}",

                sync_batch_index,

                &predicate

            );

            if branch.to_get.is_empty() {

                self.ephemeral.solution_storage.submit_and_digest_subtree_solution(

                    &mut self.inner.logger,

                    SubtreeId::PolyN,

                    predicate.clone(),

                );

            }

            branches.insert(predicate, branch);

        }

        Ok(PolyN { ekeys, branches })

    }

    pub fn sync_round(

        &mut self,

        deadline: Instant,

        sync_batches: Option<impl Iterator<Item = SyncBatch>>,

    ) -> Result<(), SyncErr> {

                }

            }

        }

    }

    // Runs a synchronous round until all the actors are in decided state OR 1+ are inconsistent.

    // If a native requires setting up, arg `sync_batches` is Some, and those are used as the sync batches.

    fn sync_round_inner(

        &mut self,

        deadline: Instant,

        sync_batches: Option<impl Iterator<Item = SyncBatch>>,

    ) -> Result<(), SyncErr> {

        assert!(self.ephemeral.is_clear());

        log!(

            &mut self.inner.logger,

            "~~~~~~~~ SYNC ROUND STARTS! ROUND={} ~~~~~~~~~",

            self.inner.round_index

        );

        // 1. Run the Mono for each Mono actor (stored in `self.mono_ps`).

        //    Some actors are dropped. some new actors are created.

        //    Ultimately, we have 0 Mono actors and a list of unnamed sync_actors

        log!(&mut self.inner.logger, "Got {} MonoP's to run!", self.inner.mono_ps.len());

        self.ephemeral.poly_ps.clear();

        // let mut poly_ps: Vec<PolyP> = vec![];

        while let Some(mut mono_p) = self.inner.mono_ps.pop() {

            let mut m_ctx = MonoPContext {

                ekeys: &mut mono_p.ekeys,

                inner: &mut self.inner,

                // endpoint_exts: &mut self.endpoint_exts,

                // mono_ps: &mut self.mono_ps,

                // channel_id_stream: &mut self.channel_id_stream,

            };

            // cross boundary into crate::protocol

            let blocker = mono_p.state.pre_sync_run(&mut m_ctx, &self.protocol_description);

            log!(&mut self.inner.logger, "... MonoP's pre_sync_run got blocker {:?}", &blocker);

            match blocker {

                MonoBlocker::Inconsistent => return Err(SyncErr::Inconsistent),

                MonoBlocker::ComponentExit => drop(mono_p),

                MonoBlocker::SyncBlockStart => self.ephemeral.poly_ps.push(mono_p.into()),

            }

        }

        log!(

            &mut self.inner.logger,

            "Finished running all MonoPs! Have {} PolyPs waiting",

            self.ephemeral.poly_ps.len()

        );

        // 3. define the mapping from ekey -> actor

        //    this is needed during the event loop to determine which actor

        //    should receive the incoming message.

        //    TODO: store and update this mapping rather than rebuilding it each round.

        let ekey_to_holder: HashMap<Key, PolyId> = {

            use PolyId::*;

            let n = self.inner.mono_n.iter().flat_map(|m| m.ekeys.iter().map(move |&e| (e, N)));

            let p = self

                .ephemeral

                .poly_ps

                .iter()

                .enumerate()

                .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));

            n.chain(p).collect()

        };

        log!(

            &mut self.inner.logger,

            "SET OF PolyPs and MonoPs final! ekey lookup map is {:?}",

            &ekey_to_holder

        );

        // 4. Create the solution storage. it tracks the solutions of "subtrees"

        //    of the controller in the overlay tree.

        self.ephemeral.solution_storage.reset({

            let n = self.inner.mono_n.iter().map(|_| SubtreeId::PolyN);

            let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });

            let c = self

                .inner

                .family

                .children_ekeys

                .iter()

                .map(|&ekey| SubtreeId::ChildController { ekey });

            let subtree_id_iter = n.chain(m).chain(c);

            log!(

                &mut self.inner.logger,

                "Solution Storage has subtree Ids: {:?}",

                &subtree_id_iter.clone().collect::<Vec<_>>()

            );

            subtree_id_iter

        });

        // 5. kick off the synchronous round of the native actor if it exists

        log!(&mut self.inner.logger, "Kicking off native's synchronous round...");

        assert_eq!(sync_batches.is_some(), self.inner.mono_n.is_some()); // TODO better err

        self.ephemeral.poly_n = if let Some(sync_batches) = sync_batches {

            // using if let because of nested ? operator

            // TODO check that there are 1+ branches or NO SOLUTION

            let poly_n = self.kick_off_native(sync_batches)?;

            log!(

                &mut self.inner.logger,

                "PolyN kicked off, and has branches with predicates... {:?}",

                poly_n.branches.keys().collect::<Vec<_>>()

            );

            Some(poly_n)

        } else {

            log!(&mut self.inner.logger, "NO NATIVE COMPONENT");

            None

        };

        // 6. Kick off the synchronous round of each protocol actor

        //    If just one actor becomes inconsistent now, there can be no solution!

        //    TODO distinguish between completed and not completed poly_p's?

        log!(&mut self.inner.logger, "Kicking off {} PolyP's.", self.ephemeral.poly_ps.len());

        for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {

            let my_subtree_id = SubtreeId::PolyP { index };

            let m_ctx = PolyPContext {

                my_subtree_id,

                inner: &mut self.inner,

                solution_storage: &mut self.ephemeral.solution_storage,

            };

            use SyncRunResult as Srr;

            let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;

            log!(&mut self.inner.logger, "... PolyP's poly_run got blocker {:?}", &blocker);

            match blocker {

                Srr::NoBranches => return Err(SyncErr::Inconsistent),

                Srr::AllBranchesComplete | Srr::BlockingForRecv => (),

            }

        }

        log!(&mut self.inner.logger, "All Poly machines have been kicked off!");

        // 7. `solution_storage` may have new solutions for this controller

        //    handle their discovery. LEADER => announce, otherwise => send to parent

        {

            let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();

            log!(

                &mut self.inner.logger,

                "Got {} controller-local solutions before a single RECV: {:?}",

                peeked.len(),

                peeked

            );

        }

        if self.handle_locals_maybe_decide()? {

            return Ok(());

        }

        // 4. Receive incoming messages until the DECISION is made

        log!(&mut self.inner.logger, "`No decision yet`. Time to recv messages");

        self.undelay_all();

        'recv_loop: loop {

            log!(&mut self.inner.logger, "`POLLING`...");

            let received = self.recv(deadline)?.ok_or_else(|| {

                log!(&mut self.inner.logger, ":( timing out");

                SyncErr::Timeout

            })?;

            log!(&mut self.inner.logger, "::: message {:?}...", &received);

            let current_content = match received.msg {

                Msg::SetupMsg(_) => {

                    // This occurs in the event the connector was malformed during connect()

                    return Err(SyncErr::UnexpectedSetupMsg);

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index < self.inner.round_index =>

                {

                    // Old message! Can safely discard

                    log!(&mut self.inner.logger, "...and its OLD! :(");

                    drop(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index > self.inner.round_index =>

                {

                    // Message from a next round. Keep for later!

                    log!(&mut self.inner.logger, "... DELAY! :(");

                    self.delay(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { contents, round_index }) => {

                    log!(

                        &mut self.inner.logger,

                        "... its a round-appropriate CommMsg with key {:?}",

                        received.recipient

                    );

                    assert_eq!(round_index, self.inner.round_index);

                    contents

                }

            };

            match current_content {

                CommMsgContents::Elaborate { partial_oracle } => {

                    // Child controller submitted a subtree solution.

                    if !self.inner.family.children_ekeys.contains(&received.recipient) {

                        return Err(SyncErr::ElaborateFromNonChild);

                    }

                    let subtree_id = SubtreeId::ChildController { ekey: received.recipient };

                    log!(

                        &mut self.inner.logger,

#[cfg(feature = "ffi")]

pub mod ffi;

mod actors;

pub(crate) mod communication;

pub(crate) mod connector;

pub(crate) mod endpoint;

pub mod errors;

// pub mod experimental;

mod serde;

pub(crate) mod setup;

pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;

pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;

use crate::common::*;

use actors::*;

use endpoint::*;

use errors::*;

#[derive(Debug, PartialEq)]

pub(crate) enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

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

pub(crate) struct Predicate {

    pub assigned: BTreeMap<ChannelId, bool>,

}

#[derive(Debug, Default)]

struct SyncBatch {

    puts: HashMap<Key, Payload>,

    gets: HashSet<Key>,

}

#[derive(Debug)]

pub enum Connector {

    Unconfigured(Unconfigured),

    Configured(Configured),

    Connected(Connected), // TODO consider boxing. currently takes up a lot of stack real estate

}

#[derive(Debug)]

pub struct Unconfigured {

    pub controller_id: ControllerId,

}

#[derive(Debug)]

pub struct Configured {

    controller_id: ControllerId,

    polarities: Vec<Polarity>,

    bindings: HashMap<usize, PortBinding>,

    protocol_description: Arc<ProtocolD>,

    main_component: Vec<u8>,

}

#[derive(Debug)]

pub struct Connected {

    native_interface: Vec<(Key, Polarity)>,

    sync_batches: Vec<SyncBatch>,

    controller: Controller,

}

#[derive(Debug, Copy, Clone)]

pub enum PortBinding {

    Native,

    Active(SocketAddr),

    Passive(SocketAddr),

}

#[derive(Debug)]

struct Arena<T> {

    storage: Vec<T>,

}

#[derive(Debug)]

struct ReceivedMsg {

    recipient: Key,

    msg: Msg,

}

#[derive(Debug)]

struct MessengerState {

    poll: Poll,

    events: Events,

    delayed: Vec<ReceivedMsg>,

    undelayed: Vec<ReceivedMsg>,

    polled_undrained: IndexSet<Key>,

}

#[derive(Debug)]

struct ChannelIdStream {

    controller_id: ControllerId,

    next_channel_index: ChannelIndex,

}

#[derive(Debug)]

enum RoundHistory {

}

#[derive(Debug)]

struct Controller {

    protocol_description: Arc<ProtocolD>,

    inner: ControllerInner,

    ephemeral: ControllerEphemeral,

    round_histories: Vec<RoundHistory>,

}

#[derive(Debug)]

struct ControllerInner {

    round_index: usize,

    channel_id_stream: ChannelIdStream,

    endpoint_exts: Arena<EndpointExt>,

    messenger_state: MessengerState,

    mono_n: Option<MonoN>,

    mono_ps: Vec<MonoP>,

    family: ControllerFamily,

    logger: String,

}

/// This structure has its state entirely reset between synchronous rounds

#[derive(Debug, Default)]

struct ControllerEphemeral {

    solution_storage: SolutionStorage,

    poly_n: Option<PolyN>,

    poly_ps: Vec<PolyP>,

    ekey_to_holder: HashMap<Key, PolyId>,

}

#[derive(Debug)]

struct ControllerFamily {

    parent_ekey: Option<Key>,

    children_ekeys: Vec<Key>,

}

#[derive(Debug)]

pub(crate) enum SyncRunResult {

    BlockingForRecv,

    AllBranchesComplete,

    NoBranches,

}

// Used to identify poly actors

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

enum PolyId {

    N,

    P { index: usize },

}

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

pub(crate) enum SubtreeId {

    PolyN,

    PolyP { index: usize },

    ChildController { ekey: Key },

}

pub(crate) struct MonoPContext<'a> {

    inner: &'a mut ControllerInner,

    ekeys: &'a mut HashSet<Key>,

}

pub(crate) struct PolyPContext<'a> {

    my_subtree_id: SubtreeId,

    inner: &'a mut ControllerInner,

    solution_storage: &'a mut SolutionStorage,

}

impl PolyPContext<'_> {

    #[inline(always)]

    fn reborrow<'a>(&'a mut self) -> PolyPContext<'a> {

        let Self { solution_storage, my_subtree_id, inner } = self;

        PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }

    }

}

struct BranchPContext<'m, 'r> {

    m_ctx: PolyPContext<'m>,

    ekeys: &'r HashSet<Key>,

    predicate: &'r Predicate,

    inbox: &'r HashMap<Key, Payload>,

}

#[derive(Default)]

pub(crate) struct SolutionStorage {

    old_local: HashSet<Predicate>,

    new_local: HashSet<Predicate>,

    // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration

    subtree_solutions: Vec<HashSet<Predicate>>,

    subtree_id_to_index: HashMap<SubtreeId, usize>,

}

trait Messengerlike {

    fn get_state_mut(&mut self) -> &mut MessengerState;

    fn get_endpoint_mut(&mut self, eekey: Key) -> &mut Endpoint;

    fn delay(&mut self, received: ReceivedMsg) {

        self.get_state_mut().delayed.push(received);

    }

    fn undelay_all(&mut self) {

        let MessengerState { delayed, undelayed, .. } = self.get_state_mut();

        undelayed.extend(delayed.drain(..))

    }

    fn send(&mut self, to: Key, msg: Msg) -> Result<(), EndpointErr> {

        self.get_endpoint_mut(to).send(msg)

    }

    // attempt to receive a message from one of the endpoints before the deadline

    fn recv(&mut self, deadline: Instant) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {

        // try get something buffered

        if let Some(x) = self.get_state_mut().undelayed.pop() {

            return Ok(Some(x));

        }

        loop {

            // polled_undrained may not be empty

            while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {

                if let Some(msg) = self

                    .get_endpoint_mut(eekey)

                    .recv()

                    .map_err(|e| MessengerRecvErr::EndpointErr(eekey, e))?

                {

                    // this endpoint MAY still have messages! check again in future

                    self.get_state_mut().polled_undrained.insert(eekey);

                    return Ok(Some(ReceivedMsg { recipient: eekey, msg }));

                }

            }

            let state = self.get_state_mut();

            match state.poll_events(deadline) {

                Ok(()) => {

                    for e in state.events.iter() {

                        state.polled_undrained.insert(Key::from_token(e.token()));

                    }

                }

                Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),

                Err(PollDeadlineErr::Timeout) => return Ok(None),

            }

        }

    }

    // attempt to receive a message from one of the endpoints before the deadline

    fn recv_until(

        &mut self,

        deadline: Option<Instant>,

    ) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {

        // try get something buffered

        if let Some(x) = self.get_state_mut().undelayed.pop() {

            return Ok(Some(x));

        }

        loop {

            // polled_undrained may not be empty

            while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {

                if let Some(msg) = self

                    .get_endpoint_mut(eekey)

                    .recv()

                    .map_err(|e| MessengerRecvErr::EndpointErr(eekey, e))?

                {

                    // this endpoint MAY still have messages! check again in future

                    self.get_state_mut().polled_undrained.insert(eekey);

                    return Ok(Some(ReceivedMsg { recipient: eekey, msg }));

                }

            }

            let state = self.get_state_mut();

            match state.poll_events_until(deadline) {

                Ok(()) => {

                    for e in state.events.iter() {

                        state.polled_undrained.insert(Key::from_token(e.token()));

                    }

                }

                Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),

                Err(PollDeadlineErr::Timeout) => return Ok(None),

            }

        }

    }

}

/////////////////////////////////

impl Debug for SolutionStorage {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        f.pad("Solutions: [")?;

        for (subtree_id, &index) in self.subtree_id_to_index.iter() {

            let sols = &self.subtree_solutions[index];

            f.write_fmt(format_args!("{:?}: {:?}, ", subtree_id, sols))?;

        }

        f.pad("]")

    }

}

impl From<EvalErr> for SyncErr {

    fn from(e: EvalErr) -> SyncErr {

        SyncErr::EvalErr(e)

    }