metaverse_core 0.2.1

use crate::environment::FetchEnvironmentEvent;
use crate::{
    capabilities::SendCapabilityRequest, inventory::RefreshInventoryEvent,
    transport::http_handler::login_to_simulator,
};
use actix::prelude::*;
use actix_rt::time;
use benthic_protocol::errors::SessionError;
use benthic_protocol::messages::ui::errors::MailboxSessionError;
#[cfg(feature = "environment")]
use benthic_protocol::session::EnvironmentCache;
#[cfg(feature = "inventory")]
use benthic_protocol::session::InventoryData;
use benthic_protocol::session::RegionData;
use benthic_protocol::{
    messages::ui::{
        errors::{CapabilityError, CircuitCodeError, CompleteAgentMovementError, FeatureError},
        login_event::Login,
        login_response::LoginResponse,
        ui_messages::{UIMessage, UIResponse},
        water_update::WaterUpdate,
    },
    session::{ServerState, Session},
};
use glam::Vec2;
use log::{error, info};
use metaverse_avatar::avatar::Avatar;
use metaverse_cache::initialize_sqlite::init_sqlite;
use metaverse_environment::land::Land;
use metaverse_messages::{
    http::capabilities::{Capability, CapabilityRequest},
    packet::packet_protocol::Packet,
    udp::{
        agent::agent_update::AgentUpdate,
        chat::chat_from_viewer::ChatFromViewer,
        core::{
            agent_throttle::{AgentThrottle, ThrottleData},
            circuit_code::CircuitCode,
            complete_agent_movement::CompleteAgentMovementData,
            complete_ping_check::CompletePingCheck,
            logout_request::LogoutRequest,
            packet_ack::PacketAck,
            region_handshake::RegionHandshake,
            region_handshake_reply::RegionHandshakeReply,
        },
    },
};
use rgb::Rgba;
use std::path::Path;
use std::{
    collections::{HashMap, HashSet},
    net::UdpSocket as SyncUdpSocket,
    path::PathBuf,
    sync::{Arc, Mutex},
    thread::sleep,
};
use tokio::{net::UdpSocket, sync::Notify, time::Duration};

/// Central Actix actor responsible for all client actix message handling within the session.
#[derive(Debug)]
pub struct Mailbox {
    /// the client socket for UDP connections
    pub client_socket: u16,
    /// UDP socket for connecting mailbox to the UI
    pub server_to_ui_socket: String,
    /// the location on disk of the inventory sqlite DB.
    pub inventory_db_location: PathBuf,
    /// queue of acks sent from the server to be responded to by the client
    pub server_acks: HashSet<u32>,
    /// queue of packet IDs sent form the core to the server that the server hasn't yet acked
    pub viewer_acks: HashSet<u32>,
    /// state of the mailbox. If it is running or not.
    pub state: Arc<Mutex<ServerState>>,
    /// notify for etablishing when it begins running
    pub notify: Arc<Notify>,
    /// Session information for after login
    pub session: Option<Session<Capability, Avatar, Land, UdpSocket>>,
    /// the global number of packets that have been sent to the UI
    pub sent_packet_count: u16,
    /// the global ping information
    pub ping_info: PingInfo,
}
impl Mailbox {
    /// Set the state of the mailbox.
    /// Determines if it's running or started or stopped.
    pub fn set_state(&mut self, new_state: ServerState, _ctx: &mut Context<Self>) {
        let state_clone: Arc<Mutex<ServerState>> = Arc::clone(&self.state);
        {
            let mut state = state_clone.lock().unwrap();
            *state = new_state.clone();
        }
        // notify on start and stop
        if new_state == ServerState::Running || new_state == ServerState::Stopped {
            self.notify.notify_one();
        }
    }
}
impl Actor for Mailbox {
    type Context = Context<Self>;
    fn started(&mut self, ctx: &mut Self::Context) {
        info!("Actix Mailbox has started");
        self.set_state(ServerState::Running, ctx);
    }
}

