use super::{ffi, AnimationRequest, Entity, EntityOwned, Physics, PlayerId, World};
use crate::{Ray3, Vec3};
use ark_api_ffi::TransparentPad;
pub use ffi::AnimationReplyType as AnimationReply;
use ffi::{
    ComponentType, ForceMode, ForceType, Space, Value, {Message, SpatialQueryOptions},
};
use std::collections::{HashMap, HashSet, VecDeque};

type AnimationRequestId = u64;

use super::ffi::v4::SpatialQueryHit as SpatialQueryHitFfi;

/// Represents an ongoing animation
///
/// Use it to request an animation and poll for a reply.

pub struct Animation {
    request_id: AnimationRequestId,
}

impl Animation {
    /// Request an animation
    ///
    /// Will put it in the global queue to be executed later
    /// when running `EntityMessenger::get().process_events()`
    pub fn enqueue_request_in_global_queue(request: &AnimationRequest) -> Self {
        let request_id = EntityMessenger::get().global_queue().animate_value(request);
        Self { request_id }
    }

    /// Request an animation
    ///
    /// Will put it in a queue of your choice to be executed on request.
    pub fn enqueue_request(request: &AnimationRequest, queue: &mut EntityMessageQueue) -> Self {
        let request_id = queue.animate_value(request);
        Self { request_id }
    }

    /// Query the state of this animation.
    pub fn result(&self) -> MessageResponse<&ffi::AnimationReplyType> {
        EntityMessenger::get().animation_reply(self.request_id)
    }
}

/// Represents a spatial query type for a spatial query request
#[derive(Copy, Clone)]
#[allow(missing_docs)]
pub enum SpatialQueryType {
    Raycast {
        direction: Vec3,
        min_distance: f32,
        max_distance: f32,
        /// If > 0.0, will cast a sphere instead of a ray
        spherecast_radius: f32,
    },
}

/// A `SpatialQueryRequest` contains all data needed to perform a spatial query
#[derive(Copy, Clone)]
pub struct SpatialQueryRequest {
    query_type: SpatialQueryType,
    position: Vec3,
    max_hits: u32,
    layer_mask: u64,
    options: SpatialQueryOptions,
    /// ignore hits with this entity.
    ignore_entity: Option<Entity>,
}

impl SpatialQueryRequest {
    /// Converts this `SpatialQueryRequest` to the corresponding ffi message struct.
    #[allow(deprecated)] // SpatialQueryCommon2
    pub fn as_message(&self, request_id: u64, hits_data_ptr: u32) -> Message {
        use ffi::*;
        match self.query_type {
            SpatialQueryType::Raycast {
                direction,
                min_distance,
                max_distance,
                spherecast_radius,
            } => Message {
                msg_type: MessageType::SpatialQuerySpherecast,
                msg_payload: MessagePayload {
                    spatial_query_spherecast: TransparentPad::new(SpatialQuerySpherecast {
                        common: SpatialQueryCommon2 {
                            id: request_id,
                            max_hits: self.max_hits,
                            hits_data_ptr,
                            layer_mask: self.layer_mask,
                            options: self.options,
                            position: self.position.into(),
                            _pad: Default::default(),
                        },
                        direction: direction.into(),
                        min_t: min_distance,
                        max_t: max_distance,
                        spherecast_radius,
                        _pad0: Default::default(),
                        _pad1: Default::default(),
                    }),
                },
                _pad: Default::default(),
            },
        }
    }
}

/// Use to build a `SpatialQueryRequest` using an ergonomic interface with sensible defaults.
#[derive(Copy, Clone)]
pub struct SpatialQueryRequestBuilder {
    request: SpatialQueryRequest,
}

impl SpatialQueryRequestBuilder {
    /// Builds a world space raycast request with the minimum and maximum distance along the ray.
    ///
    /// By default only the first hit is returned (can be changed with [`Self::with_max_hits`]).
    pub fn raycast(ray: Ray3, range: std::ops::Range<f32>) -> Self {
        Self {
            request: SpatialQueryRequest {
                query_type: SpatialQueryType::Raycast {
                    direction: ray.dir,
                    min_distance: range.start,
                    max_distance: range.end,
                    spherecast_radius: 0.0,
                },
                position: ray.origin,
                max_hits: 1,
                layer_mask: !0u64,
                options: SpatialQueryOptions::empty(),
                ignore_entity: None,
            },
        }
    }

