quic_ecs_dt/simulator/src/main.rs

//! Manual smoke runner / load driver for the substrate.
//!
//! Parses the CLI, builds the per-device sensor layout, then drives T1
//! datagrams in the foreground while T2 and T3 emitters run as background
//! tokio tasks. Helpers live in the simulator library:
//!
//! - `simulator::profile`   — `SensorProfile`, `SensorSlot`, waveform generator
//! - `simulator::emitters`  — `run_t2_emitter`, `run_t3_emitter`, `now_us`
//! - `simulator::client`    — Quinn client + TLS trust-by-cert verifier

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use anyhow::{Context, anyhow};
use clap::{Parser, ValueEnum};
use simulator::client::SimulatorClient;
use simulator::emitters::{now_us, run_t2_emitter, run_t3_emitter};
use simulator::profile::{SensorProfile, build_slots, generate_value};
use substrate::transport::{QuicMessage, SensorType};
use tokio::time::MissedTickBehavior;
use tracing_subscriber::EnvFilter;

#[derive(Parser, Debug)]
#[command(name = "simulator", about, long_about = None)]
struct Cli {
    /// Substrate address (host:port).
    #[arg(long, default_value = "127.0.0.1:9000")]
    addr: SocketAddr,

    /// SNI name presented during the TLS handshake.
    #[arg(long, default_value = "localhost")]
    server_name: String,

    /// Path to the substrate's PEM cert; used as the exact-match trust anchor.
    #[arg(long, default_value = "certs/server.crt")]
    cert: PathBuf,

    /// Sensor mix per device.
    ///
    /// - `single` (default): one sensor per device of `--sensor-type`, on
    ///   `--sensor-id`. Lowest-cardinality, easiest to reason about.
    /// - `industrial`: five sensors per device on ids 0..4 — Temperature,
    ///   Humidity, Pressure, Voltage, Current. Lights up every dashboard
    ///   panel.
    #[arg(long, value_enum, default_value_t = SensorProfile::Single)]
    profile: SensorProfile,

    /// Sensor type for the `single` profile. Ignored by `industrial`.
    #[arg(long, value_enum, default_value_t = CliSensorType::Generic)]
    sensor_type: CliSensorType,

    /// T1 datagram rate across all (device, sensor) slots (Hz). `0` disables T1.
    #[arg(long, default_value_t = 20.0)]
    rate_hz: f64,

    /// T2 uni-stream event rate (Hz). `0` disables T2 (default).
    #[arg(long, default_value_t = 0.0)]
    t2_rate_hz: f64,

    /// T3 bidirectional command rate (Hz). `0` disables T3 (default).
    #[arg(long, default_value_t = 0.0)]
    t3_rate_hz: f64,

    /// Per-command timeout for T3 ack waits (milliseconds).
    #[arg(long, default_value_t = 2000)]
    t3_timeout_ms: u64,

    /// Number of T1 datagrams to send. `0` runs until Ctrl-C.
    #[arg(long, default_value_t = 10)]
    count: u64,

    /// Number of distinct device UUIDs to round-robin.
    #[arg(long, default_value_t = 1)]
    devices: u32,

    /// Sensor index for the `single` profile. Ignored by `industrial`.
    #[arg(long, default_value_t = 0)]
    sensor_id: u16,
}

#[derive(ValueEnum, Clone, Copy, Debug, Default)]
enum CliSensorType {
    #[default]
    Generic,
    Temperature,
    Humidity,
    Pressure,
    Voltage,
    Current,
}

impl From<CliSensorType> for SensorType {
    fn from(c: CliSensorType) -> Self {
        match c {
            CliSensorType::Generic => SensorType::Generic,
            CliSensorType::Temperature => SensorType::Temperature,
            CliSensorType::Humidity => SensorType::Humidity,
            CliSensorType::Pressure => SensorType::Pressure,
            CliSensorType::Voltage => SensorType::Voltage,
            CliSensorType::Current => SensorType::Current,
        }
    }
}

