/* CBindgen generated */

#ifndef REOWOLF_HEADER_DEFINED

#define REOWOLF_HEADER_DEFINED

#include <stdarg.h>

#include <stdbool.h>

#include <stdint.h>

#include <stdlib.h>

typedef enum {

  Active,

  Passive,

} EndpointPolarity;

typedef enum {

  Putter,

  Getter,

} Polarity;

typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;

typedef struct Connector Connector;

typedef int32_t ErrorCode;

typedef uint32_t ConnectorId;

typedef uint32_t PortSuffix;

typedef struct {

  ConnectorId connector_id;

  PortSuffix u32_suffix;

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a string slice for the component's identifier in the connector's configured protocol description,

 * - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,

 * the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.

 * Usable in {setup, communication} states.

 */

ErrorCode connector_add_component(Connector *connector,

                                  const uint8_t *ident_ptr,

                                  uintptr_t ident_len,

                                  const PortId *ports_ptr,

                                  uintptr_t ports_len);

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a utf-8 encoded socket address,

 * - the logical polarity of P,

 * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,

 * returns P, a port newly added to the native interface.

 */

ErrorCode connector_add_net_port(Connector *connector,

                                 PortId *port,

                                 const uint8_t *addr_str_ptr,

                                 uintptr_t addr_str_len,

                                 Polarity port_polarity,

                                 EndpointPolarity endpoint_polarity);

/**

 * Given an initialized connector in setup or connecting state,

 * - Creates a new directed port pair with logical channel putter->getter,

 * - adds the ports to the native component's interface,

 * - and returns them using the given out pointers.

 * Usable in {setup, communication} states.

 */

void connector_add_port_pair(Connector *connector, PortId *out_putter, PortId *out_getter);

/**

 * Connects this connector to the distributed system of connectors reachable through endpoints,

 * Usable in setup state, and changes the state to communication.

 */

ErrorCode connector_connect(Connector *connector, int64_t timeout_millis);

/**

 * Destroys the given a pointer to the connector on the heap, freeing its resources.

 * Usable in {setup, communication} states.

 */

void connector_destroy(Connector *connector);

ErrorCode connector_get(Connector *connector, PortId port);

/**

 * Initializes `out` with a new connector using the given protocol description as its configuration.

 * The connector uses the given (internal) connector ID.

 */

Connector *connector_new(const Arc_ProtocolDescription *pd);

Connector *connector_new_logging(const Arc_ProtocolDescription *pd,

                                 const uint8_t *path_ptr,

                                 uintptr_t path_len);

intptr_t connector_next_batch(Connector *connector);

void connector_print_debug(Connector *connector);

/**

 * Convenience function combining the functionalities of

 * "payload_new" with "connector_put_payload".

 */

ErrorCode connector_put_bytes(Connector *connector,

                              PortId port,

                              const uint8_t *bytes_ptr,

                              uintptr_t bytes_len);

intptr_t connector_sync(Connector *connector, int64_t timeout_millis);

/**

 * Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.

 */

Arc_ProtocolDescription *protocol_description_clone(const Arc_ProtocolDescription *pd);

/**

 * Destroys the given initialized protocol description and frees its resources.

 */

void protocol_description_destroy(Arc_ProtocolDescription *pd);

/**

 * Parses the utf8-encoded string slice to initialize a new protocol description object.

 * - On success, initializes `out` and returns 0

 * - On failure, stores an error string (see `reowolf_error_peek`) and returns -1

 */

Arc_ProtocolDescription *protocol_description_parse(const uint8_t *pdl, uintptr_t pdl_len);

/**

 * Returns length (via out pointer) and pointer (via return value) of the last Reowolf error.

 * - pointer is NULL iff there was no last error

 * - data at pointer is null-delimited

 * - len does NOT include the length of the null-delimiter

 * If len is NULL, it will not written to.

 */

const uint8_t *reowolf_error_peek(uintptr_t *len);

#endif /* REOWOLF_HEADER_DEFINED */

mod communication;

mod endpoints;

pub mod error;

mod logging;

mod setup;

#[cfg(feature = "ffi")]

pub mod ffi;

#[cfg(test)]

mod tests;

use crate::common::*;

use error::*;

#[derive(Debug)]

pub struct Connector {

    unphased: ConnectorUnphased,

    phased: ConnectorPhased,

}

pub trait Logger: Debug {

    fn line_writer(&mut self) -> &mut dyn std::io::Write;

}

#[derive(Debug)]

pub struct VecLogger(ConnectorId, Vec<u8>);

#[derive(Debug)]

pub struct DummyLogger;

#[derive(Debug)]

pub struct FileLogger(ConnectorId, std::fs::File);

pub(crate) struct NonsyncProtoContext<'a> {

    logger: &'a mut dyn Logger,

    proto_component_id: ProtoComponentId,

    port_info: &'a mut PortInfo,

    id_manager: &'a mut IdManager,

    proto_component_ports: &'a mut HashSet<PortId>,

    unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,

}

