ocpp_service/db/
stations.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Charging stations, EVSEs and connectors.
5
6use evse_id::EvseId;
7use serde::{Deserialize, Serialize};
8
9use super::{Db, Ts};
10
11#[derive(Debug, Clone, sqlx::FromRow, Serialize)]
12pub struct Station {
13    pub id: i64,
14    pub ocpp_identity: String,
15    pub evse_id: String,
16    pub name: String,
17    pub latitude: Option<f64>,
18    pub longitude: Option<f64>,
19    pub address_line1: Option<String>,
20    pub address_line2: Option<String>,
21    pub postal_code: Option<String>,
22    pub city: Option<String>,
23    pub region: Option<String>,
24    pub country_code: Option<String>,
25    #[serde(skip)]
26    pub auth_password_hash: Option<String>,
27    pub allowed_versions: Option<Vec<String>>,
28    pub boot_policy: String,
29    pub registration_status: String,
30    pub vendor: Option<String>,
31    pub model: Option<String>,
32    pub serial_number: Option<String>,
33    pub firmware_version: Option<String>,
34    pub iccid: Option<String>,
35    pub imsi: Option<String>,
36    pub meter_type: Option<String>,
37    pub meter_serial_number: Option<String>,
38    pub last_boot_at: Option<Ts>,
39    pub last_boot_reason: Option<String>,
40    pub status: Option<String>,
41    pub error_code: Option<String>,
42    pub status_at: Option<Ts>,
43    pub ocpp_version: Option<String>,
44    pub connected: bool,
45    pub connected_node: Option<String>,
46    pub connected_at: Option<Ts>,
47    pub disconnected_at: Option<Ts>,
48    pub last_heartbeat_at: Option<Ts>,
49    pub last_seen_at: Option<Ts>,
50    pub security_profile: Option<i16>,
51    pub timezone: Option<String>,
52    pub cost_mode: String,
53    pub created_at: Ts,
54    pub updated_at: Ts,
55}
56
57impl Station {
58    pub fn has_password(&self) -> bool {
59        self.auth_password_hash.is_some()
60    }
61}
62
63const COLS: &str = "id, ocpp_identity, evse_id, name, latitude, longitude, address_line1, address_line2, postal_code, city, region, country_code, \
64    auth_password_hash, allowed_versions, boot_policy, registration_status, vendor, model, serial_number, firmware_version, iccid, imsi, \
65    meter_type, meter_serial_number, last_boot_at, last_boot_reason, status, error_code, status_at, ocpp_version, connected, connected_node, \
66    connected_at, disconnected_at, last_heartbeat_at, last_seen_at, security_profile, timezone, cost_mode, created_at, updated_at";
67
68#[derive(Debug, Clone, Default, Deserialize)]
69pub struct StationInput {
70    pub ocpp_identity: Option<String>,
71    pub name: Option<String>,
72    /// Body of the station e-mobility ID (without type and check characters).
73    /// Generated from the database id when absent.
74    pub evse_id_body: Option<String>,
75    pub latitude: Option<f64>,
76    pub longitude: Option<f64>,
77    pub address_line1: Option<String>,
78    pub address_line2: Option<String>,
79    pub postal_code: Option<String>,
80    pub city: Option<String>,
81    pub region: Option<String>,
82    pub country_code: Option<String>,
83    pub allowed_versions: Option<Vec<String>>,
84    pub boot_policy: Option<String>,
85    /// OCPP Security Profile 0–3 enforced when the station connects.
86    pub security_profile: Option<i16>,
87    /// IANA time zone, e.g. `Europe/Paris`.
88    pub timezone: Option<String>,
89    /// `None`, `Central` or `Local` (OCPP 2.x tariff and cost).
90    pub cost_mode: Option<String>,
91}
92
93pub async fn get_by_identity(db: &Db, identity: &str) -> sqlx::Result<Option<Station>> {
94    sqlx::query_as(&format!(
95        "SELECT {COLS} FROM charging_stations WHERE ocpp_identity = $1"
96    ))
97    .bind(identity)
98    .fetch_optional(db)
99    .await
100}
101
102pub async fn get(db: &Db, id: i64) -> sqlx::Result<Option<Station>> {
103    sqlx::query_as(&format!(
104        "SELECT {COLS} FROM charging_stations WHERE id = $1"
105    ))
106    .bind(id)
107    .fetch_optional(db)
108    .await
109}
110
111/// Creates a station. The `S` e-mobility ID is derived from the body given
112/// or, by default, from the generated database id.
113pub async fn create(
114    db: &Db,
115    input: &StationInput,
116    identity: &str,
117    name: &str,
118) -> Result<Station, crate::error::AppError> {
119    let mut tx = db.begin().await?;
120    let id: i64 = sqlx::query_scalar(
121        "INSERT INTO charging_stations (ocpp_identity, evse_id, name, latitude, longitude, address_line1, address_line2, postal_code, city, region, \
122         country_code, allowed_versions, boot_policy, security_profile, timezone, cost_mode) \
123         VALUES ($1, 'tmp-' || gen_random_uuid(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, COALESCE($12, 'Accepted'), $13, $14, COALESCE($15, 'None')) RETURNING id",
124    )
125    .bind(identity)
126    .bind(name)
127    .bind(input.latitude)
128    .bind(input.longitude)
129    .bind(&input.address_line1)
130    .bind(&input.address_line2)
131    .bind(&input.postal_code)
132    .bind(&input.city)
133    .bind(&input.region)
134    .bind(input.country_code.as_ref().map(|c| c.to_ascii_uppercase()))
135    .bind(&input.allowed_versions)
136    .bind(&input.boot_policy)
137    .bind(input.security_profile)
138    .bind(&input.timezone)
139    .bind(&input.cost_mode)
140    .fetch_one(&mut *tx)
141    .await?;
142    let evse_id = match &input.evse_id_body {
143        Some(body) => EvseId::generate(evse_id::IdType::Station, body)
144            .map_err(|e| crate::error::AppError::BadRequest(format!("evse_id_body: {e}")))?,
145        None => EvseId::station(id as u64),
146    };
147    sqlx::query("UPDATE charging_stations SET evse_id = $2 WHERE id = $1")
148        .bind(id)
149        .bind(evse_id.as_str())
150        .execute(&mut *tx)
151        .await?;
152    tx.commit().await?;
153    Ok(get(db, id).await?.expect("just inserted"))
154}
155
156pub async fn update(db: &Db, id: i64, input: &StationInput) -> sqlx::Result<Option<Station>> {
157    sqlx::query_as(&format!(
158        "UPDATE charging_stations SET \
159           name = COALESCE($2, name), latitude = COALESCE($3, latitude), longitude = COALESCE($4, longitude), \
160           address_line1 = COALESCE($5, address_line1), address_line2 = COALESCE($6, address_line2), postal_code = COALESCE($7, postal_code), \
161           city = COALESCE($8, city), region = COALESCE($9, region), country_code = COALESCE($10, country_code), \
162           allowed_versions = COALESCE($11, allowed_versions), boot_policy = COALESCE($12, boot_policy), \
163           security_profile = COALESCE($13, security_profile), timezone = COALESCE($14, timezone), cost_mode = COALESCE($15, cost_mode), updated_at = now() \
164         WHERE id = $1 RETURNING {COLS}"
165    ))
166    .bind(id)
167    .bind(&input.name)
168    .bind(input.latitude)
169    .bind(input.longitude)
170    .bind(&input.address_line1)
171    .bind(&input.address_line2)
172    .bind(&input.postal_code)
173    .bind(&input.city)
174    .bind(&input.region)
175    .bind(input.country_code.as_ref().map(|c| c.to_ascii_uppercase()))
176    .bind(&input.allowed_versions)
177    .bind(&input.boot_policy)
178    .bind(input.security_profile)
179    .bind(&input.timezone)
180    .bind(&input.cost_mode)
181    .fetch_optional(db)
182    .await
183}
184
185pub async fn delete(db: &Db, id: i64) -> sqlx::Result<bool> {
186    Ok(sqlx::query("DELETE FROM charging_stations WHERE id = $1")
187        .bind(id)
188        .execute(db)
189        .await?
190        .rows_affected()
191        > 0)
192}
193
194pub async fn set_password_hash(db: &Db, id: i64, hash: Option<&str>) -> sqlx::Result<bool> {
195    Ok(sqlx::query(
196        "UPDATE charging_stations SET auth_password_hash = $2, updated_at = now() WHERE id = $1",
197    )
198    .bind(id)
199    .bind(hash)
200    .execute(db)
201    .await?
202    .rows_affected()
203        > 0)
204}
205
206#[derive(Debug, Clone, Default, Deserialize)]
207pub struct SearchParams {
208    /// Free text over name and postal address (accent- and case-insensitive).
209    pub q: Option<String>,
210    pub lat: Option<f64>,
211    pub lon: Option<f64>,
212    /// Radius around (lat, lon), default 10 km.
213    pub radius_km: Option<f64>,
214    pub city: Option<String>,
215    pub postal_code: Option<String>,
216    pub country_code: Option<String>,
217    pub connected: Option<bool>,
218    pub limit: Option<i64>,
219    pub offset: Option<i64>,
220}
221
222#[derive(Debug, Serialize, sqlx::FromRow)]
223pub struct SearchHit {
224    #[sqlx(flatten)]
225    #[serde(flatten)]
226    pub station: Station,
227    /// Distance in km when searching around a position.
228    pub distance_km: Option<f64>,
229}
230
231pub async fn search(db: &Db, p: &SearchParams) -> sqlx::Result<Vec<SearchHit>> {
232    let radius = p.radius_km.unwrap_or(10.0).clamp(0.01, 20_000.0);
233    // Bounding box (uses the (latitude, longitude) index), then exact haversine distance.
234    let (lat_delta, lon_delta) = match p.lat {
235        Some(lat) => (
236            radius / 111.32,
237            radius / (111.32 * lat.to_radians().cos().abs().max(0.01)),
238        ),
239        None => (0.0, 0.0),
240    };
241    sqlx::query_as(&format!(
242        "SELECT * FROM ( \
243           SELECT {COLS}, \
244             CASE WHEN $3::float8 IS NULL OR latitude IS NULL THEN NULL ELSE \
245               6371.0 * 2 * asin(sqrt(power(sin(radians(latitude - $3) / 2), 2) + \
246               cos(radians($3)) * cos(radians(latitude)) * power(sin(radians(longitude - $4) / 2), 2))) END AS distance_km, \
247             CASE WHEN $1::text IS NULL THEN 0 ELSE similarity(address_search, lower(ocpp_immutable_unaccent($1))) END AS score \
248           FROM charging_stations \
249           WHERE ($1::text IS NULL OR address_search ILIKE '%' || lower(ocpp_immutable_unaccent($1)) || '%' \
250                  OR address_search % lower(ocpp_immutable_unaccent($1))) \
251             AND ($3::float8 IS NULL OR (latitude BETWEEN $3 - $5 AND $3 + $5 AND longitude BETWEEN $4 - $6 AND $4 + $6)) \
252             AND ($7::text IS NULL OR lower(ocpp_immutable_unaccent(city)) = lower(ocpp_immutable_unaccent($7))) \
253             AND ($8::text IS NULL OR postal_code = $8) \
254             AND ($9::text IS NULL OR country_code = upper($9)) \
255             AND ($10::bool IS NULL OR connected = $10) \
256         ) s \
257         WHERE $3::float8 IS NULL OR distance_km <= $2 \
258         ORDER BY distance_km ASC NULLS LAST, score DESC, name ASC \
259         LIMIT $11 OFFSET $12"
260    ))
261    .bind(p.q.as_deref().map(str::trim).filter(|s| !s.is_empty()))
262    .bind(radius)
263    .bind(p.lat.filter(|_| p.lon.is_some()))
264    .bind(p.lon)
265    .bind(lat_delta)
266    .bind(lon_delta)
267    .bind(&p.city)
268    .bind(&p.postal_code)
269    .bind(&p.country_code)
270    .bind(p.connected)
271    .bind(p.limit.unwrap_or(50).clamp(1, 500))
272    .bind(p.offset.unwrap_or(0).max(0))
273    .fetch_all(db)
274    .await
275}
276
277// --- Connection state --------------------------------------------------------
278
279pub async fn mark_connected(
280    db: &Db,
281    id: i64,
282    node: &str,
283    connection_id: &str,
284    version: &str,
285) -> sqlx::Result<()> {
286    sqlx::query(
287        "UPDATE charging_stations SET connected = true, connected_node = $2, connection_id = $3, ocpp_version = $4, \
288         connected_at = now(), last_seen_at = now(), updated_at = now() WHERE id = $1",
289    )
290    .bind(id)
291    .bind(node)
292    .bind(connection_id)
293    .bind(version)
294    .execute(db)
295    .await?;
296    Ok(())
297}
298
299/// Marks the station offline unless a newer connection took over meanwhile.
300pub async fn mark_disconnected(db: &Db, id: i64, connection_id: &str) -> sqlx::Result<()> {
301    sqlx::query(
302        "UPDATE charging_stations SET connected = false, connected_node = NULL, connection_id = NULL, disconnected_at = now(), updated_at = now() \
303         WHERE id = $1 AND connection_id = $2",
304    )
305    .bind(id)
306    .bind(connection_id)
307    .execute(db)
308    .await?;
309    Ok(())
310}
311
312/// On startup: connections recorded for this node do not exist any more.
313pub async fn reset_node(db: &Db, node: &str) -> sqlx::Result<u64> {
314    Ok(sqlx::query(
315        "UPDATE charging_stations SET connected = false, connected_node = NULL, connection_id = NULL, disconnected_at = now() \
316         WHERE connected AND connected_node = $1",
317    )
318    .bind(node)
319    .execute(db)
320    .await?
321    .rows_affected())
322}
323
324pub async fn touch(db: &Db, id: i64, heartbeat: bool) -> sqlx::Result<()> {
325    let sql = if heartbeat {
326        "UPDATE charging_stations SET last_heartbeat_at = now(), last_seen_at = now() WHERE id = $1"
327    } else {
328        "UPDATE charging_stations SET last_seen_at = now() WHERE id = $1"
329    };
330    sqlx::query(sql).bind(id).execute(db).await?;
331    Ok(())
332}
333
334#[derive(Debug, Default)]
335pub struct BootInfo<'a> {
336    pub vendor: &'a str,
337    pub model: &'a str,
338    pub serial_number: Option<&'a str>,
339    pub firmware_version: Option<&'a str>,
340    pub iccid: Option<&'a str>,
341    pub imsi: Option<&'a str>,
342    pub meter_type: Option<&'a str>,
343    pub meter_serial_number: Option<&'a str>,
344    pub reason: Option<&'a str>,
345}
346
347pub async fn record_boot(
348    db: &Db,
349    id: i64,
350    b: &BootInfo<'_>,
351    registration_status: &str,
352) -> sqlx::Result<()> {
353    sqlx::query(
354        "UPDATE charging_stations SET vendor = $2, model = $3, serial_number = $4, firmware_version = $5, iccid = $6, imsi = $7, \
355         meter_type = $8, meter_serial_number = $9, last_boot_reason = $10, registration_status = $11, last_boot_at = now(), \
356         last_seen_at = now(), updated_at = now() WHERE id = $1",
357    )
358    .bind(id)
359    .bind(b.vendor)
360    .bind(b.model)
361    .bind(b.serial_number)
362    .bind(b.firmware_version)
363    .bind(b.iccid)
364    .bind(b.imsi)
365    .bind(b.meter_type)
366    .bind(b.meter_serial_number)
367    .bind(b.reason)
368    .bind(registration_status)
369    .execute(db)
370    .await?;
371    Ok(())
372}
373
374pub async fn set_firmware_version(db: &Db, id: i64, firmware: &str) -> sqlx::Result<()> {
375    sqlx::query(
376        "UPDATE charging_stations SET firmware_version = $2, updated_at = now() WHERE id = $1",
377    )
378    .bind(id)
379    .bind(firmware)
380    .execute(db)
381    .await?;
382    Ok(())
383}
384
385// --- EVSEs and connectors ----------------------------------------------------
386
387#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
388pub struct Evse {
389    pub evse_number: i32,
390    pub evse_id: String,
391    #[sqlx(json)]
392    pub connectors: Vec<Connector>,
393}
394
395#[derive(Debug, Clone, Serialize, Deserialize)]
396pub struct Connector {
397    pub connector_number: i32,
398    pub status: String,
399    pub error_code: Option<String>,
400    pub info: Option<String>,
401    pub vendor_error_code: Option<String>,
402    pub status_at: String,
403}
404
405/// Creates the EVSE (with its `E` e-mobility ID) if it does not exist yet.
406pub async fn ensure_evse(
407    db: &Db,
408    station_id: i64,
409    station_evse_id: &str,
410    evse_number: i32,
411) -> Result<String, crate::error::AppError> {
412    if let Some(id) = sqlx::query_scalar::<_, String>(
413        "SELECT evse_id FROM evses WHERE station_id = $1 AND evse_number = $2",
414    )
415    .bind(station_id)
416    .bind(evse_number)
417    .fetch_optional(db)
418    .await?
419    {
420        return Ok(id);
421    }
422    let station = EvseId::parse(station_evse_id).map_err(|e| {
423        crate::error::AppError::Internal(format!("stored station evse_id invalid: {e}"))
424    })?;
425    let evse = EvseId::evse_of(&station, evse_number as u32)
426        .map_err(|e| crate::error::AppError::BadRequest(e.to_string()))?;
427    sqlx::query("INSERT INTO evses (station_id, evse_number, evse_id) VALUES ($1, $2, $3) ON CONFLICT (station_id, evse_number) DO NOTHING")
428        .bind(station_id)
429        .bind(evse_number)
430        .bind(evse.as_str())
431        .execute(db)
432        .await?;
433    Ok(
434        sqlx::query_scalar("SELECT evse_id FROM evses WHERE station_id = $1 AND evse_number = $2")
435            .bind(station_id)
436            .bind(evse_number)
437            .fetch_one(db)
438            .await?,
439    )
440}
441
442pub struct StatusUpdate<'a> {
443    pub evse_number: i32,
444    pub connector_number: i32,
445    pub status: &'a str,
446    pub error_code: Option<&'a str>,
447    pub info: Option<&'a str>,
448    pub vendor_error_code: Option<&'a str>,
449    pub at: Ts,
450}
451
452pub async fn upsert_connector_status(
453    db: &Db,
454    station_id: i64,
455    u: &StatusUpdate<'_>,
456) -> sqlx::Result<()> {
457    // Out-of-order (older) notifications do not overwrite newer state.
458    sqlx::query(
459        "INSERT INTO connectors (station_id, evse_number, connector_number, status, error_code, info, vendor_error_code, status_at) \
460         VALUES ($1, $2, $3, $4, $5, $6, $7, $8) \
461         ON CONFLICT (station_id, evse_number, connector_number) DO UPDATE SET status = EXCLUDED.status, error_code = EXCLUDED.error_code, \
462           info = EXCLUDED.info, vendor_error_code = EXCLUDED.vendor_error_code, status_at = EXCLUDED.status_at, updated_at = now() \
463         WHERE connectors.status_at <= EXCLUDED.status_at",
464    )
465    .bind(station_id)
466    .bind(u.evse_number)
467    .bind(u.connector_number)
468    .bind(u.status)
469    .bind(u.error_code)
470    .bind(u.info)
471    .bind(u.vendor_error_code)
472    .bind(u.at)
473    .execute(db)
474    .await?;
475    Ok(())
476}
477
478pub async fn set_station_status(
479    db: &Db,
480    station_id: i64,
481    status: &str,
482    error_code: Option<&str>,
483    at: Ts,
484) -> sqlx::Result<()> {
485    sqlx::query(
486        "UPDATE charging_stations SET status = $2, error_code = $3, status_at = $4, updated_at = now() \
487         WHERE id = $1 AND (status_at IS NULL OR status_at <= $4)",
488    )
489    .bind(station_id)
490    .bind(status)
491    .bind(error_code)
492    .bind(at)
493    .execute(db)
494    .await?;
495    Ok(())
496}
497
498pub async fn list_evses(db: &Db, station_id: i64) -> sqlx::Result<Vec<Evse>> {
499    sqlx::query_as(
500        "SELECT e.evse_number, e.evse_id, COALESCE(json_agg(json_build_object( \
501             'connector_number', c.connector_number, 'status', c.status, 'error_code', c.error_code, 'info', c.info, \
502             'vendor_error_code', c.vendor_error_code, 'status_at', to_char(c.status_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"')) \
503             ORDER BY c.connector_number) FILTER (WHERE c.connector_number IS NOT NULL), '[]') AS connectors \
504         FROM evses e LEFT JOIN connectors c ON c.station_id = e.station_id AND c.evse_number = e.evse_number \
505         WHERE e.station_id = $1 GROUP BY e.evse_number, e.evse_id ORDER BY e.evse_number",
506    )
507    .bind(station_id)
508    .fetch_all(db)
509    .await
510}
ocpp_service/db/stations.rs

ocpp_service/db/
stations.rs