fn validate(cli: &Cli) -> anyhow::Result<()> {
    if cli.rate_hz < 0.0 {
        return Err(anyhow!("--rate-hz must be >= 0"));
    }
    if cli.t2_rate_hz < 0.0 {
        return Err(anyhow!("--t2-rate-hz must be >= 0"));
    }
    if cli.t3_rate_hz < 0.0 {
        return Err(anyhow!("--t3-rate-hz must be >= 0"));
    }
    if cli.rate_hz == 0.0 && cli.t2_rate_hz == 0.0 && cli.t3_rate_hz == 0.0 {
        return Err(anyhow!(
            "at least one of --rate-hz / --t2-rate-hz / --t3-rate-hz must be > 0"
        ));
    }
    if cli.devices == 0 {
        return Err(anyhow!("--devices must be >= 1"));
    }
    Ok(())
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();
    validate(&cli)?;

    simulator::install_crypto_provider();

    let mut slots = build_slots(
        cli.profile,
        cli.devices,
        cli.sensor_type.into(),
        cli.sensor_id,
    );

    tracing::info!(
        ?cli.addr,
        rate_hz = cli.rate_hz,
        t2_rate_hz = cli.t2_rate_hz,
        t3_rate_hz = cli.t3_rate_hz,
        count = cli.count,
        devices = cli.devices,
        slots = slots.len(),
        profile = ?cli.profile,
        "simulator launching"
    );

    let client = SimulatorClient::connect(cli.addr, &cli.server_name, &cli.cert)
        .await
        .context("connect to substrate")?;

    let interrupted = Arc::new(AtomicBool::new(false));
    {
        let flag = interrupted.clone();
        tokio::spawn(async move {
            let _ = tokio::signal::ctrl_c().await;
            tracing::info!("Ctrl-C received, draining…");
            flag.store(true, Ordering::SeqCst);
        });
    }

    // T2 / T3 emitters target slot[0] for their device/sensor identity.
    let t2_slot = slots[0].clone();
    let t3_slot = slots[0].clone();

    let t2_sent = Arc::new(AtomicU64::new(0));
    let t2_handle = if cli.t2_rate_hz > 0.0 {
        let conn = client.conn.clone();
        let rate = cli.t2_rate_hz;
        let interrupted = interrupted.clone();
        let counter = t2_sent.clone();
        Some(tokio::spawn(async move {
            run_t2_emitter(conn, t2_slot, rate, interrupted, counter).await
        }))
    } else {
        None
    };

    let t3_sent = Arc::new(AtomicU64::new(0));
    let t3_timeouts = Arc::new(AtomicU64::new(0));
    let t3_handle = if cli.t3_rate_hz > 0.0 {
        let conn = client.conn.clone();
        let rate = cli.t3_rate_hz;
        let timeout = Duration::from_millis(cli.t3_timeout_ms);
        let interrupted = interrupted.clone();
        let sent_counter = t3_sent.clone();
        let to_counter = t3_timeouts.clone();
        Some(tokio::spawn(async move {
            run_t3_emitter(
                conn,
                t3_slot,
                rate,
                timeout,
                interrupted,
                sent_counter,
                to_counter,
            )
            .await
        }))
    } else {
        None
    };

    let presence_slot_opt = slots.iter().find(|s| s.sensor_type == SensorType::Presence).cloned();
    let conn_clone = client.conn.clone();
    if let Some(presence_slot) = presence_slot_opt {
        tokio::spawn(async move {
            if let Ok(listener) = tokio::net::TcpListener::bind("0.0.0.0:9002").await {
                tracing::info!("Simulator HTTP trigger API listening on 0.0.0.0:9002");
                while let Ok((mut socket, _)) = listener.accept().await {
                    let conn = conn_clone.clone();
                    let slot = presence_slot.clone();
                    tokio::spawn(async move {
                        let mut buf = [0; 1024];
                        use tokio::io::{AsyncReadExt, AsyncWriteExt};
                        if let Ok(n) = socket.read(&mut buf).await {
                            let req = String::from_utf8_lossy(&buf[..n]);
                            if req.starts_with("OPTIONS") {
                                let res = "HTTP/1.1 204 No Content\r\nAccess-Control-Allow-Origin: *\r\nAccess-Control-Allow-Methods: POST, OPTIONS\r\n\r\n";
                                let _ = socket.write_all(res.as_bytes()).await;
                            } else if req.starts_with("POST /trigger") {
                                if let Ok(mut send) = conn.open_uni().await {
                                    let msg = QuicMessage {
                                        device_id: slot.device_id,
                                        sensor_id: slot.sensor_id,
                                        raw_value: 0.0,
                                        timestamp_us: now_us(),
                                        sequence_number: 0,
                                        sensor_type: slot.sensor_type.as_u8(),
                                    };
                                    let _ = send.write_all(&msg.to_bytes()).await;
                                    let _ = send.finish();
                                    tracing::info!("HTTP API triggered: pushed Presence=0.0 over T2");
                                }
                                let res = "HTTP/1.1 200 OK\r\nAccess-Control-Allow-Origin: *\r\n\r\nTriggered";
                                let _ = socket.write_all(res.as_bytes()).await;
                            } else {
                                let res = "HTTP/1.1 404 Not Found\r\nAccess-Control-Allow-Origin: *\r\n\r\n";
                                let _ = socket.write_all(res.as_bytes()).await;
                            }
                        }
                    });
                }
            }
        });
    }

    let started = Instant::now();
    let mut t1_sent: u64 = 0;
    let mut send_errors: u64 = 0;

    if cli.rate_hz > 0.0 {
        let period = Duration::from_nanos((1.0e9 / cli.rate_hz) as u64);
        let mut ticker = tokio::time::interval(period);
        ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);

        let unlimited = cli.count == 0;
        let mut last_progress = started;

        loop {
            ticker.tick().await;
            if interrupted.load(Ordering::SeqCst) {
                break;
            }
            if !unlimited && t1_sent >= cli.count {
                break;
            }

            let slot_idx = (t1_sent as usize) % slots.len();
            let slot = &mut slots[slot_idx];
            let msg = QuicMessage {
                device_id: slot.device_id,
                sensor_id: slot.sensor_id,
                raw_value: generate_value(slot.sensor_type, slot.seq),
                timestamp_us: now_us(),
                sequence_number: slot.seq,
                sensor_type: slot.sensor_type.as_u8(),
            };
            slot.seq = slot.seq.wrapping_add(1);
            t1_sent += 1;

            if let Err(e) = client.send_datagram(&msg) {
                send_errors += 1;
                tracing::warn!(error = %e, "send_datagram failed");
            }

            let now = Instant::now();
            if now.duration_since(last_progress) >= Duration::from_secs(1) {
                let elapsed = now.duration_since(started).as_secs_f64();
                let t1_hz = (t1_sent as f64) / elapsed.max(1e-9);
                let t2_now = t2_sent.load(Ordering::Relaxed);
                let t2_hz = (t2_now as f64) / elapsed.max(1e-9);
                let t3_now = t3_sent.load(Ordering::Relaxed);
                let t3_hz = (t3_now as f64) / elapsed.max(1e-9);
                let t3_to = t3_timeouts.load(Ordering::Relaxed);
                tracing::info!(
                    t1_sent,
                    t2_sent = t2_now,
                    t3_sent = t3_now,
                    t3_timeouts = t3_to,
                    send_errors,
                    t1_hz = format_args!("{:.1}", t1_hz),
                    t2_hz = format_args!("{:.1}", t2_hz),
                    t3_hz = format_args!("{:.1}", t3_hz),
                    "progress"
                );
                last_progress = now;
            }
        }
    } else {
        while !interrupted.load(Ordering::SeqCst) {
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    }

    interrupted.store(true, Ordering::SeqCst);
    let t2_total: u64 = match t2_handle {
        Some(h) => h.await.unwrap_or_else(|e| {
            tracing::warn!(error = %e, "T2 emitter task ended unexpectedly");
            0
        }),
        None => 0,
    };
    let (t3_total, t3_timeouts_total): (u64, u64) = match t3_handle {
        Some(h) => h.await.unwrap_or_else(|e| {
            tracing::warn!(error = %e, "T3 emitter task ended unexpectedly");
            (0, 0)
        }),
        None => (0, 0),
    };

    let elapsed = started.elapsed().as_secs_f64();
    let t1_hz = (t1_sent as f64) / elapsed.max(1e-9);
    let t2_hz = (t2_total as f64) / elapsed.max(1e-9);
    let t3_hz = (t3_total as f64) / elapsed.max(1e-9);
    tracing::info!(
        t1_sent,
        t2_sent = t2_total,
        t3_sent = t3_total,
        t3_timeouts = t3_timeouts_total,
        send_errors,
        elapsed_s = format_args!("{:.3}", elapsed),
        t1_observed_hz = format_args!("{:.1}", t1_hz),
        t2_observed_hz = format_args!("{:.1}", t2_hz),
        t3_observed_hz = format_args!("{:.1}", t3_hz),
        "simulator done"
    );

    client.close().await;
    Ok(())
}