use crate::common::*;

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

pub fn random_controller_id() -> ControllerId {

    type Bytes8 = [u8; std::mem::size_of::<ControllerId>()];

    let mut bytes = Bytes8::default();

    getrandom::getrandom(&mut bytes).unwrap();

    unsafe { std::mem::transmute::<Bytes8, ControllerId>(bytes) }

}

impl Default for Unconfigured {

    fn default() -> Self {

        let controller_id = random_controller_id();

        Self { controller_id }

    }

}

impl Default for Connector {

    fn default() -> Self {

        Self::Unconfigured(Unconfigured::default())

    }

}

impl Connector {

    /// Configure the Connector with the given Pdl description.

    pub fn configure(&mut self, pdl: &[u8]) -> Result<(), ConfigErr> {

        use ConfigErr::*;

        let controller_id = match self {

            Connector::Configured(_) => return Err(AlreadyConfigured),

            Connector::Connected(_) => return Err(AlreadyConnected),

            Connector::Unconfigured(Unconfigured { controller_id }) => *controller_id,

        };

        let protocol_description = Arc::new(ProtocolD::parse(pdl).map_err(ParseErr)?);

        *self = Connector::Configured(configured);

        Ok(())

    }

    /// Bind the (configured) connector's port corresponding to the

    pub fn bind_port(

        &mut self,

        proto_port_index: usize,

        binding: PortBinding,

    ) -> Result<(), PortBindErr> {

        use PortBindErr::*;

        match self {

            Connector::Unconfigured { .. } => Err(NotConfigured),

            Connector::Connected(_) => Err(AlreadyConnected),

            Connector::Configured(configured) => {

                Ok(())

            }

        }

    }

    pub fn connect(&mut self, timeout: Duration) -> Result<(), ConnectErr> {

        let deadline = Instant::now() + timeout;

        use ConnectErr::*;

        let configured = match self {

            Connector::Unconfigured { .. } => return Err(NotConfigured),

            Connector::Connected(_) => return Err(AlreadyConnected),

            Connector::Configured(configured) => configured,

        };

        // 1. Unwrap bindings or err

            .proto_maybe_bindings

            .iter()

            .copied()

            .enumerate()

            .map(|(native_index, (polarity, maybe_binding))| {

                Ok((maybe_binding.ok_or(PortNotBound { native_index })?, polarity))

            })

            .collect::<Result<Vec<(_, _)>, ConnectErr>>()?;

        let (controller, native_interface) = Controller::connect(

            configured.controller_id,

            configured.protocol_description.clone(),

            &bound_proto_interface[..],

            deadline,

        )?;

        *self = Connector::Connected(Connected {

            native_interface,

            sync_batches: vec![Default::default()],

            controller,

        });

        Ok(())

    }

    pub fn get_mut_logger(&mut self) -> Option<&mut String> {

        match self {

            Connector::Connected(connected) => Some(&mut connected.controller.inner.logger),

            _ => None,

        }

    }

    pub fn put(&mut self, native_port_index: usize, payload: Payload) -> Result<(), PortOpErr> {

        use PortOpErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        let (ekey, native_polarity) =

            *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;

        if native_polarity != Putter {

            return Err(WrongPolarity);

        }

        let sync_batch = connected.sync_batches.iter_mut().last().unwrap();

        if sync_batch.puts.contains_key(&ekey) {

            return Err(DuplicateOperation);

        }

        sync_batch.puts.insert(ekey, payload);

        Ok(())

    }

    pub fn get(&mut self, native_port_index: usize) -> Result<(), PortOpErr> {

        use PortOpErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        let (ekey, native_polarity) =

            *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;

        if native_polarity != Getter {

            return Err(WrongPolarity);

        }

        let sync_batch = connected.sync_batches.iter_mut().last().unwrap();

        if sync_batch.gets.contains(&ekey) {

            return Err(DuplicateOperation);

        }

        sync_batch.gets.insert(ekey);

        Ok(())

    }

    pub fn next_batch(&mut self) -> Result<usize, ()> {

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(()),

        };

        connected.sync_batches.push(SyncBatch::default());

        Ok(connected.sync_batches.len() - 1)

    }

    pub fn sync(&mut self, timeout: Duration) -> Result<usize, SyncErr> {

        let deadline = Instant::now() + timeout;

        use SyncErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        // do the synchronous round!

        connected.controller.sync_round(deadline, Some(connected.sync_batches.drain(..)))?;

        connected.sync_batches.push(SyncBatch::default());

        let mono_n = connected.controller.inner.mono_n.as_mut().unwrap();

        let result = mono_n.result.as_mut().unwrap();

        Ok(result.0)

    }

    pub fn read_gotten(&self, native_port_index: usize) -> Result<&[u8], ReadGottenErr> {

        use ReadGottenErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        let &(key, polarity) =

            connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;

        if polarity != Getter {

            return Err(WrongPolarity);

        }

        let mono_n = connected.controller.inner.mono_n.as_ref().expect("controller has no mono_n?");

        let result = mono_n.result.as_ref().ok_or(NoPreviousRound)?;

        let payload = result.1.get(&key).ok_or(DidntGet)?;

        Ok(payload)

    }

}