/// Message used for initializing the session
///
/// # Cause
/// [`HandleUIResponse`]
///
/// # Effects
/// Spawns UDP handlers and initializes session
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct StartSession {
    /// The session to start
    pub session: Session<Capability, Avatar, Land, UdpSocket>,
}
impl Handler<StartSession> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: StartSession, ctx: &mut Self::Context) -> Self::Result {
        let mut session = msg.session;
        if let Some(current_session) = self.session.as_ref() {
            session.socket = current_session.socket.clone();
        }
        self.session = Some(session);

        // if the session doesn't already have a UDP socket to watch, create one
        if let Some(session) = self.session.as_ref()
            && session.socket.is_none()
        {
            let addr = format!("{}:{}", session.local_ip, self.client_socket);

            let addr_clone = addr.clone();
            let mailbox_addr = ctx.address();

            info!("session established, starting UDP processing");

            let fut = async move {
                match UdpSocket::bind(&addr).await {
                    Ok(sock) => {
                        info!("Successfully bound to {}", addr);
                        let sock = Arc::new(sock);
                        // Spawn a new Tokio task for reading from the socket
                        tokio::spawn(Mailbox::start_udp_read(sock.clone(), mailbox_addr));
                        Ok(sock) // Return the socket wrapped in Arc
                    }
                    Err(e) => {
                        error!("Failed to bind to {}: {}", addr_clone, e);
                        Err(e)
                    }
                }
            };

            // wait for the socket to be successfully bound and then assign it
            ctx.spawn(fut.into_actor(self).map(|result, act, _| match result {
                Ok(sock) => {
                    if let Some(session) = &mut act.session {
                        session.socket = Some(sock);
                    }
                }
                Err(_) => {
                    panic!("Socket binding failed");
                }
            }));
        }
    }
}

/// Information struct for storing latency and ping info
#[derive(Debug)]
pub struct PingInfo {
    /// the number of the ping
    pub ping_number: u8,
    /// how long the latency is. Currently not doing anything.
    pub ping_latency: Duration,
    /// time of last ping
    pub last_ping: time::Instant,
}
/// Handles incoming pings from the server
///
/// # Cause
/// - Received StartPingCheck packet from UDP socket
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct HandlePing {
    /// The ID of the ping
    pub ping_id: u8,
}
impl Handler<HandlePing> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: HandlePing, ctx: &mut Self::Context) -> Self::Result {
        ctx.address().do_send(OutgoingPacket {
            packet: Packet::new_complete_ping_check(CompletePingCheck {
                ping_id: msg.ping_id,
            }),
        });
        self.ping_info.ping_latency = time::Instant::now() - self.ping_info.last_ping;
    }
}

/// Message used for acknowledging reliable headers for incoming packets.
///
/// When a UDP packet is received from the server with a reliable header, its sequence number is
/// sent back to the server in a PacketAck, so the server knows that sequence number was received.
/// Then the ack queue is cleared, to prevent sending acks for packets that have already been
/// received.
///
/// # Cause
/// - [`AddToAckList`]
///
/// # Effects
/// - Dispatches a [`PacketAck`] packet to the server
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct SendAckList {}
impl Handler<SendAckList> for Mailbox {
    type Result = ();
    fn handle(&mut self, _: SendAckList, ctx: &mut Self::Context) -> Self::Result {
        if let Some(ref session) = self.session {
            // send ack directly to the server
            if self.server_acks.is_empty() {
                return;
            }
            let packet_ids: Vec<u32> = self.server_acks.drain().collect();

            let addr = session.address.clone();
            let packet = Packet::new_packet_ack(PacketAck { packet_ids }).to_bytes();
            let sock_clone = session.socket.clone().unwrap();
            let ack_wait = async move {
                if let Err(e) = sock_clone.send_to(&packet, addr).await {
                    println!("Failed to send ack: {:?}", e)
                };
            };
            ctx.spawn(ack_wait.into_actor(self));
        }
    }
}

/// Message for updating the list of unacked packets sent from the core to the server
///
/// Messages marked as reliable sent from the core must be resent until the server replies with a
/// PacketAck message. This adds thoes packets to the ack list.
///
/// # Cause
/// - Received a reliable packet from UDP socket
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct AddToAckList {
    /// The ID of the packet to be acked
    pub id: u32,
}
impl Handler<AddToAckList> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: AddToAckList, ctx: &mut Self::Context) -> Self::Result {
        self.server_acks.insert(msg.id);
        ctx.address().do_send(SendAckList {});
    }
}

