codetics-ocpp-service 0.1.0

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)

//! Accept loop for the OCPP listeners (plain and TLS).
//!
//! Each accepted connection runs its TLS handshake (if any) in its own task,
//! then hyper serves HTTP/1.1 with upgrades. Connection metadata (peer
//! address, TLS outcome) is attached to every request as a [`ConnMeta`]
//! extension for the WebSocket handshake handler.

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use axum::Router;
use axum::body::Body;
use hyper::body::Incoming;
use hyper_util::rt::{TokioIo, TokioTimer};
use tokio::net::TcpListener;
use tokio::sync::{Semaphore, watch};
use tower::ServiceExt;

use crate::tls::{TlsAcceptor, TlsInfo};

const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(15);
const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(15);

#[derive(Debug, Clone)]
pub struct ConnMeta {
    pub peer: SocketAddr,
    /// `None` for the plain (unencrypted) listener.
    pub tls: Option<TlsInfo>,
}

pub async fn serve_ocpp(
    listener: TcpListener,
    tls: Option<Arc<TlsAcceptor>>,
    router: Router,
    limit: Arc<Semaphore>,
    mut shutdown: watch::Receiver<bool>,
) {
    loop {
        let (tcp, peer) = tokio::select! {
            _ = shutdown.changed() => return,
            accepted = listener.accept() => match accepted {
                Ok(a) => a,
                Err(e) => {
                    // EMFILE and similar: back off instead of spinning.
                    tracing::warn!(error = %e, "accept failed");
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    continue;
                }
            },
        };
        let Ok(permit) = limit.clone().try_acquire_owned() else {
            tracing::warn!(%peer, "connection limit reached, refusing connection");
            continue;
        };
        let _ = tcp.set_nodelay(true);
        let router = router.clone();
        let tls = tls.clone();
        tokio::spawn(async move {
            let _permit = permit;
            match tls {
                None => serve_conn(tcp, ConnMeta { peer, tls: None }, router).await,
                Some(acceptor) => {
                    match tokio::time::timeout(TLS_HANDSHAKE_TIMEOUT, acceptor.accept(tcp)).await {
                        Ok(Ok((stream, info))) => {
                            serve_conn(
                                stream,
                                ConnMeta {
                                    peer,
                                    tls: Some(info),
                                },
                                router,
                            )
                            .await
                        }
                        Ok(Err(e)) => tracing::info!(%peer, error = %e, "TLS handshake failed"),
                        Err(_) => tracing::info!(%peer, "TLS handshake timed out"),
                    }
                }
            }
        });
    }
}

async fn serve_conn<S>(stream: S, meta: ConnMeta, router: Router)
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
    let service = hyper::service::service_fn(move |mut req: hyper::Request<Incoming>| {
        req.extensions_mut().insert(meta.clone());
        router.clone().oneshot(req.map(Body::new))
    });
    let conn = hyper::server::conn::http1::Builder::new()
        .timer(TokioTimer::new())
        .header_read_timeout(HEADER_READ_TIMEOUT)
        .serve_connection(TokioIo::new(stream), service)
        .with_upgrades();
    if let Err(e) = conn.await {
        tracing::debug!(error = %e, "HTTP connection error");
    }
}