    /// Sweep a sphere along a ray. Like a "thick" raycast.
    ///
    /// Spherecasts can currently only be performed on the physical world.
    pub fn spherecast(ray: Ray3, range: std::ops::Range<f32>, radius: f32) -> Self {
        Self {
            request: SpatialQueryRequest {
                query_type: SpatialQueryType::Raycast {
                    direction: ray.dir,
                    min_distance: range.start,
                    max_distance: range.end,
                    spherecast_radius: radius,
                },
                position: ray.origin,
                max_hits: 1,
                layer_mask: !0u64,
                options: SpatialQueryOptions::PHYSX_QUERY,
                ignore_entity: None,
            },
        }
    }

    /// Max number of hits that we want to receive.
    pub fn with_max_hits(&mut self, max_hits: usize) -> &mut Self {
        self.request.max_hits = max_hits.min(u32::MAX as usize) as u32;
        self
    }

    /// Which layers we can hit.
    pub fn with_layer_mask(&mut self, layer_mask: super::EntityLayerMask) -> &mut Self {
        self.request.layer_mask = layer_mask.value();
        self
    }

    /// Ignore hits with this entity.
    /// If you want to ignore several entities you ewill have to use a layer mask instead.
    pub fn with_ignore_entity(&mut self, entity: Entity) -> &mut Self {
        self.request.ignore_entity = Some(entity);
        self
    }

    /// Add options to the spatial query.
    pub fn with_options(&mut self, options: SpatialQueryOptions) -> &mut Self {
        self.request.options |= options;
        self
    }

    /// Builds a [`SpatialQueryRequest`].
    pub fn build(&self) -> SpatialQueryRequest {
        self.request
    }
}

impl std::ops::Deref for SpatialQueryRequestBuilder {
    type Target = SpatialQueryRequest;

    fn deref(&self) -> &Self::Target {
        &self.request
    }
}

type SpatialQueryRequestId = u64;

/// Represents an ongoing spatial query.

pub struct SpatialQuery {
    request_id: SpatialQueryRequestId,
}

impl SpatialQuery {
    /// Request a spatial query
    ///
    /// Will put it in the global queue to be executed later
    /// when running `EntityMessenger::get().process_events()`
    pub fn enqueue_request_in_global_queue(request: &SpatialQueryRequest) -> Self {
        let request_id = EntityMessenger::get()
            .global_queue()
            .spatial_query_request(request);
        Self { request_id }
    }

    /// Request a spatial query
    ///
    /// Will put it in a queue of your choice to be executed on request.
    pub fn enqueue_request(request: &SpatialQueryRequest, queue: &mut EntityMessageQueue) -> Self {
        let request_id = queue.spatial_query_request(request);
        Self { request_id }
    }

    /// Query the result of the spatial query
    pub fn result(&self) -> MessageResponse<&[SpatialQueryHit]> {
        EntityMessenger::get().spatial_query_result(self.request_id)
    }
}

/// What we return from a query
#[derive(Copy, Clone, Debug)]
#[repr(C)]
pub struct SpatialQueryHit {
    /// The entity we hit.
    pub entity: Entity,

    /// Where in the world we hit it.
    pub point: Vec3,

    /// The distance along the ray.
    ///
    /// The hits are sorted by this distance when returned from raycast and spherecast queries.
    /// In a raycast this is equal to `ray.origin.distance(hit.position)`,
    /// but in a spherecast it can be different.
    pub distance: f32,

    /// The contact normal at the hit point.
    pub normal: Vec3,
}

// Expects exact same layout so we can mem-cast between them:
static_assertions::assert_eq_size!(SpatialQueryHit, SpatialQueryHitFfi);

/// The type of a callback function to receive information about if an animation has been stopped or cancelled
pub type AnimationReplyFn<'message> = dyn FnMut(&ffi::AnimationReplyType) + 'message;

/// The type of a callback function to receive the result of a spatial query
pub type SpatialQueryResultFn<'message> = dyn FnMut(&[SpatialQueryHit]) + 'message;

/// The type of a callback function to receive collision events.
pub type CollisionEventFn<'message> = dyn FnMut(&[ffi::OnCollision]) + 'message;

/// The type of a callback function to receive trigger events.
pub type TriggerEventFn<'message> = dyn FnMut(&[ffi::OnTrigger]) + 'message;

