settled-reach/server/src/bridge/mod.rs

// Bridge module - Client-server communication
// Implements D-020 subprocess/IPC architecture
// MessagePack serialization for Rust<->Godot communication

use bevy_app::prelude::*;
use bevy_ecs::prelude::*;
use bevy_ecs::schedule::IntoScheduleConfigs;

pub mod framing;
pub mod local;
pub mod tcp;
pub mod types;
pub use types::*;

/// Error type for bridge operations
#[derive(Debug, thiserror::Error)]
pub enum BridgeError {
    #[error("serialization error: {0}")]
    Serialization(#[from] rmp_serde::encode::Error),
    #[error("deserialization error: {0}")]
    Deserialization(#[from] rmp_serde::decode::Error),
    #[error("deserialization error (raw bytes logged): {0}")]
    DeserializationWithDump(String),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("transport error: {0}")]
    Transport(String),
    #[error("client disconnected")]
    Disconnected,
    #[error("internal mutex poisoned: {0}")]
    MutexPoisoned(String),
}

/// Abstracts transport layer (D-020)
/// Implemented by LocalBridge (stdio) and future NetworkBridge
pub trait SimBridge: Send + Sync {
    /// Send an observer snapshot to the client
    fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>;

    /// Receive player inputs from the client
    fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError>;
}

/// BridgeResource: Bevy Resource wrapper for SimBridge trait object
#[derive(Resource)]
pub struct BridgeResource {
    inner: Box<dyn SimBridge>,
}

impl BridgeResource {
    pub fn new(bridge: impl SimBridge + 'static) -> Self {
        Self {
            inner: Box::new(bridge),
        }
    }

    pub fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> {
        self.inner.send_snapshot(snapshot)
    }

    pub fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
        self.inner.receive_inputs()
    }
}

/// Receive inputs from bridge and push to InputQueue
pub fn receive_bridge_inputs(
    bridge: Option<Res<BridgeResource>>,
    mut input_queue: ResMut<crate::simulation::input::InputQueue>,
    mut running: ResMut<ServerRunning>,
) {
    let Some(bridge) = bridge else { return };
    match bridge.receive_inputs() {
        Ok(inputs) => {
            for input in inputs {
                input_queue.push(input);
            }
        }
        Err(BridgeError::Disconnected) => {
            tracing::info!("Client disconnected, shutting down");
            running.0 = false;
        }
        Err(BridgeError::Io(ref e))
            if e.kind() == std::io::ErrorKind::BrokenPipe
                || e.kind() == std::io::ErrorKind::ConnectionReset =>
        {
            tracing::info!("Pipe broken, shutting down cleanly");
            running.0 = false;
        }
        Err(BridgeError::MutexPoisoned(ref msg)) => {
            tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg);
            running.0 = false;
        }
        Err(BridgeError::DeserializationWithDump(ref msg)) => {
            // Recoverable: skip this frame's input, don't shut down
            tracing::error!("Skipping malformed input frame: {}", msg);
        }
        Err(e) => {
            tracing::error!("Bridge receive error: {}", e);
        }
    }
}

/// Send snapshot from buffer to bridge.
/// Any send error is fatal — the client cannot proceed without snapshots.
pub fn send_bridge_snapshot(
    bridge: Option<Res<BridgeResource>>,
    mut buffer: ResMut<SnapshotBuffer>,
    mut running: ResMut<ServerRunning>,
) {
    let Some(bridge) = bridge else { return };
    if let Some(snapshot) = buffer.snapshot.take() {
        if let Err(e) = bridge.send_snapshot(&snapshot) {
            match &e {
                BridgeError::Disconnected => {
                    tracing::info!("Client disconnected during send, shutting down");
                }
                BridgeError::MutexPoisoned(msg) => {
                    tracing::error!("Bridge mutex poisoned during send: {}", msg);
                }
                _ => {
                    tracing::error!("Bridge send error: {}", e);
                }
            }
            running.0 = false;
        }
    }
}

/// Server running flag resource
#[derive(Resource, Debug, Clone)]
pub struct ServerRunning(pub bool);

impl Default for ServerRunning {
    fn default() -> Self {
        Self(true)
    }
}

/// Bridge plugin for client-server communication
/// Abstracts transport layer (LocalBridge/NetworkBridge)
pub struct BridgePlugin;

impl Plugin for BridgePlugin {
    fn build(&self, app: &mut App) {
        app.init_resource::<SnapshotBuffer>()
            .init_resource::<ServerRunning>()
            .add_systems(
                Update,
                (
                    receive_bridge_inputs.before(crate::simulation::input::process_player_input),
                    crate::perception::observer::compute_observer_snapshot
                        .after(crate::simulation::movement::validate_movement)
                        .before(crate::simulation::time::advance_tick),
                    crate::perception::observation::emit_observation_events
                        .after(crate::perception::observer::compute_observer_snapshot),
                    send_bridge_snapshot
                        .after(crate::perception::observer::compute_observer_snapshot)
                        .after(crate::simulation::interaction::compute_nearby_interactions),
                ),
            );
        tracing::debug!("BridgePlugin initialized");
    }
}