use crate::common::*;

#[derive(Debug)]

pub enum PortBindErr {

    AlreadyConnected,

    IndexOutOfBounds,

    NotConfigured,

    ParseErr,

    AlreadyConfigured,

}

#[derive(Debug)]

pub enum ReadGottenErr {

    NotConnected,

    IndexOutOfBounds,

    WrongPolarity,

    NoPreviousRound,

    DidntGet,

}

#[derive(Debug)]

pub enum PortOpErr {

    IndexOutOfBounds,

    NotConnected,

    WrongPolarity,

    DuplicateOperation,

}

#[derive(Debug)]

pub enum ConfigErr {

    AlreadyConnected,

    ParseErr(String),

    AlreadyConfigured,

}

#[derive(Debug, Clone)]

pub enum ConnectErr {

    PortNotBound { native_index: usize },

    NotConfigured,

    AlreadyConnected,

    MetaProtocolDeviation,

    Disconnected,

    PollInitFailed,

    MessengerRecvErr(MessengerRecvErr),

    Timeout,

    PollingFailed,

    PolarityMatched(SocketAddr),

    AcceptFailed(SocketAddr),

    PassiveConnectFailed(SocketAddr),

    BindFailed(SocketAddr),

}

#[derive(Debug, Clone)]

pub enum PollDeadlineErr {

    PollingFailed,

    Timeout,

}

#[derive(Debug, Clone)]

pub enum EndpointErr {

    Disconnected,

    MetaProtocolDeviation,

}

#[derive(Debug, Clone)]

pub enum SyncErr {

    NotConnected,

    MessengerRecvErr(MessengerRecvErr),

    Inconsistent,

    Timeout,

    ElaborateFromNonChild,

    AnnounceFromNonParent,

    PayloadPremiseExcludesTheChannel(ChannelId),

    UnexpectedSetupMsg,

    EndpointErr(EndpointErr),

    EvalErr(EvalErr),

}

#[derive(Debug, Clone)]

pub enum EvalErr {

    ComponentExitWhileBranching,

}

#[derive(Debug, Clone)]

pub enum MessengerRecvErr {

    PollingFailed,

    EndpointErr(EndpointErr),

}

#[cfg(feature = "ffi")]

pub mod ffi;

mod actors;

pub(crate) mod communication;

pub(crate) mod connector;

pub(crate) mod endpoint;

pub mod errors;

mod predicate; // TODO later

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

struct Controller {

    protocol_description: Arc<ProtocolD>,

    inner: ControllerInner,

    ephemeral: ControllerEphemeral,

}

#[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(Debug, 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()? {

                    // 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),

            }

        }

    }

}

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

impl From<EvalErr> for SyncErr {

    fn from(e: EvalErr) -> SyncErr {

        SyncErr::EvalErr(e)

    }

}

impl From<MessengerRecvErr> for SyncErr {

    fn from(e: MessengerRecvErr) -> SyncErr {

        SyncErr::MessengerRecvErr(e)

    }

}

impl From<MessengerRecvErr> for ConnectErr {

    fn from(e: MessengerRecvErr) -> ConnectErr {

        ConnectErr::MessengerRecvErr(e)

    }

}

impl From<EndpointErr> for MessengerRecvErr {

    fn from(e: EndpointErr) -> MessengerRecvErr {

        MessengerRecvErr::EndpointErr(e)

    }

}

impl<T> Default for Arena<T> {

    fn default() -> Self {

        Self { storage: vec![] }

    }

}

impl<T> Arena<T> {

    pub fn alloc(&mut self, t: T) -> Key {

        self.storage.push(t);

        Key::from_raw(self.storage.len() as u64 - 1)

    }

    pub fn get(&self, key: Key) -> Option<&T> {

        self.storage.get(key.to_raw() as usize)

    }

    pub fn get_mut(&mut self, key: Key) -> Option<&mut T> {

        self.storage.get_mut(key.to_raw() as usize)

    }

    pub fn type_convert<X>(self, f: impl FnMut((Key, T)) -> X) -> Arena<X> {

        Arena { storage: self.keyspace().zip(self.storage.into_iter()).map(f).collect() }

    }

    pub fn iter(&self) -> impl Iterator<Item = (Key, &T)> {

        self.keyspace().zip(self.storage.iter())

    }

    pub fn len(&self) -> usize {

        self.storage.len()

    }

    pub fn keyspace(&self) -> impl Iterator<Item = Key> {

        (0..(self.storage.len() as u64)).map(Key::from_raw)

    }

}

impl ChannelIdStream {

    fn new(controller_id: ControllerId) -> Self {

        Self { controller_id, next_channel_index: 0 }

    }

    fn next(&mut self) -> ChannelId {

        self.next_channel_index += 1;

        ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }

    }

}

