harmony/network_stress_test/src/web.rs

use std::sync::Arc;

use actix_web::{HttpResponse, Responder, get, web};
use log::debug;
use tokio::sync::broadcast;

use crate::db;
use crate::engine::{ChaosCoordinator, DashboardMessage};

pub struct AppState {
    pub coordinator: Arc<ChaosCoordinator>,
}

#[get("/")]
async fn index() -> impl Responder {
    HttpResponse::Ok()
        .content_type("text/html; charset=utf-8")
        .body(include_str!("../templates/dashboard.html"))
}

#[get("/api/status")]
async fn status(data: web::Data<AppState>) -> impl Responder {
    let status = data.coordinator.get_status();
    HttpResponse::Ok().json(status)
}

#[get("/api/events")]
async fn events(data: web::Data<AppState>, query: web::Query<EventsQuery>) -> impl Responder {
    let events = db::get_events(
        &data.coordinator.db,
        query.since.as_deref(),
        query.limit.unwrap_or(100),
    )
    .await;

    match events {
        Ok(events) => HttpResponse::Ok().json(events),
        Err(e) => HttpResponse::InternalServerError().body(format!("DB error: {e}")),
    }
}

#[get("/api/metrics")]
async fn metrics(data: web::Data<AppState>, query: web::Query<MetricsQuery>) -> impl Responder {
    let metrics = db::get_metrics(
        &data.coordinator.db,
        query.metric_type.as_deref(),
        query.since.as_deref(),
        query.limit.unwrap_or(1000),
    )
    .await;

    match metrics {
        Ok(metrics) => HttpResponse::Ok().json(metrics),
        Err(e) => HttpResponse::InternalServerError().body(format!("DB error: {e}")),
    }
}

#[get("/api/stream")]
async fn stream(data: web::Data<AppState>) -> impl Responder {
    let rx = data.coordinator.broadcast_tx.subscribe();

    let stream = async_stream::stream! {
        let mut rx = rx;
        loop {
            match rx.recv().await {
                Ok(msg) => {
                    if let Ok(json) = serde_json::to_string(&msg) {
                        let event_type = match &msg {
                            DashboardMessage::Event(_) => "event",
                            DashboardMessage::Metric(_) => "metric",
                            DashboardMessage::Status(_) => "status",
                        };
                        yield Ok::<_, actix_web::Error>(
                            web::Bytes::from(format!("event: {event_type}\ndata: {json}\n\n"))
                        );
                    }
                }
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    debug!("SSE client lagged by {n} messages");
                    continue;
                }
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    HttpResponse::Ok()
        .content_type("text/event-stream")
        .insert_header(("Cache-Control", "no-cache"))
        .insert_header(("X-Accel-Buffering", "no"))
        .streaming(stream)
}

#[get("/api/report")]
async fn report(data: web::Data<AppState>) -> impl Responder {
    match crate::report::generate_report(&data.coordinator.db).await {
        Ok(html) => HttpResponse::Ok()
            .content_type("text/html; charset=utf-8")
            .body(html),
        Err(e) => HttpResponse::InternalServerError().body(format!("Report error: {e}")),
    }
}

#[derive(serde::Deserialize)]
pub struct EventsQuery {
    pub since: Option<String>,
    pub limit: Option<i64>,
}

#[derive(serde::Deserialize)]
pub struct MetricsQuery {
    pub metric_type: Option<String>,
    pub since: Option<String>,
    pub limit: Option<i64>,
}

pub async fn start_server(port: u16, coordinator: Arc<ChaosCoordinator>) -> std::io::Result<()> {
    let data = web::Data::new(AppState { coordinator });

    actix_web::HttpServer::new(move || {
        actix_web::App::new()
            .app_data(data.clone())
            .service(index)
            .service(status)
            .service(events)
            .service(metrics)
            .service(stream)
            .service(report)
    })
    .bind(("0.0.0.0", port))?
    .run()
    .await
}