/// The `EntityMessageDispatcher` is a hub for managing messages to / from the host
///
/// Use it to perform various requests and assign closures that will be
/// called when a result is received. You need to call `EntityMessageDispatcher::update`
/// somewhere in your update routine. This will in turn check if any results
/// have been received and call the corresponding closures.
#[derive(Default)]
pub struct EntityMessageDispatcher<'message> {
    spatial_queries: VecDeque<(SpatialQuery, Box<SpatialQueryResultFn<'message>>)>,
    animations: VecDeque<(Animation, Box<AnimationReplyFn<'message>>)>,
    collisions: Vec<(Entity, Box<CollisionEventFn<'message>>)>,
    triggers: Vec<(Entity, Box<TriggerEventFn<'message>>)>,
}

impl<'message> EntityMessageDispatcher<'message> {
    /// Remove all listenerers
    pub fn remove_all_listeners(&mut self) {
        self.spatial_queries.clear();
        self.animations.clear();
        self.collisions.clear();
        self.triggers.clear();
    }

    fn notify_destroyed(&mut self, entity: Entity) {
        // TODO: If an entity is destroyed before a reply is received
        // these could potentially be left hanging, but usually shouldn't.
        // Could be solved by either putting a timeout on them or map
        // entity to request id. I leave it for now.
        // self.animations
        // self.spatial_queries.

        self.collisions.retain(|(e, _func)| *e != entity);
        self.triggers.retain(|(e, _func)| *e != entity);
    }

    /// Registers a request into this `EntityMessageDispatcher` and puts the outgoing message
    /// into the global queue.
    pub fn spatial_query_in_global_queue<T>(&mut self, request: &SpatialQueryRequest, f: T)
    where
        T: FnMut(&[SpatialQueryHit]) + 'message,
    {
        self.spatial_queries.push_back((
            SpatialQuery::enqueue_request_in_global_queue(request),
            Box::new(f),
        ));
    }

    /// Registers a request into this `EntityMessageDispatcher` and puts the outgoing message
    /// into a queue of your choice.
    pub fn spatial_query<T>(
        &mut self,
        queue: &mut EntityMessageQueue,
        request: &SpatialQueryRequest,
        f: T,
    ) where
        T: FnMut(&[SpatialQueryHit]) + 'message,
    {
        self.spatial_queries
            .push_back((SpatialQuery::enqueue_request(request, queue), Box::new(f)));
    }

    /// Registers an animation into this `EntityMessageDispatcher` and puts the outgoing message into
    /// the global queue.
    pub fn animation_in_global_queue<T>(&mut self, request: &AnimationRequest, f: T)
    where
        T: FnMut(&ffi::AnimationReplyType) + 'message,
    {
        self.animations.push_back((
            Animation::enqueue_request_in_global_queue(request),
            Box::new(f),
        ));
    }

    /// Registers an animation into this `EntityMessageDispatcher` and puts the outgoing message into
    /// a queue of your choice.
    pub fn animation<T>(&mut self, queue: &mut EntityMessageQueue, request: &AnimationRequest, f: T)
    where
        T: FnMut(&ffi::AnimationReplyType) + 'message,
    {
        self.animations
            .push_back((Animation::enqueue_request(request, queue), Box::new(f)));
    }

    // Returns a spatial query result if there are any
    fn spatial_query_result(&mut self) -> bool {
        for (idx, (req, func)) in self.spatial_queries.iter_mut().enumerate() {
            if let MessageResponse::Ok(query_hit) =
                EntityMessenger::get().spatial_query_result(req.request_id)
            {
                func(query_hit);
                let _ = self.spatial_queries.swap_remove_front(idx);
                return true;
            }
        }

        false
    }

    // Returns an animation result if there are any
    fn animation_reply(&mut self) -> bool {
        for (idx, (req, func)) in self.animations.iter_mut().enumerate() {
            if let MessageResponse::Ok(result) =
                EntityMessenger::get().animation_reply(req.request_id)
            {
                func(result);
                let _ = self.animations.swap_remove_front(idx);
                return true;
            }
        }

        false
    }

    /// Registers a callback for listening on collisions
    pub fn on_collision<T>(&mut self, entity: Entity, f: T)
    where
        T: FnMut(&[ffi::OnCollision]) + 'message,
    {
        EntityMessenger::get().listen_collisions(entity);

        self.collisions.push((entity, Box::new(f)));
    }

    /// Registers a callback for listening on triggers
    pub fn on_trigger<T>(&mut self, entity: Entity, f: T)
    where
        T: FnMut(&[ffi::OnTrigger]) + 'message,
    {
        EntityMessenger::get().listen_triggers(entity);

        self.triggers.push((entity, Box::new(f)));
    }

    /// This needs to be executed to process replies and invoke closures.
    ///
    /// The function is responsible for polling `EntityMessenger` to see if any
    /// reply, corresponding to the requests that have been performed on this
    /// `EntityMessageDispatcher`, is received.
    pub fn update(&mut self) {
        puffin::profile_function!();

        // Make sure we cleanup all invalid callbacks before we try to execute them.
        self.cleanup();

        while self.animation_reply() {}
        while self.spatial_query_result() {}

        for (entity, func) in &mut self.collisions {
            if let Some(collisions) = EntityMessenger::get().collisions(*entity) {
                if !collisions.is_empty() {
                    func(&collisions);
                }
            }
        }

        for (entity, func) in &mut self.triggers {
            if let Some(triggers) = EntityMessenger::get().triggers(*entity) {
                if !triggers.is_empty() {
                    func(&triggers);
                }
            }
        }
    }

    fn cleanup(&mut self) {
        let world = World::instance();
        for e in world.entities_destroyed_this_and_last_frame() {
            self.notify_destroyed(*e);
        }
    }

    /// Get the global `EntityMessageDispatcher`.
    pub fn global() -> &'static mut Self {
        unsafe {
            static mut MESSENGER: Option<EntityMessageDispatcher<'static>> = None;

            if MESSENGER.is_none() {
                MESSENGER = Some(Self::default());
            }
            MESSENGER.as_mut().unwrap()
        }
    }
}

struct EntityMessageQueueState {
    message_request_count: u32, // low bits
    message_request_range: u32, // high bits

    listen_to_entities: HashMap<Entity, u32>, // contains reference count
    spatial_query_requests: HashMap<SpatialQueryRequestId, Vec<SpatialQueryHitFfi>>, // request id to max_hits of data
    collision_requests: HashSet<Entity>,
    trigger_requests: HashSet<Entity>,
    animation_requests: HashSet<AnimationRequestId>,

    global_entity: Entity,

    outbox: HashMap<Entity, Vec<Message>>,
}

impl EntityMessageQueueState {
    fn new(message_request_range: u32, global_entity: Entity) -> Self {
        Self {
            message_request_count: 0,
            message_request_range,

            listen_to_entities: HashMap::new(),
            spatial_query_requests: HashMap::new(),
            animation_requests: HashSet::new(),
            collision_requests: HashSet::new(),
            trigger_requests: HashSet::new(),

            global_entity,

            outbox: HashMap::new(),
        }
    }
}

/// Queue for entity messages

pub struct EntityMessageQueue {
    state: EntityMessageQueueState,
}

impl EntityMessageQueue {
    fn new(message_request_range: u32, global_entity: Entity) -> Self {
        Self {
            state: EntityMessageQueueState::new(message_request_range, global_entity),
        }
    }

    fn notify_destroyed(&mut self, entity: Entity) {
        if self.state.listen_to_entities.contains_key(&entity) {
            self.state.listen_to_entities.remove(&entity);

            // TODO: If an entity is destroyed before a reply is received
            // these could potentially be left hanging, but usually shouldn't.
            // Could be solved by either putting a timeout on them or map
            // entity to request id. I leave it for now.
            //self.state.spatial_query_requests;
            //self.state.animation_requests;

            self.state.collision_requests.remove(&entity);
            self.state.trigger_requests.remove(&entity);
        }
    }

    fn push_message(&mut self, entity: Entity, msg: &Message) {
        self.state
            .outbox
            .entry(entity)
            .or_insert_with(Vec::new)
            .push(*msg);
    }

    fn add_listener_to_entity(&mut self, entity: Entity) {
        *self.state.listen_to_entities.entry(entity).or_insert(0) += 1;
    }

    /// Sends a message to an entity to set a value on any of its components.
    pub fn set_value(
        &mut self,
        entity: Entity,
        component_type: ComponentType,
        param_id: u32,
        value: &Value,
    ) {
        use ffi::*;
        self.push_message(
            entity,
            &Message {
                msg_type: MessageType::SetValue,
                msg_payload: MessagePayload {
                    set_value: TransparentPad::new(SetValue {
                        component_type,
                        param_id,
                        value: *value,
                        _pad: Default::default(),
                    }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Sends a message to an entity to animate a value on any of its components.
    fn animate_value_send(&mut self, request: &AnimationRequest, request_id: u64) {
        use ffi::*;
        self.push_message(
            request.id,
            &Message {
                msg_type: if request.enqueue {
                    MessageType::AnimateValueEnqueue
                } else {
                    MessageType::AnimateValue
                },
                msg_payload: MessagePayload {
                    animate_value: TransparentPad::new(AnimateValue {
                        id: request_id,
                        param: SetValue {
                            component_type: request.component,
                            param_id: request.param,
                            value: request.target,
                            _pad: Default::default(),
                        },
                        options: request.options,
                        _pad: Default::default(),
                    }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Starts animating a value, but expect no reply.
    pub fn animate_value_no_reply(&mut self, request: &AnimationRequest) {
        use ffi::*;

        self.animate_value_send(request, ANIMATION_REQUEST_ID_NONE);
    }

    fn alloc_request_id(&mut self) -> u64 {
        let request_id = u64::from(self.state.message_request_count)
            | (u64::from(self.state.message_request_range) << 32);
        self.state.message_request_count += 1;
        request_id
    }

    /// Starts animating a value
    #[allow(clippy::too_many_arguments)]
    pub fn animate_value(&mut self, request: &AnimationRequest) -> AnimationRequestId {
        let request_id = self.alloc_request_id();

        self.animate_value_send(request, request_id);

        let _ = self.state.animation_requests.insert(request_id);

        // We expect a reply, so we add ourselves as a listener
        // for messages from this entity.
        self.add_listener_to_entity(request.id);

        request_id
    }

    /// Sends a message to an entity to stop animating a value on any of its components.
    ///
    /// Note: A `value` of None means leave it as is
    pub fn stop_animating_value(
        &mut self,
        entity: Entity,
        component_type: ComponentType,
        param_id: u32,
        value: &Value,
    ) {
        use ffi::*;
        self.push_message(
            entity,
            &Message {
                msg_type: MessageType::StopAnimatingValue,
                msg_payload: MessagePayload {
                    stop_animating_value: TransparentPad::new(SetValue {
                        component_type,
                        param_id,
                        value: *value,
                        _pad: Default::default(),
                    }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Sends a message to an entity to make a specific component the globally active one if it is a singleton (such as a camera or environment).
    ///
    /// In a multiplayer environment, `set_active` with `ComponentType::Camera` will activate the specified camera
    /// for the host player. Use `set_active_for_player` for activating a camera for a specific player.
    pub fn set_active(&mut self, entity: Entity, component_type: ComponentType) {
        use ffi::*;
        self.push_message(
            entity,
            &Message {
                msg_type: MessageType::SetActive,
                msg_payload: MessagePayload {
                    set_active: TransparentPad::new(SetActive { component_type }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Sends a message to an entity to make a specific component the active one for that player. Only works with `ComponentType::Camera`.
    ///
    /// * `player_id`: this identifier can be retrieved through the Applet API.
    pub fn set_active_for_player(
        &mut self,
        entity: Entity,
        component_type: ComponentType,
        player_id: PlayerId,
    ) {
        use ffi::*;
        self.push_message(
            entity,
            &Message {
                msg_type: MessageType::SetActiveForPlayer,
                msg_payload: MessagePayload {
                    set_active_for_player: TransparentPad::new(SetActiveForPlayer {
                        component_type,
                        player_id,
                    }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Sends a message to an entity to apply a force to its physics rigid body (if available).
    ///
    /// Both the force vector and the position the force is applied at can be either local or global,
    /// just set `force_space` and `pos_space` accordingly.
    #[allow(clippy::too_many_arguments)]
    pub fn physics_force_at_request(
        &mut self,
        entity: Entity,
        force_type: ForceType,
        force_mode: ForceMode,
        force_space: Space,
        pos_space: Space,
        vector: Vec3,
        pos: Vec3,
    ) {
        // Adding NaN forces is very bad, and this is the latest place we can check it
        // synchronously (so that the callstack shows who-dun-it).
        // We don't have macaw here, so...
        fn is_finite(v: Vec3) -> bool {
            v.x.is_finite() && v.y.is_finite() && v.z.is_finite()
        }

        assert!(is_finite(vector));
        assert!(is_finite(pos));

        use ffi::*;
        self.push_message(
            entity,
            &Message {
                msg_type: MessageType::PhysicsForceRequest2,
                msg_payload: MessagePayload {
                    physics_force_request2: TransparentPad::new(PhysicsForceRequest2 {
                        force_type,
                        force_mode,
                        force_space,
                        pos_space,
                        vector: vector.into(),
                        pos: pos.into(),
                    }),
                },
                _pad: Default::default(),
            },
        );
    }

    /// Sends a spatial query request message and return a handle that can be used to poll for a reply.
    pub fn spatial_query_request(
        &mut self,
        request: &SpatialQueryRequest,
    ) -> SpatialQueryRequestId {
        let request_id = self.alloc_request_id();

        let entity = request
            .ignore_entity
            .unwrap_or_else(|| self.get_global_entity());
        let mut hits: Vec<SpatialQueryHitFfi> =
            vec![SpatialQueryHitFfi::invalid(); request.max_hits as usize];

        self.push_message(
            entity,
            &request.as_message(request_id, hits.as_mut_ptr() as u32),
        );

        let _ = self.state.spatial_query_requests.insert(request_id, hits);

        // We expect a reply, so we add ourselves as a listener
        // for messages from this entity.
        self.add_listener_to_entity(entity);

        request_id
    }

    fn get_global_entity(&self) -> Entity {
        self.state.global_entity
    }

    /// Notifies the system that we're interested in getting collision events for this specific entity.
    pub fn listen_collisions(&mut self, entity: Entity) {
        let _ = self.state.collision_requests.insert(entity);
    }

    /// Notifies the system that we're interested in getting trigger events for this specific entity.
    pub fn listen_triggers(&mut self, entity: Entity) {
        let _ = self.state.trigger_requests.insert(entity);
    }
}

impl Drop for EntityMessageQueue {
    fn drop(&mut self) {
        let mut state = EntityMessageQueueState::new(
            self.state.message_request_range,
            self.state.global_entity,
        );
        std::mem::swap(&mut state, &mut self.state);
        EntityMessenger::get().submit_queue_state(state);
    }
}

/// Sends multiple messages
pub fn send_messages(outbox: &HashMap<Entity, Vec<Message>>) {
    for (entity, messages) in outbox {
        World::send_messages(*entity, messages);
    }
}

/// Represents the state of a message response.
#[allow(missing_docs)]
pub enum MessageResponse<T> {
    NonExistent,
    Pending,
    Empty,
    Ok(T),
}

/// Utility for submitting message queues and collecting message responses.
///
/// For this to work, you must call `EntityMessenger::get().process_messages();`
/// first thing in your `world_update`.

pub struct EntityMessenger {
    message_queue_range: u32,

    listen_to_entities: HashMap<Entity, u32>, // contains reference count
    /// true = has been received
    spatial_query_result_handlers: HashMap<SpatialQueryRequestId, (bool, Vec<SpatialQueryHitFfi>)>,
    collision_event_queues: HashMap<Entity, Vec<ffi::OnCollision>>,
    trigger_event_queues: HashMap<Entity, Vec<ffi::OnTrigger>>,
    animation_replies: HashMap<AnimationRequestId, Option<ffi::AnimationReplyType>>,

    global_entity: EntityOwned,
    global_queue: EntityMessageQueue,

    allow_sending_messages_to_dead_entities: bool,
}

impl EntityMessenger {
    fn new() -> Self {
        let entity = EntityOwned::create_empty("entity_messenger");
        let entity_handle = entity.as_entity();
        Self {
            //message_request_count: 0,
            message_queue_range: 1,

            listen_to_entities: HashMap::new(),
            spatial_query_result_handlers: HashMap::new(),
            animation_replies: HashMap::new(),
            collision_event_queues: Default::default(),
            trigger_event_queues: Default::default(),

            global_entity: entity,

            //outbox: Vec::new(),
            global_queue: EntityMessageQueue::new(0, entity_handle),

            allow_sending_messages_to_dead_entities: false,
        }
    }

    fn notify_destroyed(&mut self, entity: Entity) {
        if self.listen_to_entities.contains_key(&entity) {
            self.listen_to_entities.remove(&entity);

            // TODO: If an entity is destroyed before a reply is received
            // these could potentially be left hanging, but usually shouldn't.
            // Could be solved by either putting a timeout on them or map
            // entity to request id. I leave it for now.
            //self.state.spatial_query_result_handlers
            //self.state.animation_replies

            self.collision_event_queues.remove(&entity);
            self.trigger_event_queues.remove(&entity);
        }

        // We might have a queue in flight for which we just removed an entity.
        self.global_queue.notify_destroyed(entity);
    }

    fn add_listener_to_entity(&mut self, entity: Entity, count: u32) {
        *self.listen_to_entities.entry(entity).or_insert(0) += count;
    }

    fn remove_listener_from_entity(&mut self, entity: Entity) {
        if let Some(count) = self.listen_to_entities.get_mut(&entity) {
            if *count > 0 {
                *count -= 1;
                if *count == 0 {
                    let _ = self.listen_to_entities.remove(&entity);
                }
            }
        }
    }

    /// Poll a spatial query request to see if a result has been received.
    pub fn spatial_query_result(
        &self,
        id: SpatialQueryRequestId,
    ) -> MessageResponse<&[SpatialQueryHit]> {
        if let Some((received, hits)) = self.spatial_query_result_handlers.get(&id) {
            if *received {
                let slice = unsafe {
                    static_assertions::assert_eq_size!(SpatialQueryHit, SpatialQueryHitFfi);
                    let ptr = hits.as_slice().as_ptr().cast::<SpatialQueryHit>();
                    std::slice::from_raw_parts(ptr, hits.len())
                };
                MessageResponse::Ok(slice)
            } else {
                MessageResponse::Pending
            }
        } else {
            MessageResponse::NonExistent
        }
    }

    /// Poll an animation request to see if a reply has been received.
    pub fn animation_reply(
        &self,
        id: AnimationRequestId,
    ) -> MessageResponse<&ffi::AnimationReplyType> {
        if let Some(reply) = self.animation_replies.get(&id) {
            if let Some(reply) = reply {
                MessageResponse::Ok(reply)
            } else {
                MessageResponse::Pending
            }
        } else {
            MessageResponse::NonExistent
        }
    }

    fn setup_collision_events_mask(physics_entity: Entity, ty: &'static str) {
        if !physics_entity.is_valid() {
            log::warn!(
                "Cannot listen for {} on an invalid entity (was it destroyed?)",
                ty
            );
            return;
        }

        if !physics_entity.has::<Physics>() {
            log::warn!("Cannot listen for {} on a non-physics entity", ty);
            return;
        }

        let mask = physics_entity.physics().collision_events_mask().get();
        if mask == 0u64 {
            // If we haven't setup the mask at all, we set it to generate events for all collisions.
            physics_entity.physics().collision_events_mask().set(!0u64);
        }
    }

    /// Register this entity as a listener for collision events
    ///
    /// The entity needs a Physics component for this to work.
    /// If its collision events mask is set to 0 this function will
    /// automatically set it to listen to collisions from all layers (!0u64).
    pub fn listen_collisions(&mut self, entity: Entity) {
        Self::setup_collision_events_mask(entity, "collisions");

        if !self.collision_event_queues.contains_key(&entity) {
            self.add_listener_to_entity(entity, 1);
        }

        let _ = self
            .collision_event_queues
            .entry(entity)
            .or_insert_with(Vec::new);
    }

    /// Register this entity as a listener for trigger events
    ///
    /// The entity needs a Physics component for this to work.
    /// If its collision events mask is set to 0 this function will
    /// automatically set it to listen to collisions from all layers (!0u64).
    pub fn listen_triggers(&mut self, entity: Entity) {
        Self::setup_collision_events_mask(entity, "triggers");

        if !self.trigger_event_queues.contains_key(&entity) {
            self.add_listener_to_entity(entity, 1);
        }

        let _ = self
            .trigger_event_queues
            .entry(entity)
            .or_insert_with(Vec::new);
    }

    /// Will return a collision if is has been received, otherwise register a listener
    /// for collisions (collisions will be returned in a later invocation if they happen).
    pub fn collisions(&mut self, entity: Entity) -> Option<Vec<ffi::OnCollision>> {
        self.listen_collisions(entity);
        self.collision_event_queues.get(&entity).cloned()
    }

    /// Will return a trigger event if is has been received, otherwise register a listener
    /// for trigger events (trigger events will be returned in a later invocation if they happen).
    pub fn triggers(&mut self, entity: Entity) -> Option<Vec<ffi::OnTrigger>> {
        self.listen_triggers(entity);
        self.trigger_event_queues.get(&entity).cloned()
    }

    /// Returns a global `EntityMessageQueue` that can be used to send
    /// delayed messages from anywhere.
    pub fn global_queue(&mut self) -> &mut EntityMessageQueue {
        &mut self.global_queue
    }

    /// This can be called anytime to send away all globally pending messages,
    /// will otherwise be called from `process_messages`
    pub fn send_messages(&mut self) {
        puffin::profile_function!();
        self.global_queue = self.new_queue();
    }

    /// Submits an `EntityMessageQueue`
    ///
    /// An `EntityMessageQueue` will however automatically be
    /// submitted when it goes out of scope.
    pub fn submit_queue(&mut self, queue_to_submit: &mut EntityMessageQueue) {
        *queue_to_submit = self.new_queue();
    }

    /// Creates a new `EntityMessageQueue` that can be used
    /// for sending messages.
    pub fn new_queue(&mut self) -> EntityMessageQueue {
        let queue =
            EntityMessageQueue::new(self.message_queue_range, self.global_entity.as_entity());
        self.message_queue_range += 1;
        queue
    }

    fn submit_queue_state(&mut self, queue: EntityMessageQueueState) {
        for (entity, count) in queue.listen_to_entities {
            self.add_listener_to_entity(entity, count);
        }

        for entity in queue.collision_requests {
            self.listen_collisions(entity);
        }

        for entity in queue.trigger_requests {
            self.listen_triggers(entity);
        }

        for (request_id, hits) in queue.spatial_query_requests {
            let _ = self
                .spatial_query_result_handlers
                .insert(request_id, (false, hits));
        }

        for request_id in queue.animation_requests {
            let _ = self.animation_replies.insert(request_id, None);
        }

        let mut outbox = queue.outbox;

        if !self.allow_sending_messages_to_dead_entities {
            let world = World::instance();

            // Let's make sure we clear outboxes of destroyed entities to not
            // send messages to dead entities.
            for e in world.entities_destroyed_this_and_last_frame() {
                outbox.remove(e);
            }
        }

        send_messages(&outbox);
    }

    fn receive_messages(&self) -> Vec<(Entity, Message)> {
        puffin::profile_function!();

        self.listen_to_entities
            .keys()
            .flat_map(|entity| {
                let messages = World::retrieve_messages(*entity);
                std::iter::repeat(*entity).zip(messages.into_iter())
            })
            .collect()
    }

    /// Turn this on to allow sending messages to dead entities
    /// Useful for debugging
    pub fn set_allow_sending_messages_to_dead_entities(&mut self, allow: bool) {
        self.allow_sending_messages_to_dead_entities = allow;
    }

    fn cleanup(&mut self) {
        let world = World::instance();
        // Make sure we release all invalid maps / caches
        for e in world.entities_destroyed_this_and_last_frame() {
            self.notify_destroyed(*e);
        }
    }

    /// This needs to be called first once per frame in the world update function as
    /// it sends pending messages and processes incoming messages and builds a database
    /// of all replies gathered.
    ///
    /// These replies will be cleared every frame so one need to poll for replies
    /// every frame if one is expecting them.
    pub fn process_messages(&mut self) -> usize {
        puffin::profile_function!();
        use ffi::*;

        self.cleanup();

        let mut listeners_to_remove = vec![];

        self.send_messages();
        let incoming = self.receive_messages();

        let processed_count = incoming.len();

        // We only clear each queue, the fact that it is there means
        // that we're interested in events.
        for queue in self.collision_event_queues.values_mut() {
            queue.clear();
        }
        // We only clear each queue, the fact that it is there means
        // that we're interested in events.
        for queue in self.trigger_event_queues.values_mut() {
            queue.clear();
        }

        // Keep all spatial query results that we haven't received yet
        self.spatial_query_result_handlers
            .retain(|_request_id, (received, _entities)| !*received);

        // Keep all animation reply slots that we haven't received yet
        self.animation_replies
            .retain(|_request_id, reply| reply.is_none());

        for (entity, message) in incoming {
            match message {
                Message {
                    msg_type: MessageType::SpatialQueryHits,
                    msg_payload: payload,
                    ..
                } => {
                    let (id, num_hits) = unsafe {
                        (
                            payload.spatial_query_hits.id,
                            payload.spatial_query_hits.num_hits,
                        )
                    };
                    if let Some((received, result_data)) =
                        self.spatial_query_result_handlers.get_mut(&id)
                    {
                        *received = true;
                        result_data.resize(num_hits as usize, SpatialQueryHitFfi::invalid());
                    }
                }
                Message {
                    msg_type: MessageType::OnCollision,
                    msg_payload: payload,
                    ..
                } => {
                    let on_collision = unsafe { &payload.on_collision };
                    if let Some(queue) = self.collision_event_queues.get_mut(&entity) {
                        queue.push(*on_collision.as_ref());
                    } else {
                        let _ = self
                            .collision_event_queues
                            .insert(entity, vec![*on_collision.as_ref()]);
                    }
                }
                Message {
                    msg_type: MessageType::OnTrigger,
                    msg_payload: payload,
                    ..
                } => {
                    let on_trigger = unsafe { &payload.on_trigger };
                    if let Some(queue) = self.trigger_event_queues.get_mut(&entity) {
                        queue.push(*on_trigger.as_ref());
                    } else {
                        let _ = self
                            .trigger_event_queues
                            .insert(entity, vec![*on_trigger.as_ref()]);
                    }
                }
                Message {
                    msg_type: MessageType::AnimationReply,
                    msg_payload: payload,
                    ..
                } => {
                    let id = unsafe { payload.animation_reply.id };
                    let reply_type = unsafe { payload.animation_reply.reply_type };

                    let _ = self.animation_replies.insert(id, Some(reply_type));
                    listeners_to_remove.push(entity);
                }
                _ => {
                    log::warn!("Unknown message type {:?}", message);
                }
            }
        }

        for remove in listeners_to_remove {
            self.remove_listener_from_entity(remove);
        }

        processed_count
    }

    /// Get the `EntityMessenger` singleton.
    pub fn get() -> &'static mut Self {
        unsafe {
            static mut MESSENGER: Option<EntityMessenger> = None;

            if MESSENGER.is_none() {
                MESSENGER = Some(Self::new());
            }
            MESSENGER.as_mut().unwrap()
        }
    }
}