Milestone 1

server_node/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "server_node"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+anyhow = "1.0.102"
+core_protocol = { version = "0.1.0", path = "../core_protocol" }
+dashmap = "6.1.0"
+futures = "0.3.32"
+tokio = { version = "1.52.1", features = ["full"] }
+tokio-serde = { version = "0.9.0", features = ["bincode"] }
+tokio-util = { version = "0.7.18", features = ["codec"] }
+tracing = "0.1.44"
+tracing-subscriber = "0.3.23"

server_node/src/main.rs

@@ -0,0 +1,43 @@
+//! Server Node entry point.
+//!
+//! This module initializes the headless relay server, binds the network listeners,
+//! and spawns isolated tasks for each connected TCP client.
+
+#![forbid(unsafe_code)]
+#![deny(clippy::all, clippy::pedantic)]
+#![deny(clippy::unwrap_used, clippy::expect_used)]
+
+mod state;
+mod tcp_router;
+
+use core_protocol::constants;
+use state::AppState;
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tracing::{error, info};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    info!("Starting server node...");
+
+    let state = Arc::new(AppState::new());
+
+    let addr = format!("0.0.0.0:{}", constants::TCP_PORT);
+    let listener = TcpListener::bind(&addr).await?;
+    info!("TCP Listener bound to {}", addr);
+
+    loop {
+        match listener.accept().await {
+            Ok((stream, _)) => {
+                let state_clone = state.clone();
+                tokio::spawn(async move {
+                    tcp_router::handle_connection(stream, state_clone).await;
+                });
+            }
+            Err(e) => {
+                error!("Failed to accept connection: {:?}", e);
+            }
+        }
+    }
+}

server_node/src/state.rs

@@ -0,0 +1,49 @@
+//! Concurrent server state management.
+//!
+//! This module defines the global application state shared across all active 
+//! Tokio tasks. It heavily relies on lock-free and fine-grained locking primitives 
+//! like `DashMap` and `AtomicU32` to ensure high performance without stuttering.
+
+use dashmap::DashMap;
+use std::sync::atomic::{AtomicU32, Ordering};
+
+/// Represents the session data for a single connected user.
+#[allow(dead_code)]
+pub struct UserState {
+    /// The client-chosen username for display in channels.
+    pub username: String,
+}
+
+/// The global application state tracking all connections and rooms.
+pub struct AppState {
+    /// A highly concurrent hash map linking active session tokens to their user data.
+    pub active_users: DashMap<u32, UserState>,
+    /// A simple atomic counter for issuing unique sequential session tokens.
+    next_token: AtomicU32,
+}
+
+impl AppState {
+    /// Creates a new, empty application state.
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            active_users: DashMap::new(),
+            next_token: AtomicU32::new(1),
+        }
+    }
+
+    /// Generates a strictly unique, monotonically increasing session token.
+    /// 
+    /// We use `Ordering::Relaxed` because we only need atomicity on the counter 
+    /// itself, not strict memory ordering with other operations.
+    #[must_use]
+    pub fn generate_token(&self) -> u32 {
+        self.next_token.fetch_add(1, Ordering::Relaxed)
+    }
+}
+
+impl Default for AppState {
+    fn default() -> Self {
+        Self::new()
+    }
+}

server_node/src/tcp_router.rs

@@ -0,0 +1,95 @@
+//! Reliable TCP routing and connection management.
+//!
+//! This module implements the reliable control lane for client communication.
+//! It handles the initial authentication handshake and routes incoming TCP events 
+//! from the length-delimited codec to the broader application state.
+
+use anyhow::{Context, Result};
+use core_protocol::tcp_events::TcpEvent;
+use futures::{SinkExt, StreamExt};
+use std::sync::Arc;
+use tokio::net::TcpStream;
+use tokio_serde::formats::Bincode;
+use tokio_serde::SymmetricallyFramed;
+use tokio_util::codec::{Framed, LengthDelimitedCodec};
+use tracing::{error, info, instrument, warn};
+
+use crate::state::AppState;
+use crate::state::UserState;
+
+/// A type alias for the heavily-nested framed stream type, combining `LengthDelimitedCodec` and `Bincode`.
+type FramedStream = SymmetricallyFramed<
+    Framed<TcpStream, LengthDelimitedCodec>,
+    TcpEvent,
+    Bincode<TcpEvent, TcpEvent>,
+>;
+
+/// Handles the lifecycle of a newly connected client's TCP stream.
+///
+/// This spans an instrumented task for the connection, setting up the necessary 
+/// framers and codecs before entering the event loop.
+#[instrument(skip(stream, state))]
+pub async fn handle_connection(stream: TcpStream, state: Arc<AppState>) {
+    let peer_addr = match stream.peer_addr() {
+        Ok(addr) => addr,
+        Err(e) => {
+            error!("Failed to get peer address: {:?}", e);
+            return;
+        }
+    };
+    info!("New connection from {}", peer_addr);
+
+    // We pad the TCP stream with a length-delimited codec to guarantee frame boundaries, 
+    // avoiding fragmentation issues common in raw TCP sockets.
+    let length_delimited = Framed::new(stream, LengthDelimitedCodec::new());
+    let mut framed: FramedStream = SymmetricallyFramed::new(
+        length_delimited,
+        Bincode::<TcpEvent, TcpEvent>::default(),
+    );
+
+    if let Err(e) = process_connection(&mut framed, state).await {
+        warn!("Connection closed with error: {:?}", e);
+    } else {
+        info!("Connection closed cleanly");
+    }
+}
+
+/// The inner event loop that processes deserialized `TcpEvent`s from the client.
+///
+/// # Errors
+/// Returns an `anyhow::Result` if deserialization fails, the connection drops unexpectedly,
+/// or a serialization error occurs when transmitting a response.
+async fn process_connection(framed: &mut FramedStream, state: Arc<AppState>) -> Result<()> {
+    while let Some(event) = framed.next().await {
+        let event = event.context("Failed to deserialize event")?;
+
+        match event {
+            TcpEvent::AuthRequest { username } => {
+                // REDACTED standard: we might log the username, but this is a reminder 
+                // for future sensitive items to use [REDACTED].
+                info!("AuthRequest received for user: {}", username);
+                
+                let session_token = state.generate_token();
+
+                state.active_users.insert(
+                    session_token,
+                    UserState {
+                        username: username.clone(),
+                    },
+                );
+
+                framed
+                    .send(TcpEvent::AuthResponse { session_token })
+                    .await
+                    .context("Failed to send AuthResponse")?;
+
+                info!("AuthResponse sent to {}", username);
+            }
+            _ => {
+                warn!("Received unhandled event before auth or unsupported event");
+            }
+        }
+    }
+
+    Ok(())
+}