pub(crate) struct SyncProtoContext<'a> {

    logger: &'a mut dyn Logger,

    predicate: &'a Predicate,

    port_info: &'a PortInfo,

    inbox: &'a HashMap<PortId, Payload>,

}

#[derive(Debug)]

struct RoundOk {

    batch_index: usize,

    gotten: HashMap<PortId, Payload>,

}

struct VecSet<T: std::cmp::Ord> {

    // invariant: ordered, deduplicated

    vec: Vec<T>,

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

enum ComponentId {

    Native,

    Proto(ProtoComponentId),

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

enum Route {

    LocalComponent(ComponentId),

    Endpoint { index: usize },

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

enum Decision {

    Failure,

    Success(Predicate),

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

enum Msg {

    SetupMsg(SetupMsg),

    CommMsg(CommMsg),

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

enum SetupMsg {

    MyPortInfo(MyPortInfo),

    LeaderWave { wave_leader: ConnectorId },

    LeaderAnnounce { tree_leader: ConnectorId },

    YouAreMyParent,

    SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },

    SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    port_info: PortInfo,

    proto_components: HashMap<ProtoComponentId, ProtoComponent>,

}

#[derive(Debug, Clone)]

struct SerdeProtocolDescription(Arc<ProtocolDescription>);

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct CommMsg {

    round_index: usize,

    contents: CommMsgContents,

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

enum CommMsgContents {

    SendPayload(SendPayloadMsg),

    Suggest { suggestion: Decision }, // SINKWARD

    Announce { decision: Decision },  // SINKAWAYS

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct SendPayloadMsg {

    predicate: Predicate,

    payload: Payload,

}

#[derive(Debug, PartialEq)]

enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

struct Endpoint {

    inbox: Vec<u8>,

    stream: TcpStream,

}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

struct ProtoComponent {

    state: ComponentState,

    ports: HashSet<PortId>,

}

#[derive(Debug, Clone)]

struct EndpointSetup {

    sock_addr: SocketAddr,

    endpoint_polarity: EndpointPolarity,

}

#[derive(Debug)]

struct EndpointExt {

    endpoint: Endpoint,

    getter_for_incoming: PortId,

}

#[derive(Debug)]

struct Neighborhood {

    parent: Option<usize>,

    children: VecSet<usize>,

}

#[derive(Debug)]

struct IdManager {

    connector_id: ConnectorId,

    port_suffix_stream: U32Stream,

    proto_component_suffix_stream: U32Stream,

}

#[derive(Debug)]

struct EndpointManager {

    // invariants:

    // 1. endpoint N is registered READ | WRITE with poller

    // 2. Events is empty

    poll: Poll,

    events: Events,

    polled_undrained: IndexSet<usize>,

    delayed_messages: Vec<(usize, Msg)>,

    undelayed_messages: Vec<(usize, Msg)>,

    endpoint_exts: Vec<EndpointExt>,

}

#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]

struct PortInfo {

    polarities: HashMap<PortId, Polarity>,

    peers: HashMap<PortId, PortId>,

    routes: HashMap<PortId, Route>,

}

#[derive(Debug)]

struct ConnectorCommunication {

    round_index: usize,

    endpoint_manager: EndpointManager,

    neighborhood: Neighborhood,

    native_batches: Vec<NativeBatch>,

    round_result: Result<Option<RoundOk>, SyncError>,

}

#[derive(Debug)]

struct ConnectorUnphased {

    proto_description: Arc<ProtocolDescription>,

    proto_components: HashMap<ProtoComponentId, ProtoComponent>,

    logger: Box<dyn Logger>,

    id_manager: IdManager,

    native_ports: HashSet<PortId>,

    port_info: PortInfo,

}

#[derive(Debug)]

enum ConnectorPhased {

    Setup { endpoint_setups: Vec<(PortId, EndpointSetup)>, surplus_sockets: u16 },

    Communication(Box<ConnectorCommunication>),

}

#[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

struct Predicate {

    assigned: BTreeMap<FiringVar, bool>,

}

#[derive(Debug, Default)]

struct NativeBatch {

    // invariant: putters' and getters' polarities respected

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

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

pub fn would_block(err: &std::io::Error) -> bool {

    err.kind() == std::io::ErrorKind::WouldBlock

}

impl<T: std::cmp::Ord> VecSet<T> {

    fn new(mut vec: Vec<T>) -> Self {

        vec.sort();

        vec.dedup();

        Self { vec }

    }

    fn contains(&self, element: &T) -> bool {

        self.vec.binary_search(element).is_ok()

    }

    // fn insert(&mut self, element: T) -> bool {

    //     match self.vec.binary_search(&element) {

    //         Ok(_) => false,

    //         Err(index) => {

    //             self.vec.insert(index, element);

    //             true

    //         }

    //     }

    // }

    fn iter(&self) -> std::slice::Iter<T> {

        self.vec.iter()

    }

}

impl PortInfo {

    fn firing_var_for(&self, port: PortId) -> FiringVar {

        FiringVar(match self.polarities.get(&port).unwrap() {

            Getter => port,

            Putter => *self.peers.get(&port).unwrap(),

        })

    }

}

impl IdManager {

    fn new(connector_id: ConnectorId) -> Self {

        Self {

            connector_id,

            port_suffix_stream: Default::default(),

            proto_component_suffix_stream: Default::default(),

        }

    }

    fn new_port_id(&mut self) -> PortId {

        Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }.into()

    }

    fn new_proto_component_id(&mut self) -> ProtoComponentId {

        Id {

            connector_id: self.connector_id,

            u32_suffix: self.proto_component_suffix_stream.next(),

        }

        .into()

    }

}

impl Drop for Connector {

    fn drop(&mut self) {

        log!(&mut *self.unphased.logger, "Connector dropping. Goodbye!");

    }

}

impl Connector {

    fn random_id() -> ConnectorId {

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

        unsafe {

            let mut bytes = std::mem::MaybeUninit::<Bytes8>::uninit();

            // getrandom is the canonical crate for a small, secure rng

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

            // safe! representations of all valid Byte8 values are valid ConnectorId values

            std::mem::transmute::<_, _>(bytes.assume_init())

        }

    }

    pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {

        std::mem::swap(&mut self.unphased.logger, &mut new_logger);

        new_logger

    }

    pub fn get_logger(&mut self) -> &mut dyn Logger {

        &mut *self.unphased.logger

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        let cu = &mut self.unphased;

        // adds two new associated ports, related to each other, and exposed to the native

        let [o, i] = [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];

        cu.native_ports.insert(o);

        cu.native_ports.insert(i);

        // {polarity, peer, route} known. {} unknown.

        cu.port_info.polarities.insert(o, Putter);

        cu.port_info.polarities.insert(i, Getter);

        cu.port_info.peers.insert(o, i);

        cu.port_info.peers.insert(i, o);

        let route = Route::LocalComponent(ComponentId::Native);

        cu.port_info.routes.insert(o, route);

        cu.port_info.routes.insert(i, route);

        log!(cu.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);

        [o, i]

    }

    pub fn add_component(

        &mut self,

        identifier: &[u8],

        ports: &[PortId],

    ) -> Result<(), AddComponentError> {

        // called by the USER. moves ports owned by the NATIVE

        use AddComponentError::*;

        // 1. check if this is OK

        let cu = &mut self.unphased;

        let polarities = cu.proto_description.component_polarities(identifier)?;

        if polarities.len() != ports.len() {

            return Err(WrongNumberOfParamaters { expected: polarities.len() });

        }

        for (&expected_polarity, port) in polarities.iter().zip(ports.iter()) {

            if !cu.native_ports.contains(port) {

                return Err(UnknownPort(*port));

            }

            if expected_polarity != *cu.port_info.polarities.get(port).unwrap() {

                return Err(WrongPortPolarity { port: *port, expected_polarity });

            }

        }

        // 3. remove ports from old component & update port->route

        let new_id = cu.id_manager.new_proto_component_id();

        for port in ports.iter() {

            cu.port_info.routes.insert(*port, Route::LocalComponent(ComponentId::Proto(new_id)));

        }

        cu.native_ports.retain(|port| !ports.contains(port));

        // 4. add new component

        cu.proto_components.insert(

            new_id,

            ProtoComponent {

                state: cu.proto_description.new_main_component(identifier, ports),

                ports: ports.iter().copied().collect(),

            },

        );

        Ok(())

    }

}

impl Predicate {

    #[inline]

    pub fn inserted(mut self, k: FiringVar, v: bool) -> Self {

        self.assigned.insert(k, v);

        self

    }

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

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

        use CommonSatResult as Csr;

        // 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((oid, ob));

                        o = o_it.next();

                    } else if sb != ob {

                        assert_eq!(sid, oid);

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

                        return Csr::Nonexistant;

                    } else {

                        // both predicates assign the variable to the same value

                        s = s_it.next();