link/sidecar/src/shard.rs

//! The loopback link to the ServUO shard.
//!
//! The shard is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the
//! shard's outbound socket is the only thing that ever connects. This is the whole reason the game
//! is never directly reachable from the website — it exposes no port. See https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/link/PLAN.md §2.
//!
//! Framing is newline-delimited JSON, bidirectional: the shard sends events, we send commands. We
//! accept one shard connection at a time and re-accept when it drops (the shard reconnects on its
//! own, with a bounded backoff).

use std::sync::Arc;

use serde_json::Value;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;
use tokio::sync::{mpsc, Mutex};
use tracing::{info, warn};

/// An event line received from the shard, parsed. `kind` is lifted out for routing.
#[derive(Debug, Clone)]
pub struct ShardEvent {
    pub kind: String,
    pub value: Value,
}

/// A handle for sending command lines to the shard. Cloneable and cheap.
///
/// Commands are dropped (with a warning) when no shard is connected, rather than buffered: a
/// website query that arrives during a shard outage should fail fast and be retried, not silently
/// queue behind a reconnect. Live *events* are what must survive an outage, and those the shard
/// buffers on its side.
#[derive(Clone)]
pub struct ShardHandle {
    tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
}

impl ShardHandle {
    fn new() -> Self {
        Self {
            tx: Arc::new(Mutex::new(None)),
        }
    }

    async fn set(&self, sender: Option<mpsc::UnboundedSender<String>>) {
        *self.tx.lock().await = sender;
    }

    /// Send one command line (a complete JSON object, no newline — we add the frame delimiter).
    /// Returns false if no shard is currently connected.
    pub async fn send(&self, line: String) -> bool {
        let guard = self.tx.lock().await;
        match guard.as_ref() {
            Some(sender) => sender.send(line).is_ok(),
            None => {
                warn!("dropping command; no shard connected");
                false
            }
        }
    }

    pub async fn is_connected(&self) -> bool {
        self.tx.lock().await.is_some()
    }
}

/// Binds the loopback listener and accepts shard connections forever. Each accepted connection
/// runs until it drops, then we loop back to accept the next one. Events are forwarded to
/// `event_tx`; the returned handle sends commands to whichever shard is currently connected.
pub async fn serve(
    addr: &str,
    event_tx: mpsc::UnboundedSender<ShardEvent>,
) -> std::io::Result<ShardHandle> {
    let listener = TcpListener::bind(addr).await?;
    info!(%addr, "shard link listening");

    let handle = ShardHandle::new();
    let accept_handle = handle.clone();

    tokio::spawn(async move {
        loop {
            match listener.accept().await {
                Ok((stream, peer)) => {
                    info!(%peer, "shard connected");
                    if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await {
                        warn!(error = %e, "shard connection ended");
                    } else {
                        info!("shard disconnected");
                    }
                    accept_handle.set(None).await;
                }
                Err(e) => {
                    warn!(error = %e, "accept failed");
                    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                }
            }
        }
    });

    Ok(handle)
}

async fn handle_connection(
    stream: tokio::net::TcpStream,
    event_tx: &mpsc::UnboundedSender<ShardEvent>,
    handle: &ShardHandle,
) -> std::io::Result<()> {
    stream.set_nodelay(true).ok();
    let (read_half, mut write_half) = stream.into_split();

    // Install the outbound command channel for this connection.
    let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<String>();
    handle.set(Some(cmd_tx)).await;

    let mut reader = BufReader::new(read_half);
    let mut line = String::new();

    loop {
        tokio::select! {
            // Inbound: a line from the shard.
            result = reader.read_line(&mut line) => {
                let n = result?;
                if n == 0 {
                    return Ok(()); // clean EOF: shard closed
                }
                let trimmed = line.trim_end();
                if !trimmed.is_empty() {
                    match serde_json::from_str::<Value>(trimmed) {
                        Ok(value) => {
                            let kind = value
                                .get("kind")
                                .and_then(|k| k.as_str())
                                .unwrap_or("")
                                .to_string();
                            let _ = event_tx.send(ShardEvent { kind, value });
                        }
                        Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
                    }
                }
                line.clear();
            }
            // Outbound: a command to write to the shard.
            cmd = cmd_rx.recv() => {
                match cmd {
                    Some(mut c) => {
                        c.push('\n');
                        write_half.write_all(c.as_bytes()).await?;
                        write_half.flush().await?;
                    }
                    None => return Ok(()), // channel closed
                }
            }
        }
    }
}