smo-server/src/server/prox.rs

use core::str;
use std::sync::Arc;

use tokio::io::AsyncWriteExt;
use tracing::{error, info, info_span, trace, warn, Instrument};
use uuid::Uuid;
use wtransport::endpoint::IncomingSession;
use xtra::{Actor, Address, Handler, Mailbox};
use zerocopy::{FromZeros, IntoBytes};

use crate::protocol::String;

use super::{listeners, ChangedStage, Manager, PlayerConnected, PlayerDisconnected, PlayerMoved, RequestState};

pub struct ProximityPlayer {
  id: Uuid,
  send: wtransport::SendStream,
  connection: Arc<wtransport::Connection>,
}

impl ProximityPlayer {
  pub fn spawn(session: IncomingSession, manager: Address<Manager>) {
    tokio::spawn(
      async move {
        trace!("proximity chat client connected");
        let connection = Arc::new(
          session.await.expect("failed to acknowledge session").accept().await.expect("failed to accept session"),
        );

        let (mut send, mut recv) = connection.accept_bi().await.expect("failed to start channel");
        trace!("getting peerjs uuid");
        let mut buffer = [0; 36];
        recv.read_exact(buffer.as_mut_bytes()).await.expect("failed to read uuid");
        let id = Uuid::parse_str(str::from_utf8(&buffer).expect("expected utf8")).expect("failed to parse uuid");
        let span = info_span!("", %id);
        span.in_scope(|| trace!("uuid parsed"));

        let state = manager.send(RequestState).await.unwrap();

        send.write_u8(state.len() as u8).await.expect("failed to write length");
        for player in state.values() {
          trace!("sending player {player:?}");
          send.write_all(player.as_bytes()).await.expect("failed to write player");
        }

        {
          let listeners = listeners().read().await;
          send.write_u32_le(listeners.len() as u32).await.expect("failed to write peer length");
          for (id, _) in listeners.iter() {
            let mut str = String::<36>::new_zeroed();
            id.as_hyphenated().encode_lower(str.as_mut_bytes());
            send.write_all(str.as_bytes()).await.expect("failed to write peer id")
          }
        }

        let (address, mailbox) = Mailbox::unbounded();

        listeners().write().await.insert(id, address.clone());
        tokio::spawn({
          let connection = connection.clone();
          async move {
            connection.closed().await;
            let _ = address.send(Stop).await;
          }
          .instrument(span.clone())
        });
        xtra::run(mailbox, ProximityPlayer { id, send, connection }).instrument(span).await;
      }
      .in_current_span(),
    );
  }
}

impl Actor for ProximityPlayer {
  type Stop = ();

  async fn started(&mut self, _: &Mailbox<Self>) -> Result<(), Self::Stop> {
    for listener in listeners().write().await.iter() {
      if *listener.0 != self.id {
        let _ = listener
          .1
          .send(PeerConnectionChanged {
            id: self.id,
            connected: true,
          })
          .detach()
          .await;
      }
    }
    Ok(())
  }
  async fn stopped(self) -> Self::Stop {
    listeners().write().await.remove(&self.id);
    for listener in listeners().write().await.iter() {
      if *listener.0 != self.id {
        let _ = listener
          .1
          .send(PeerConnectionChanged {
            id: self.id,
            connected: false,
          })
          .detach()
          .await;
      }
    }
  }
}

struct Stop;
impl Handler<Stop> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, _: Stop, ctx: &mut xtra::Context<Self>) -> Self::Return {
    info!("connection closed (stopped)");
    ctx.stop_self();
  }
}

struct PeerConnectionChanged {
  id: Uuid,
  connected: bool,
}
impl Handler<PeerConnectionChanged> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, message: PeerConnectionChanged, ctx: &mut xtra::Context<Self>) -> Self::Return {
    let mut id = String::new_zeroed();

    message.id.hyphenated().encode_lower(id.as_mut_bytes());
    let event = packet::Packet {
      kind: packet::Kind::PeerConnectionChanged,
      data: packet::PeerConnectionChanged {
        id,
        connected: message.connected,
      },
    };

    if let Err(error) = self.send.write_all(event.as_bytes()).await {
      error!("error while sending player move {error}");
      ctx.stop_self();
    }
  }
}
impl Handler<PlayerConnected> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, message: PlayerConnected, ctx: &mut xtra::Context<Self>) -> Self::Return {
    let event = packet::Packet {
      kind: packet::Kind::Connected,
      data: packet::Connected {
        id: message.id as u32,
        name: message.name,
      },
    };

    if let Err(error) = self.send.write_all(event.as_bytes()).await {
      error!("error while sending player move {error}");
      ctx.stop_self();
    }
  }
}

impl Handler<PlayerDisconnected> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, message: PlayerDisconnected, ctx: &mut xtra::Context<Self>) -> Self::Return {
    warn!("todo: implement player disconnected");
    let event = packet::Packet {
      kind: packet::Kind::Disconnected,
      data: packet::Disconnected { id: message.id as u32 },
    };

    if let Err(error) = self.send.write_all(event.as_bytes()).await {
      error!("error while sending player move {error}");
      ctx.stop_self();
    }
  }
}

impl Handler<PlayerMoved> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, message: PlayerMoved, ctx: &mut xtra::Context<Self>) -> Self::Return {
    let event = packet::Packet {
      kind: packet::Kind::Moved,
      data: packet::Moved {
        id: message.id as u32,
        position: message.position,
      },
    };

    if let Err(error) = self.connection.send_datagram(event.as_bytes()) {
      error!("error while sending player move {error}");
      ctx.stop_self();
    }
  }
}

impl Handler<ChangedStage> for ProximityPlayer {
  type Return = ();

  async fn handle(&mut self, message: ChangedStage, ctx: &mut xtra::Context<Self>) -> Self::Return {
    warn!("todo: implement changed stage");
    let event = packet::Packet {
      kind: packet::Kind::StageChanged,
      data: packet::StageChanged {
        id: message.id as u32,
        stage_name: message.stage,
      },
    };

    if let Err(error) = self.send.write_all(event.as_bytes()).await {
      error!("error while sending player move {error}");
      ctx.stop_self();
    }
  }
}

pub mod packet {
  use zerocopy::{Immutable, IntoBytes};

  use crate::{
    packet::{CLIENT_NAME_SIZE, STAGE_GAME_NAME_SIZE},
    protocol::String,
  };

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct Packet<T> {
    pub kind: Kind,
    pub data: T,
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(u8)]
  pub enum Kind {
    Connected = 0,
    Disconnected = 1,
    Moved = 2,
    StageChanged = 3,
    PeerConnectionChanged = 4,
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct Connected {
    pub id: u32,
    pub name: String<CLIENT_NAME_SIZE>,
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct Disconnected {
    pub id: u32,
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct Moved {
    pub id: u32,
    pub position: [f32; 3],
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct StageChanged {
    pub id: u32,
    pub stage_name: String<STAGE_GAME_NAME_SIZE>,
  }

  #[derive(IntoBytes, Immutable)]
  #[repr(C, packed)]
  pub struct PeerConnectionChanged {
    pub id: String<36>,
    pub connected: bool,
  }
}