lavina/src/projections/irc.rs

use std::net::SocketAddr;

use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::oneshot::channel;

use crate::core::player::PlayerRegistry;
use crate::prelude::*;
use crate::protos::irc::client::{client_message, ClientMessage};
use crate::protos::irc::server::{ServerMessage, ServerMessageBody};
use crate::util::Terminator;

#[derive(Deserialize, Debug, Clone)]
pub struct ServerConfig {
    pub listen_on: SocketAddr,
    pub server_name: String,
}

async fn handle_socket(
    config: ServerConfig,
    mut stream: TcpStream,
    socket_addr: SocketAddr,
    mut players: PlayerRegistry,
    current_connections: IntGauge,
) {
    let (reader, writer) = stream.split();
    let mut reader = BufReader::new(reader);
    let mut writer = BufWriter::new(writer);

    let mut buffer = vec![];

    // let (player_id, mut player_handle) = players.create_player().await;
    // let mut sub = player_handle.subscribe().await;

    loop {
        select! {
            biased;
            res = reader.read_until(b'\n', &mut buffer) => {
                match res {
                    Ok(len) => if len == 0 {
                        log::info!("Terminating socket");
                        break;
                    },
                    Err(err) => log::warn!("Failed to read from socket: {err}"),
                }
                let parsed = client_message(&buffer[..]);
                match parsed {
                    Ok((rest, msg)) => {
                        log::info!("Incoming IRC message: {msg:?}");
                        match msg {
                            ClientMessage::Ping { token } => {
                                let response = ServerMessage {
                                    tags: vec![],
                                    sender: None,
                                    body: ServerMessageBody::Pong {
                                        from: config.server_name.as_bytes().to_vec(),
                                        token,
                                    }
                                };
                                let mut buffer = vec![];
                                response.write(&mut buffer).unwrap();
                                writer.write_all(buffer.as_slice()).await;
                                writer.flush().await;
                            },
                            _ => {},
                        }
                    },
                    Err(err) => {
                        log::warn!("Failed to parse IRC message: {err}");
                    },
                }
                buffer.clear();
            },
        }
    }
    current_connections.dec();
}

pub async fn launch(
    config: ServerConfig,
    players: PlayerRegistry,
    metrics: MetricsRegistry,
) -> Result<Terminator> {
    log::info!("Starting IRC projection");
    let (signal, mut rx) = channel();
    let current_connections =
        IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
    let total_connections = IntCounter::new(
        "irc_total_connections",
        "Total number of opened connections",
    )?;
    metrics.register(Box::new(current_connections.clone()))?;
    metrics.register(Box::new(total_connections.clone()))?;

    let listener = TcpListener::bind(config.listen_on).await?;
    log::debug!("Listener started");

    let handle = tokio::task::spawn(async move {
        loop {
            select! {
                biased;
                _ = &mut rx => break,
                new_conn = listener.accept() => {
                    match new_conn {
                        Ok((stream, socket_addr)) => {
                            let config = config.clone();
                            total_connections.inc();
                            current_connections.inc();
                            log::debug!("Incoming connection from {socket_addr}");
                            let handle = tokio::task::spawn(handle_socket(config, stream, socket_addr, players.clone(), current_connections.clone()));
                        },
                        Err(err) => log::warn!("Failed to accept new connection: {err}"),
                    }
                },
            }
        }

        log::info!("Stopping IRC projection");
        Ok(())
    });

    let terminator = Terminator::from_raw(signal, handle);
    Ok(terminator)
}