use super::*;

use crate::common::*;

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

// Guard protecting an incrementally unfoldable slice of MapTempGuard elements

struct MapTempsGuard<'a, K, V>(&'a mut [HashMap<K, V>]);

struct MapTempGuard<'a, K, V>(&'a mut HashMap<K, V>);

#[derive(Default)]

struct GetterBuffer {

    getters_and_sends: Vec<(PortId, SendPayloadMsg)>,

}

struct RoundCtx {

    solution_storage: SolutionStorage,

    spec_var_stream: SpecVarStream,

    getter_buffer: GetterBuffer,

    deadline: Option<Instant>,

}

struct BranchingNative {

    branches: HashMap<Predicate, NativeBranch>,

}

#[derive(Clone, Debug)]

struct NativeBranch {

    index: usize,

    gotten: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

#[derive(Debug)]

struct SolutionStorage {

    old_local: HashSet<Predicate>,

struct CyclicDrainer<'a, K: Eq + Hash, V> {

    input: &'a mut HashMap<K, V>,

    inner: CyclicDrainInner<'a, K, V>,

}

struct CyclicDrainInner<'a, K: Eq + Hash, V> {

    swap: &'a mut HashMap<K, V>,

    output: &'a mut HashMap<K, V>,

}

trait ReplaceBoolTrue {

    fn replace_with_true(&mut self) -> bool;

}

impl ReplaceBoolTrue for bool {

    fn replace_with_true(&mut self) -> bool {

        let was = *self;

        *self = true;

        !was

    }

}

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

impl<'a, K, V> MapTempsGuard<'a, K, V> {

    fn reborrow(&mut self) -> MapTempsGuard<'_, K, V> {

        MapTempsGuard(self.0)

    }

        let (head, tail) = self.0.split_first_mut().expect("Cache exhausted");

        (MapTempGuard::new(head), MapTempsGuard(tail))

    }

}

impl<'a, K, V> MapTempGuard<'a, K, V> {

    fn new(map: &'a mut HashMap<K, V>) -> Self {

        Self(map)

    }

}

impl<'a, K, V> Drop for MapTempGuard<'a, K, V> {

    fn drop(&mut self) {

    }

}

impl<'a, K, V> Deref for MapTempGuard<'a, K, V> {

    type Target = HashMap<K, V>;

    fn deref(&self) -> &<Self as Deref>::Target {

        self.0

    }

}

impl<'a, K, V> DerefMut for MapTempGuard<'a, K, V> {

    fn deref_mut(&mut self) -> &mut <Self as Deref>::Target {

        self.0

    }

}

impl RoundCtxTrait for RoundCtx {

    fn get_deadline(&self) -> &Option<Instant> {

        &self.deadline

    }

    fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg) {

        self.getter_buffer.getter_add(getter, msg)

    }

}

impl Connector {

    fn get_comm_mut(&mut self) -> Option<&mut ConnectorCommunication> {

        if let ConnectorPhased::Communication(comm) = &mut self.phased {

                    } else {

                        log!(cu.logger, "Already requested failure");

                    }

                }

