ocpp_service/db/
charging.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Charging profiles and external charging limits.
5
6use serde::Serialize;
7use serde_json::Value;
8
9use super::{Db, Ts};
10
11#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
12pub struct ProfileRow {
13    pub id: i64,
14    pub ocpp_version: String,
15    pub profile_id: i32,
16    pub evse_number: i32,
17    pub purpose: String,
18    pub stack_level: i32,
19    pub kind: String,
20    pub transaction_uid: Option<String>,
21    pub valid_from: Option<Ts>,
22    pub valid_to: Option<Ts>,
23    pub source: String,
24    pub status: String,
25    pub status_info: Option<Value>,
26    pub profile: Value,
27    pub dyn_update: Option<Value>,
28    pub dyn_updated_at: Option<Ts>,
29    pub created_at: Ts,
30    pub updated_at: Ts,
31}
32
33const COLS: &str = "id, ocpp_version, profile_id, evse_number, purpose, stack_level, kind, transaction_uid, valid_from, valid_to, source, status, \
34    status_info, profile, dyn_update, dyn_updated_at, created_at, updated_at";
35
36/// Normalized columns of a profile row.
37pub struct NewProfile<'a> {
38    pub ocpp_version: &'a str,
39    pub profile_id: i64,
40    pub evse_number: i64,
41    pub purpose: &'a str,
42    pub stack_level: i64,
43    pub kind: &'a str,
44    pub transaction_uid: Option<&'a str>,
45    pub valid_from: Option<Ts>,
46    pub valid_to: Option<Ts>,
47    pub source: &'a str,
48    pub profile: &'a Value,
49}
50
51fn i32_of(v: i64) -> i32 {
52    v.clamp(i32::MIN as i64, i32::MAX as i64) as i32
53}
54
55/// Records a profile that is being sent (status Pending). A previous Pending
56/// row with the same id (an unanswered attempt) is discarded.
57pub async fn insert_pending(db: &Db, station_id: i64, p: &NewProfile<'_>) -> sqlx::Result<i64> {
58    let mut tx = db.begin().await?;
59    sqlx::query("DELETE FROM charging_profiles WHERE station_id = $1 AND profile_id = $2 AND status = 'Pending'")
60        .bind(station_id)
61        .bind(i32_of(p.profile_id))
62        .execute(&mut *tx)
63        .await?;
64    let id = insert(&mut tx, station_id, p, "Pending").await?;
65    tx.commit().await?;
66    Ok(id)
67}
68
69async fn insert(
70    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
71    station_id: i64,
72    p: &NewProfile<'_>,
73    status: &str,
74) -> sqlx::Result<i64> {
75    sqlx::query_scalar(
76        "INSERT INTO charging_profiles (station_id, ocpp_version, profile_id, evse_number, purpose, stack_level, kind, transaction_uid, \
77           valid_from, valid_to, source, status, profile) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING id",
78    )
79    .bind(station_id)
80    .bind(p.ocpp_version)
81    .bind(i32_of(p.profile_id))
82    .bind(i32_of(p.evse_number))
83    .bind(p.purpose)
84    .bind(i32_of(p.stack_level))
85    .bind(p.kind)
86    .bind(p.transaction_uid)
87    .bind(p.valid_from)
88    .bind(p.valid_to)
89    .bind(p.source)
90    .bind(status)
91    .bind(sqlx::types::Json(p.profile))
92    .fetch_one(&mut **tx)
93    .await
94}
95
96/// The station accepted the profile: it replaces any installed profile with
97/// the same id and, when `replace_same_slot`, the one with the same
98/// EVSE + purpose + stack level (and transaction for TxProfiles), as the
99/// station itself does (1.6 §3.13.2, §5.16.3).
100pub async fn mark_installed(
101    db: &Db,
102    station_id: i64,
103    row_id: i64,
104    replace_same_slot: bool,
105) -> sqlx::Result<()> {
106    let mut tx = db.begin().await?;
107    sqlx::query(
108        "UPDATE charging_profiles o SET status = 'Cleared', updated_at = now() FROM charging_profiles n \
109         WHERE n.id = $2 AND o.station_id = $1 AND o.id <> n.id AND o.status = 'Installed' AND (o.profile_id = n.profile_id OR \
110           ($3 AND o.evse_number = n.evse_number AND o.purpose = n.purpose AND o.stack_level = n.stack_level \
111            AND o.transaction_uid IS NOT DISTINCT FROM n.transaction_uid))",
112    )
113    .bind(station_id)
114    .bind(row_id)
115    .bind(replace_same_slot)
116    .execute(&mut *tx)
117    .await?;
118    sqlx::query(
119        "UPDATE charging_profiles SET status = 'Installed', updated_at = now() WHERE id = $1",
120    )
121    .bind(row_id)
122    .execute(&mut *tx)
123    .await?;
124    tx.commit().await
125}
126
127pub async fn mark_rejected(db: &Db, row_id: i64, status_info: Option<&Value>) -> sqlx::Result<()> {
128    sqlx::query("UPDATE charging_profiles SET status = 'Rejected', status_info = $2, updated_at = now() WHERE id = $1")
129        .bind(row_id)
130        .bind(status_info.map(sqlx::types::Json))
131        .execute(db)
132        .await?;
133    Ok(())
134}
135
136/// Removes the Pending row of a profile that could not be delivered.
137pub async fn discard_pending(db: &Db, row_id: i64) -> sqlx::Result<()> {
138    sqlx::query("DELETE FROM charging_profiles WHERE id = $1 AND status = 'Pending'")
139        .bind(row_id)
140        .execute(db)
141        .await?;
142    Ok(())
143}
144
145pub async fn list(db: &Db, station_id: i64, status: Option<&str>) -> sqlx::Result<Vec<ProfileRow>> {
146    sqlx::query_as(&format!(
147        "SELECT {COLS} FROM charging_profiles WHERE station_id = $1 AND ($2::text IS NULL OR status = $2) ORDER BY evse_number, purpose, stack_level DESC, id"
148    ))
149    .bind(station_id)
150    .bind(status)
151    .fetch_all(db)
152    .await
153}
154
155pub async fn installed(db: &Db, station_id: i64) -> sqlx::Result<Vec<ProfileRow>> {
156    list(db, station_id, Some("Installed")).await
157}
158
159pub async fn find_installed(
160    db: &Db,
161    station_id: i64,
162    profile_id: i64,
163) -> sqlx::Result<Option<ProfileRow>> {
164    sqlx::query_as(&format!("SELECT {COLS} FROM charging_profiles WHERE station_id = $1 AND profile_id = $2 AND status = 'Installed'"))
165        .bind(station_id)
166        .bind(i32_of(profile_id))
167        .fetch_optional(db)
168        .await
169}
170
171/// Criteria of a ClearChargingProfile request (all optional, combined with AND).
172#[derive(Debug, Default)]
173pub struct ClearCriteria<'a> {
174    pub profile_id: Option<i64>,
175    pub evse_number: Option<i64>,
176    pub purpose: Option<&'a str>,
177    pub stack_level: Option<i64>,
178}
179
180pub async fn clear(db: &Db, station_id: i64, c: &ClearCriteria<'_>) -> sqlx::Result<u64> {
181    Ok(sqlx::query(
182        "UPDATE charging_profiles SET status = 'Cleared', updated_at = now() WHERE station_id = $1 AND status = 'Installed' AND source = 'CSO' \
183         AND ($2::int IS NULL OR profile_id = $2) AND ($3::int IS NULL OR evse_number = $3) AND ($4::text IS NULL OR purpose = $4) \
184         AND ($5::int IS NULL OR stack_level = $5)",
185    )
186    .bind(station_id)
187    .bind(c.profile_id.map(i32_of))
188    .bind(c.evse_number.map(i32_of))
189    .bind(c.purpose)
190    .bind(c.stack_level.map(i32_of))
191    .execute(db)
192    .await?
193    .rows_affected())
194}
195
196/// A TxProfile ends with its transaction (1.6 §3.13.1, 2.x K §3.2).
197pub async fn clear_for_transaction(
198    db: &Db,
199    station_id: i64,
200    transaction_uid: &str,
201) -> sqlx::Result<u64> {
202    Ok(sqlx::query(
203        "UPDATE charging_profiles SET status = 'Cleared', updated_at = now() \
204         WHERE station_id = $1 AND status = 'Installed' AND purpose IN ('TxProfile', 'PriorityCharging') AND transaction_uid = $2",
205    )
206    .bind(station_id)
207    .bind(transaction_uid)
208    .execute(db)
209    .await?
210    .rows_affected())
211}
212
213/// Stores a profile reported by the station (ReportChargingProfiles) as installed.
214pub async fn upsert_reported(db: &Db, station_id: i64, p: &NewProfile<'_>) -> sqlx::Result<()> {
215    let mut tx = db.begin().await?;
216    sqlx::query("UPDATE charging_profiles SET status = 'Cleared', updated_at = now() WHERE station_id = $1 AND profile_id = $2 AND status IN ('Installed', 'Pending')")
217        .bind(station_id)
218        .bind(i32_of(p.profile_id))
219        .execute(&mut *tx)
220        .await?;
221    insert(&mut tx, station_id, p, "Installed").await?;
222    tx.commit().await
223}
224
225pub async fn set_dyn_update(db: &Db, row_id: i64, update: &Value) -> sqlx::Result<()> {
226    sqlx::query("UPDATE charging_profiles SET dyn_update = $2, dyn_updated_at = now(), updated_at = now() WHERE id = $1")
227        .bind(row_id)
228        .bind(sqlx::types::Json(update))
229        .execute(db)
230        .await?;
231    Ok(())
232}
233
234/// Next free profile id for CSMS-generated profiles. Starts high to stay clear of ids
235/// that stations generate for external constraints (2.1 K01.FR.80).
236pub async fn next_profile_id(db: &Db, station_id: i64) -> sqlx::Result<i64> {
237    sqlx::query_scalar("SELECT GREATEST(COALESCE(MAX(profile_id), 0) + 1, 100000)::bigint FROM charging_profiles WHERE station_id = $1")
238        .bind(station_id)
239        .fetch_one(db)
240        .await
241}
242
243#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
244pub struct ExternalLimit {
245    pub evse_number: i32,
246    pub source: String,
247    pub is_grid_critical: Option<bool>,
248    pub is_local_generation: Option<bool>,
249    pub schedules: Option<Value>,
250    pub received_at: Ts,
251}
252
253pub async fn upsert_external_limit(
254    db: &Db,
255    station_id: i64,
256    evse_number: i64,
257    source: &str,
258    grid_critical: Option<bool>,
259    local_generation: Option<bool>,
260    schedules: Option<&Value>,
261) -> sqlx::Result<()> {
262    sqlx::query(
263        "INSERT INTO external_charging_limits (station_id, evse_number, source, is_grid_critical, is_local_generation, schedules, received_at) \
264         VALUES ($1, $2, $3, $4, $5, $6, now()) ON CONFLICT (station_id, evse_number, source) DO UPDATE SET \
265         is_grid_critical = EXCLUDED.is_grid_critical, is_local_generation = EXCLUDED.is_local_generation, schedules = EXCLUDED.schedules, received_at = now()",
266    )
267    .bind(station_id)
268    .bind(i32_of(evse_number))
269    .bind(source)
270    .bind(grid_critical)
271    .bind(local_generation)
272    .bind(schedules.map(sqlx::types::Json))
273    .execute(db)
274    .await?;
275    Ok(())
276}
277
278/// ClearedChargingLimit: without an EVSE, the limits of that source are released everywhere.
279pub async fn delete_external_limits(
280    db: &Db,
281    station_id: i64,
282    source: &str,
283    evse_number: Option<i64>,
284) -> sqlx::Result<u64> {
285    Ok(sqlx::query("DELETE FROM external_charging_limits WHERE station_id = $1 AND source = $2 AND ($3::int IS NULL OR evse_number = $3)")
286        .bind(station_id)
287        .bind(source)
288        .bind(evse_number.map(i32_of))
289        .execute(db)
290        .await?
291        .rows_affected())
292}
293
294pub async fn external_limits(db: &Db, station_id: i64) -> sqlx::Result<Vec<ExternalLimit>> {
295    sqlx::query_as("SELECT evse_number, source, is_grid_critical, is_local_generation, schedules, received_at FROM external_charging_limits WHERE station_id = $1 ORDER BY evse_number, source")
296        .bind(station_id)
297        .fetch_all(db)
298        .await
299}
ocpp_service/db/charging.rs

ocpp_service/db/
charging.rs