impl MessengerState {

    // does NOT guarantee that events is non-empty

    fn poll_events(&mut self, deadline: Instant) -> Result<(), PollDeadlineErr> {

        use PollDeadlineErr::*;

        self.events.clear();

        let poll_timeout = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;

        self.poll.poll(&mut self.events, Some(poll_timeout)).map_err(|_| PollingFailed)?;

        Ok(())

    }

}

impl From<PollDeadlineErr> for ConnectErr {

    fn from(e: PollDeadlineErr) -> ConnectErr {

        match e {

            PollDeadlineErr::Timeout => ConnectErr::Timeout,

            PollDeadlineErr::PollingFailed => ConnectErr::PollingFailed,

        }

    }

}

impl std::ops::Not for Polarity {

    type Output = Self;

    fn not(self) -> Self::Output {

        use Polarity::*;

        match self {

            Putter => Getter,

            Getter => Putter,

        }

    }

}

impl Predicate {

    // returns true IFF self.unify would return Equivalent OR FormerNotLatter

    pub fn satisfies(&self, other: &Self) -> bool {

        let mut s_it = self.assigned.iter();

        let mut s = if let Some(s) = s_it.next() {

            s

        } else {

            return other.assigned.is_empty();

        };

        for (oid, ob) in other.assigned.iter() {

            while s.0 < oid {

                s = if let Some(s) = s_it.next() {

                    s

                } else {

                    return false;

                };

            }

            if s.0 > oid || s.1 != ob {

                return false;

            }

        }

        true

    }

    /// Given self and other, two predicates, return the most general Predicate possible, N

    /// such that n.satisfies(self) && n.satisfies(other).

    /// If none exists Nonexistant is returned.

    /// If the resulting predicate is equivlanet to self, other, or both,

    /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.

    /// otherwise New(N) is returned.

    pub fn common_satisfier(&self, other: &Self) -> CommonSatResult {

        use CommonSatResult::*;

        // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.

        let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];

        let [mut s, mut o] = [s_it.next(), o_it.next()];

        // lists of assignments in self but not other and vice versa.

        let [mut s_not_o, mut o_not_s] = [vec![], vec![]];

        loop {

            match [s, o] {

                [None, None] => break,

                [None, Some(x)] => {

                    o_not_s.push(x);

                    o_not_s.extend(o_it);

                    break;

                }

                [Some(x), None] => {

                    s_not_o.push(x);

                    s_not_o.extend(s_it);

                    break;

                }

                [Some((sid, sb)), Some((oid, ob))] => {

                    if sid < oid {

                        // o is missing this element

                        s_not_o.push((sid, sb));

                        s = s_it.next();

                    } else if sid > oid {

                        // s is missing this element

                        o_not_s.push((sid, sb));

                        o = o_it.next();

                    } else if sb != ob {

                        assert_eq!(sid, oid);

                        // both predicates assign the variable but differ on the value

                        return Nonexistant;

                    } else {

                        // both predicates assign the variable to the same value

                        s = s_it.next();

                        o = o_it.next();

                    }

                }

            }

        }

        // Observed zero inconsistencies. A unified predicate exists...

        match [s_not_o.is_empty(), o_not_s.is_empty()] {

            [true, true] => Equivalent,       // ... equivalent to both.

            [false, true] => FormerNotLatter, // ... equivalent to self.

            [true, false] => LatterNotFormer, // ... equivalent to other.

            [false, false] => {

                // ... which is the union of the predicates' assignments but

                //     is equivalent to neither self nor other.

                let mut predicate = self.clone();

                for (&id, &b) in o_not_s {

                    predicate.assigned.insert(id, b);

                }

                New(predicate)

            }

        }

    }

    pub fn iter_matching(&self, value: bool) -> impl Iterator<Item = ChannelId> + '_ {

        self.assigned

            .iter()

            .filter_map(move |(&channel_id, &b)| if b == value { Some(channel_id) } else { None })

    }

    pub fn batch_assign_nones(

        &mut self,

        channel_ids: impl Iterator<Item = ChannelId>,

        value: bool,

    ) {

        for channel_id in channel_ids {

            self.assigned.entry(channel_id).or_insert(value);

        }

    }

    pub fn replace_assignment(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {

        self.assigned.insert(channel_id, value)

    }

    pub fn union_with(&self, other: &Self) -> Option<Self> {

        let mut res = self.clone();

        for (&channel_id, &assignment_1) in other.assigned.iter() {

            match res.assigned.insert(channel_id, assignment_1) {

                Some(assignment_2) if assignment_1 != assignment_2 => return None,

                _ => {}

            }

        }

        Some(res)

    }

    pub fn query(&self, x: ChannelId) -> Option<bool> {

        self.assigned.get(&x).copied()

    }

    pub fn new_trivial() -> Self {

        Self { assigned: Default::default() }

    }

}

impl Debug for Predicate {

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

        f.pad("{")?;