/// Message for determining if a packet should be resent
///
/// When an outgoing packet is labeled reliable, this message is used to determine if it should be
/// resent to the server. An [`OutgoingPacket`] packet is sent initially, followed by a brief
/// timeout. This allows the server enough time to respond with an ack. If an ack is not received,
/// the packet is not resent. If it isn't, the packet will be resent.
///
/// # Cause
/// - [`OutgoingPacket`] on a reliable packet
///
/// # Effect
/// - [`OutgoingPacket`] if the ack was not received
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct ResendPacket {
    /// the packet to resend
    pub packet: Packet,
}
impl Handler<ResendPacket> for Mailbox {
    type Result = ();
    fn handle(&mut self, mut msg: ResendPacket, ctx: &mut Self::Context) -> Self::Result {
        if self
            .viewer_acks
            .contains(&msg.packet.header.sequence_number)
        {
            msg.packet.header.resent = true;
            ctx.address().do_send(OutgoingPacket { packet: msg.packet });
        }
    }
}

/// Message for sending packets from the core to the server
///
/// Simply a wrapper for the packet struct to send UDP packets to the server
///
/// # Effect
/// - UDP packet sent to the server
/// - [`ResendPacket`] if the packet is marked reliable
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct OutgoingPacket {
    /// The outgoing packet
    pub packet: Packet,
}
impl Handler<OutgoingPacket> for Mailbox {
    type Result = ();
    fn handle(&mut self, mut msg: OutgoingPacket, ctx: &mut Self::Context) -> Self::Result {
        if let Some(session) = self.session.as_mut() {
            let addr = session.address.clone();
            if !msg.packet.header.resent {
                msg.packet.header.sequence_number = session.sequence_number as u32;
                session.sequence_number += 1;
            }

            let data = msg.packet.to_bytes().clone();
            let socket_clone = session.socket.as_ref().unwrap().clone();
            let fut = async move {
                if let Err(e) = socket_clone.send_to(&data, &addr).await {
                    error!("Failed to send data: {}", e);
                }
            };
            ctx.spawn(fut.into_actor(self));

            // if the header is reliable, resend the packet until the viewer_acks contains the key
            if msg.packet.header.reliable {
                self.viewer_acks.insert(msg.packet.header.sequence_number);
                // give one second for the ack to come in.
                // the ResendPacket message check if viewer_acks still contains the sequence
                // number. if it doesn't, that means it's been removed by an ack. If it does,
                // that means it should be resent with the resent flag
                ctx.notify_later(ResendPacket { packet: msg.packet }, Duration::from_secs(1));
            };
        }
    }
}

/// Message for handling region handshakes
///
/// Handles receiving region handshake data, and sending region handshake response packets.
///
/// # Cause
/// - Received a RegionHandshake packet from the UDP socket
///
/// # Effect
/// - Dispatches a [`RegionHandshakeReply`] packet to the server
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct HandleRegionHandshake {
    /// The region hanshake data
    pub region_handshake: RegionHandshake,
}
impl Handler<HandleRegionHandshake> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: HandleRegionHandshake, ctx: &mut Self::Context) -> Self::Result {
        if let Some(session) = &mut self.session {
            // set some region info
            session.region_data.water_height = msg.region_handshake.water_height;
            ctx.address().do_send(SendUIMessage {
                ui_message: UIMessage::new_water_update(WaterUpdate {
                    height: msg.region_handshake.water_height,
                    color: Rgba::from((0.0f32, 94.0, 184.0, 0.5)),
                }),
            });

            ctx.address().do_send(OutgoingPacket {
                packet: Packet::new_region_handshake_reply(RegionHandshakeReply {
                    session_id: session.session_id,
                    agent_id: session.agent_id,
                    flags: 0,
                }),
            });
        }
    }
}

