codetics-ocpp-service 0.1.0

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

//! OCPP-J endpoint: handshake, per-connection session and the handlers of
//! each OCPP version. Version modules are fully separate; they only share the
//! version-neutral persistence helpers of [`StationCtx`].

pub mod handshake;
pub mod security;
pub mod session;
pub mod v16;
pub mod v201;
pub mod v21;

use ocpp_core::{ErrorCode, OcppVersion, RpcError};

use crate::db::{self, Ts, stations::StatusUpdate};
use crate::state::App;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Registration {
    /// No BootNotification seen yet.
    Unknown,
    Pending,
    Accepted,
    Rejected,
}

impl Registration {
    pub fn from_db(s: &str) -> Self {
        match s {
            "Accepted" => Registration::Accepted,
            "Pending" => Registration::Pending,
            "Rejected" => Registration::Rejected,
            _ => Registration::Unknown,
        }
    }

    /// Whether a station-initiated `action` is allowed in this state.
    ///
    /// OCPP-J 1.6 errata §3.10 / OCPP 2.x B02–B03: before being accepted a
    /// station should only send BootNotification; the CSMS answers other
    /// CALLs with SecurityError. Status and notification messages are still
    /// accepted while Pending/Unknown (they may answer a TriggerMessage);
    /// transaction-related messages are not.
    pub fn allows(self, action: &str) -> bool {
        const TRANSACTIONAL: &[&str] = &[
            "Authorize",
            "StartTransaction",
            "StopTransaction",
            "TransactionEvent",
            "MeterValues",
        ];
        match self {
            Registration::Accepted => true,
            _ if action == "BootNotification" => true,
            Registration::Pending | Registration::Unknown => !TRANSACTIONAL.contains(&action),
            Registration::Rejected => false,
        }
    }
}

/// State of a connected station shared with the version handlers.
pub struct StationCtx {
    pub app: App,
    pub station_id: i64,
    pub identity: String,
    /// Station `S` e-mobility ID, used to derive EVSE IDs.
    pub station_evse_id: String,
    pub version: OcppVersion,
    pub registration: Registration,
}

/// Logs the detailed error and returns a generic InternalError to the station.
pub fn internal(e: impl std::fmt::Display) -> RpcError {
    tracing::error!(error = %e, "error while handling an OCPP message");
    RpcError::new(
        ErrorCode::InternalError,
        "internal error while processing the message",
    )
}

impl StationCtx {
    pub fn now() -> ocpp_core::DateTime {
        ocpp_core::DateTime::now()
    }

    pub fn heartbeat_interval(&self) -> i64 {
        self.app.cfg.heartbeat_interval.as_secs() as i64
    }

    /// Stores a message in the station event log. Failures are logged, not propagated:
    /// losing an audit entry must not make the station retry the message.
    pub async fn log_event(&self, action: &str, payload: &str) {
        let value = serde_json::from_str(payload).unwrap_or(serde_json::Value::Null);
        if let Err(e) = db::events::insert(&self.app.db, self.station_id, action, &value).await {
            tracing::warn!(station = %self.identity, action, error = %e, "failed to store station event");
        }
    }

    /// Records a BootNotification and decides the registration status from
    /// the station's `boot_policy` (the operator's onboarding control).
    pub async fn boot(
        &mut self,
        info: &db::stations::BootInfo<'_>,
    ) -> Result<Registration, RpcError> {
        let station = db::stations::get(&self.app.db, self.station_id)
            .await
            .map_err(internal)?
            .ok_or_else(|| internal("station deleted"))?;
        let reg = Registration::from_db(&station.boot_policy);
        let reg = if reg == Registration::Unknown {
            Registration::Accepted
        } else {
            reg
        };
        let status = match reg {
            Registration::Pending => "Pending",
            Registration::Rejected => "Rejected",
            _ => "Accepted",
        };
        db::stations::record_boot(&self.app.db, self.station_id, info, status)
            .await
            .map_err(internal)?;
        self.registration = reg;
        tracing::info!(station = %self.identity, vendor = info.vendor, model = info.model, status, "BootNotification");
        Ok(reg)
    }

    pub async fn heartbeat(&self) -> Result<(), RpcError> {
        db::stations::touch(&self.app.db, self.station_id, true)
            .await
            .map_err(internal)
    }

    pub async fn authorize(
        &self,
        token: &str,
        token_type: Option<&str>,
    ) -> Result<db::tokens::AuthResult, RpcError> {
        db::tokens::authorize(
            &self.app.db,
            token,
            token_type,
            self.app.cfg.authorize_unknown_tokens,
        )
        .await
        .map_err(internal)
    }

    /// Connector status; `evse_number == 0` is the station itself.
    pub async fn status(&self, u: &StatusUpdate<'_>) -> Result<(), RpcError> {
        if u.evse_number <= 0 {
            return db::stations::set_station_status(
                &self.app.db,
                self.station_id,
                u.status,
                u.error_code,
                u.at,
            )
            .await
            .map_err(internal);
        }
        db::stations::ensure_evse(
            &self.app.db,
            self.station_id,
            &self.station_evse_id,
            u.evse_number,
        )
        .await
        .map_err(internal)?;
        db::stations::upsert_connector_status(&self.app.db, self.station_id, u)
            .await
            .map_err(internal)
    }

    pub async fn meter_values(
        &self,
        transaction_id: Option<i64>,
        evse_number: i32,
        at: Ts,
        sampled: &serde_json::Value,
    ) -> Result<(), RpcError> {
        db::transactions::insert_meter_values(
            &self.app.db,
            self.station_id,
            transaction_id,
            evse_number,
            at,
            sampled,
        )
        .await
        .map_err(internal)
    }
}

/// Clamps an OCPP integer into an `i32` database column.
pub fn i32_of(v: i64) -> i32 {
    v.clamp(i32::MIN as i64, i32::MAX as i64) as i32
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn registration_gating() {
        assert!(Registration::Accepted.allows("StartTransaction"));
        assert!(Registration::Pending.allows("BootNotification"));
        assert!(Registration::Pending.allows("StatusNotification"));
        assert!(!Registration::Pending.allows("TransactionEvent"));
        assert!(!Registration::Unknown.allows("Authorize"));
        assert!(!Registration::Rejected.allows("Heartbeat"));
        assert!(Registration::Rejected.allows("BootNotification"));
    }
}