                None => {

                    log!(cu.logger, "No parent. Deciding on failure");

                    return Ok(Decision::Failure);

                }

            }

        }

        log!(cu.logger, "Done translating native batches into branches");

        let mut pcb_temps_owner = <[HashMap<Predicate, ProtoComponentBranch>; 3]>::default();

        let mut pcb_temps = MapTempsGuard(&mut pcb_temps_owner);

        let mut bn_temp_owner = <HashMap<Predicate, NativeBranch>>::default();

        // run all proto components to their sync blocker

        log!(

            cu.logger,

            "Running all {} proto components to their sync blocker...",

            branching_proto_components.len()

        );

        for (&proto_component_id, proto_component) in branching_proto_components.iter_mut() {

            let BranchingProtoComponent { ports, branches } = proto_component;

            let (blocked, _pcb_temps) = pcb_temps.split_first_mut();

            // initially, no components have .ended==true

            // drain from branches --> blocked

            let cd = CyclicDrainer::new(branches, swap.0, blocked.0);

            BranchingProtoComponent::drain_branches_to_blocked(

                cd,

                cu,

                rctx,

                proto_component_id,

                ports,

            )?;

            // swap the blocked branches back

            std::mem::swap(blocked.0, branches);

            if branches.is_empty() {

                log!(cu.logger, "{:?} has become inconsistent!", proto_component_id);

                if let Some(parent) = comm.neighborhood.parent {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger, "Already requested failure");

                    }

                } else {

                    log!(cu.logger, "As the leader, deciding on timeout");

                    return Ok(Decision::Failure);

                        drainer.add_input(predicate, branch);

                    }

                }

            }

            Ok(())

        })

    }

    // fn branch_merge_func(

    //     mut a: ProtoComponentBranch,

    //     b: &mut ProtoComponentBranch,

    // ) -> ProtoComponentBranch {

    //     if b.ended && !a.ended {

    //         a.ended = true;

    //         std::mem::swap(&mut a, b);

    //     }

    //     a

    // }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        proto_component_id: ProtoComponentId,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

    ) -> Result<(), UnrecoverableSyncError> {

        let logger = &mut *cu.logger;

        log!(

            logger,

            "feeding proto component {:?} getter {:?} {:?}",

            proto_component_id,

            getter,

            &send_payload_msg

        );

        let BranchingProtoComponent { branches, ports } = self;

        // partition drain from branches -> {unblocked, blocked}

        log!(logger, "visiting {} blocked branches...", branches.len());

        for (predicate, mut branch) in branches.drain() {

            if branch.ended {

                log!(logger, "Skipping ended branch with {:?}", &predicate);

                blocked.insert(predicate, branch);

                continue;

            }

            use AssignmentUnionResult as Aur;

            log!(logger, "visiting branch with pred {:?}", &predicate);

            match predicate.assignment_union(&send_payload_msg.predicate) {

                Aur::Nonexistant => {

                    // this branch does not receive the message

                    log!(logger, "skipping branch");

                    blocked.insert(predicate, branch);

                }

                Aur::Equivalent | Aur::FormerNotLatter => {

                    // retain the existing predicate, but add this payload

                    log!(logger, "feeding this branch without altering its predicate");

                    branch.feed_msg(getter, send_payload_msg.payload.clone());

                    unblocked.insert(predicate, branch);

                }

                Aur::LatterNotFormer => {

                    // fork branch, give fork the message and payload predicate. original branch untouched

            inbox: Default::default(),

            did_put_or_get: Default::default(),

            state,

            ended: false,

            untaken_choice: None,

        };

        Self { ports, branches: hashmap! { Predicate::default() => branch } }

    }

}

impl SolutionStorage {

    fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {

        let mut subtree_id_to_index: HashMap<SubtreeId, usize> = Default::default();

        let mut subtree_solutions = vec![];

        for id in subtree_ids {

            subtree_id_to_index.insert(id, subtree_solutions.len());

            subtree_solutions.push(Default::default())

        }

        Self {

            subtree_solutions,

            subtree_id_to_index,

            old_local: Default::default(),

            new_local: Default::default(),

        }

    }

    pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {

        let Self { old_local, new_local, .. } = self;

        new_local.drain().map(move |local| {

            old_local.insert(local.clone());

            local

        })

    }

    pub(crate) fn submit_and_digest_subtree_solution(

        &mut self,

        logger: &mut dyn Logger,

        subtree_id: SubtreeId,

        predicate: Predicate,

    ) {

        let index = self.subtree_id_to_index[&subtree_id];

        let left = 0..index;

        let right = (index + 1)..self.subtree_solutions.len();

        let Self { subtree_solutions, new_local, old_local, .. } = self;

        let was_new = subtree_solutions[index].insert(predicate.clone());

        if was_new {

            let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);

            Self::elaborate_into_new_local_rec(

                logger,

                predicate,

                set_visitor,

                old_local,

                new_local,

            );

        }

    }

    fn elaborate_into_new_local_rec<'a, 'b>(

        logger: &mut dyn Logger,

        partial: Predicate,

        mut set_visitor: impl Iterator<Item = &'b HashSet<Predicate>> + Clone,

        old_local: &'b HashSet<Predicate>,

        new_local: &'a mut HashSet<Predicate>,

    ) {

            c.put(p1, TEST_MSG.clone()).unwrap();

            c.sync(SEC1).unwrap_err();

        });

        s.spawn(|_| {

            // "bob"

            let pdl = b"

            primitive ac_not_b(in a, in b, out c){

                // forward A to C but keep B silent

                synchronous{ put(c, get(a)); }

            }";

            let pd = Arc::new(reowolf::ProtocolDescription::parse(pdl).unwrap());

            let mut c = file_logged_configured_connector(1, test_log_path, pd);

            let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();

            let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();

            let [a, b] = c.new_port_pair();

            c.add_component(b"ac_not_b", &[p0, p1, a]).unwrap();

            c.connect(SEC1).unwrap();

            c.get(b).unwrap();

            c.sync(SEC1).unwrap_err();

        });

    })

    .unwrap();

}