ocpp_service/ocpp/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! OCPP-J endpoint: handshake, per-connection session and the handlers of
5//! each OCPP version. Version modules are fully separate; they only share the
6//! version-neutral persistence helpers of [`StationCtx`].
7
8pub mod handshake;
9pub mod security;
10pub mod session;
11pub mod v16;
12pub mod v201;
13pub mod v21;
14
15use ocpp_core::{ErrorCode, OcppVersion, RpcError};
16
17use crate::db::{self, Ts, stations::StatusUpdate};
18use crate::state::App;
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum Registration {
22    /// No BootNotification seen yet.
23    Unknown,
24    Pending,
25    Accepted,
26    Rejected,
27}
28
29impl Registration {
30    pub fn from_db(s: &str) -> Self {
31        match s {
32            "Accepted" => Registration::Accepted,
33            "Pending" => Registration::Pending,
34            "Rejected" => Registration::Rejected,
35            _ => Registration::Unknown,
36        }
37    }
38
39    /// Whether a station-initiated `action` is allowed in this state.
40    ///
41    /// OCPP-J 1.6 errata §3.10 / OCPP 2.x B02–B03: before being accepted a
42    /// station should only send BootNotification; the CSMS answers other
43    /// CALLs with SecurityError. Status and notification messages are still
44    /// accepted while Pending/Unknown (they may answer a TriggerMessage);
45    /// transaction-related messages are not.
46    pub fn allows(self, action: &str) -> bool {
47        const TRANSACTIONAL: &[&str] = &[
48            "Authorize",
49            "StartTransaction",
50            "StopTransaction",
51            "TransactionEvent",
52            "MeterValues",
53        ];
54        match self {
55            Registration::Accepted => true,
56            _ if action == "BootNotification" => true,
57            Registration::Pending | Registration::Unknown => !TRANSACTIONAL.contains(&action),
58            Registration::Rejected => false,
59        }
60    }
61}
62
63/// State of a connected station shared with the version handlers.
64pub struct StationCtx {
65    pub app: App,
66    pub station_id: i64,
67    pub identity: String,
68    /// Station `S` e-mobility ID, used to derive EVSE IDs.
69    pub station_evse_id: String,
70    pub version: OcppVersion,
71    pub registration: Registration,
72}
73
74/// Logs the detailed error and returns a generic InternalError to the station.
75pub fn internal(e: impl std::fmt::Display) -> RpcError {
76    tracing::error!(error = %e, "error while handling an OCPP message");
77    RpcError::new(
78        ErrorCode::InternalError,
79        "internal error while processing the message",
80    )
81}
82
83impl StationCtx {
84    pub fn now() -> ocpp_core::DateTime {
85        ocpp_core::DateTime::now()
86    }
87
88    pub fn heartbeat_interval(&self) -> i64 {
89        self.app.cfg.heartbeat_interval.as_secs() as i64
90    }
91
92    /// Stores a message in the station event log. Failures are logged, not propagated:
93    /// losing an audit entry must not make the station retry the message.
94    pub async fn log_event(&self, action: &str, payload: &str) {
95        let value = serde_json::from_str(payload).unwrap_or(serde_json::Value::Null);
96        if let Err(e) = db::events::insert(&self.app.db, self.station_id, action, &value).await {
97            tracing::warn!(station = %self.identity, action, error = %e, "failed to store station event");
98        }
99    }
100
101    /// Records a BootNotification and decides the registration status from
102    /// the station's `boot_policy` (the operator's onboarding control).
103    pub async fn boot(
104        &mut self,
105        info: &db::stations::BootInfo<'_>,
106    ) -> Result<Registration, RpcError> {
107        let station = db::stations::get(&self.app.db, self.station_id)
108            .await
109            .map_err(internal)?
110            .ok_or_else(|| internal("station deleted"))?;
111        let reg = Registration::from_db(&station.boot_policy);
112        let reg = if reg == Registration::Unknown {
113            Registration::Accepted
114        } else {
115            reg
116        };
117        let status = match reg {
118            Registration::Pending => "Pending",
119            Registration::Rejected => "Rejected",
120            _ => "Accepted",
121        };
122        db::stations::record_boot(&self.app.db, self.station_id, info, status)
123            .await
124            .map_err(internal)?;
125        self.registration = reg;
126        tracing::info!(station = %self.identity, vendor = info.vendor, model = info.model, status, "BootNotification");
127        Ok(reg)
128    }
129
130    pub async fn heartbeat(&self) -> Result<(), RpcError> {
131        db::stations::touch(&self.app.db, self.station_id, true)
132            .await
133            .map_err(internal)
134    }
135
136    pub async fn authorize(
137        &self,
138        token: &str,
139        token_type: Option<&str>,
140    ) -> Result<db::tokens::AuthResult, RpcError> {
141        db::tokens::authorize(
142            &self.app.db,
143            token,
144            token_type,
145            self.app.cfg.authorize_unknown_tokens,
146        )
147        .await
148        .map_err(internal)
149    }
150
151    /// Connector status; `evse_number == 0` is the station itself.
152    pub async fn status(&self, u: &StatusUpdate<'_>) -> Result<(), RpcError> {
153        if u.evse_number <= 0 {
154            return db::stations::set_station_status(
155                &self.app.db,
156                self.station_id,
157                u.status,
158                u.error_code,
159                u.at,
160            )
161            .await
162            .map_err(internal);
163        }
164        db::stations::ensure_evse(
165            &self.app.db,
166            self.station_id,
167            &self.station_evse_id,
168            u.evse_number,
169        )
170        .await
171        .map_err(internal)?;
172        db::stations::upsert_connector_status(&self.app.db, self.station_id, u)
173            .await
174            .map_err(internal)
175    }
176
177    pub async fn meter_values(
178        &self,
179        transaction_id: Option<i64>,
180        evse_number: i32,
181        at: Ts,
182        sampled: &serde_json::Value,
183    ) -> Result<(), RpcError> {
184        db::transactions::insert_meter_values(
185            &self.app.db,
186            self.station_id,
187            transaction_id,
188            evse_number,
189            at,
190            sampled,
191        )
192        .await
193        .map_err(internal)
194    }
195}
196
197/// Clamps an OCPP integer into an `i32` database column.
198pub fn i32_of(v: i64) -> i32 {
199    v.clamp(i32::MIN as i64, i32::MAX as i64) as i32
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205
206    #[test]
207    fn registration_gating() {
208        assert!(Registration::Accepted.allows("StartTransaction"));
209        assert!(Registration::Pending.allows("BootNotification"));
210        assert!(Registration::Pending.allows("StatusNotification"));
211        assert!(!Registration::Pending.allows("TransactionEvent"));
212        assert!(!Registration::Unknown.allows("Authorize"));
213        assert!(!Registration::Rejected.allows("Heartbeat"));
214        assert!(Registration::Rejected.allows("BootNotification"));
215    }
216}
ocpp_service/ocpp/mod.rs

ocpp_service/ocpp/
mod.rs