/// Message for handling packet acks
///
/// Removes IDs from the outgoing packet ack queue. Received from the server to inform the core
/// that a reliable message sent from the core has successfully been received from the server
///
/// # Cause
/// - Received a PacketAck packet from the UDP socket
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct HandlePacketAck {
    /// Ack data
    pub packet_ack: PacketAck,
}
impl Handler<HandlePacketAck> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: HandlePacketAck, _ctx: &mut Self::Context) -> Self::Result {
        for id in msg.packet_ack.packet_ids {
            self.viewer_acks.remove(&id);
        }
    }
}

/// Message for receiving updates from the UI to the core
///
/// # Cause
/// - Received a UIRespones message forom the UI-Core UDP socket
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct HandleUIResponse {
    /// UI response data
    pub ui_response: UIResponse,
}
impl Handler<HandleUIResponse> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: HandleUIResponse, ctx: &mut Self::Context) -> Self::Result {
        // handle login before session is established
        if let UIResponse::Login(data) = &msg.ui_response {
            let ctx_addr = ctx.address().clone();
            let login_data = data.clone();
            let db_location = self.inventory_db_location.clone();
            actix::spawn(async move {
                if let Err(e) = handle_login(login_data, &ctx_addr, &db_location).await {
                    error!("{:?}", e);
                };
            });
        }
        // other messages require session
        if let Some(ref session) = self.session {
            match msg.ui_response {
                UIResponse::ChatFromViewer(data) => {
                    ctx.address().do_send(OutgoingPacket {
                        packet: Packet::new_chat_from_viewer(ChatFromViewer {
                            session_id: session.session_id,
                            agent_id: session.agent_id,
                            channel: data.channel,
                            message: data.message,
                            message_type: data.message_type,
                        }),
                    });
                }
                UIResponse::Logout(_) => {
                    ctx.address().do_send(OutgoingPacket {
                        packet: Packet::new_logout_request(LogoutRequest {
                            session_id: session.session_id,
                            agent_id: session.agent_id,
                        }),
                    });
                }
                UIResponse::AgentUpdate(mut data) => {
                    data.agent_id = session.agent_id;
                    data.session_id = session.session_id;
                    ctx.address().do_send(OutgoingPacket {
                        packet: Packet::new_agent_update(AgentUpdate {
                            session_id: data.session_id,
                            agent_id: data.session_id,
                            body_rotation: data.body_rotation,
                            head_rotation: data.head_rotation,
                            state: data.state,
                            camera_center: data.camera_center,
                            camera_at_axis: data.camera_at_axis,
                            camera_left_axis: data.camera_left_axis,
                            camera_up_axis: data.camera_up_axis,
                            far: data.far,
                            control_flags: data.control_flags,
                            flags: data.flags,
                        }),
                    });
                }
                data => {
                    error!("Unrecognized UIMessage: {:?}", data)
                }
            }
        }
    }
}

/// Message for sending data from the core to the UI
///
/// # Effect
/// - UDP packet sent to UI
#[derive(Debug, Message)]
#[rtype(result = "()")]
pub struct SendUIMessage {
    /// UI response data
    pub ui_message: UIMessage,
}
impl Handler<SendUIMessage> for Mailbox {
    type Result = ();
    fn handle(&mut self, msg: SendUIMessage, _: &mut Self::Context) -> Self::Result {
        let client_socket = SyncUdpSocket::bind("0.0.0.0:0").unwrap();
        if let Err(e) = client_socket.send_to(&msg.ui_message.to_bytes(), &self.server_to_ui_socket)
        {
            error!("Failed to send UI message:{:?}", e)
        }
    }
}

async fn handle_login(
    login_data: Login,
    mailbox_addr: &actix::Addr<Mailbox>,
    db_path: &Path,
) -> Result<(), SessionError> {
    let (login_response, local_ip) = match login_to_simulator(login_data).await {
        Ok((login_response, local_ip)) => {
            if let Err(e) = mailbox_addr
                .send(SendUIMessage {
                    ui_message: UIMessage::new_login_response_event(LoginResponse {
                        firstname: login_response.first_name.clone(),
                        lastname: login_response.last_name.clone(),
                        agent_id: login_response.agent_id,
                    }),
                })
                .await
            {
                error!("Failed to send login response to UI {:?}", e)
            };
            (login_response, local_ip)
        }
        Err(e) => {
            if let Err(e) = mailbox_addr
                .send(SendUIMessage {
                    ui_message: UIMessage::new_session_error(e.clone().into()),
                })
                .await
            {
                error!("Failed to send session error: {:?}", e)
            };
            Err(e)?
        }
    };

    let connection = init_sqlite(db_path.to_path_buf())
        .await
        .map_err(|e| FeatureError::Inventory(format!("Failed to initialize SQLite: {}", e)))?;

    if let Err(e) = mailbox_addr
        .send(StartSession {
            session: Session {
                inventory_db_connection: connection,
                agent_id: login_response.agent_id,
                session_id: login_response.session_id,
                address: format!("{}:{}", login_response.sim_ip, login_response.sim_port),
                seed_capability_url: login_response.seed_capability.unwrap(),
                sequence_number: 0,
                local_ip,
                capability_urls: HashMap::new(),
                region_data: RegionData {
                    region_coordinates: Vec2 {
                        x: (login_response.region_x.unwrap() as f32),
                        y: (login_response.region_y.unwrap() as f32),
                    },
                    region_id: format!("{}:{}", login_response.sim_ip, login_response.sim_port),
                    ..Default::default()
                },

                environment_cache: EnvironmentCache {
                    patch_queue: HashMap::new(),
                    patch_cache: HashMap::new(),
                },

                inventory_data: InventoryData {
                    inventory_root: login_response.inventory_root.ok_or_else(|| {
                        FeatureError::Inventory(
                            "Login response contained no inventory_root".to_string(),
                        )
                    })?,
                    inventory_lib_owner: login_response.inventory_lib_owner.ok_or_else(|| {
                        FeatureError::Inventory(
                            "Login response contained no inventory_lib_owner".to_string(),
                        )
                    })?,
                    inventory_init: false,
                },
                socket: None,

                #[cfg(feature = "avatar")]
                avatars: HashMap::new(),
            },
        })
        .await
    {
        Err(MailboxSessionError {
            message: e.to_string(),
        })?;
    };

    match CapabilityRequest::new_capability_request(vec![
        Capability::ViewerAsset,
        Capability::FetchInventoryDescendents2,
        Capability::ExtEnvironment,
    ]) {
        Ok(caps) => {
            if let Err(e) = mailbox_addr
                .send(SendCapabilityRequest {
                    capability_request: caps,
                })
                .await
            {
                Err(CapabilityError {
                    message: e.to_string(),
                })?
            }
        }
        Err(e) => {
            error!("{:?}", e)
        }
    };

    if let Err(e) = mailbox_addr
        .send(OutgoingPacket {
            packet: Packet::new_circuit_code(CircuitCode {
                code: login_response.circuit_code,
                session_id: login_response.session_id,
                id: login_response.agent_id,
            }),
        })
        .await
    {
        Err(CircuitCodeError {
            message: e.to_string(),
        })?;
    };

    sleep(Duration::from_secs(1));
    if let Err(e) = mailbox_addr
        .send(OutgoingPacket {
            packet: Packet::new_complete_agent_movement(CompleteAgentMovementData {
                circuit_code: login_response.circuit_code,
                session_id: login_response.session_id,
                agent_id: login_response.agent_id,
            }),
        })
        .await
    {
        Err(CompleteAgentMovementError {
            message: e.to_string(),
        })?;
    };
    if let Err(e) = mailbox_addr
        .send(OutgoingPacket {
            packet: Packet::new_agent_throttle(AgentThrottle {
                agent_id: login_response.agent_id,
                session_id: login_response.session_id,
                circuit_code: login_response.circuit_code,
                gen_counter: 0,
                throttles: ThrottleData {
                    ..Default::default()
                },
            }),
        })
        .await
    {
        Err(CompleteAgentMovementError {
            message: e.to_string(),
        })?;
    };

    #[cfg(feature = "inventory")]
    if let Err(e) = mailbox_addr
        .send(RefreshInventoryEvent {
            agent_id: login_response.agent_id,
        })
        .await
    {
        Err(CapabilityError {
            message: e.to_string(),
        })?
    }

    if let Err(e) = mailbox_addr.send(FetchEnvironmentEvent {}).await {
        Err(CapabilityError {
            message: e.to_string(),
        })?
    }

    Ok(())
}