ocpp_service/db/
transactions.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Charging transactions and meter values.
5
6use serde::Serialize;
7
8use super::{Db, Ts};
9
10#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
11pub struct Transaction {
12    pub id: i64,
13    pub station_id: i64,
14    pub ocpp_version: String,
15    pub transaction_uid: String,
16    pub evse_number: Option<i32>,
17    pub connector_number: Option<i32>,
18    pub id_token: Option<String>,
19    pub id_token_type: Option<String>,
20    pub status: String,
21    pub charging_state: Option<String>,
22    pub started_at: Option<Ts>,
23    pub ended_at: Option<Ts>,
24    pub meter_start_wh: Option<f64>,
25    pub meter_stop_wh: Option<f64>,
26    pub stop_reason: Option<String>,
27    pub reservation_id: Option<i32>,
28    pub remote_start_id: Option<i32>,
29    pub last_seq_no: Option<i32>,
30}
31
32const COLS: &str = "id, station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, id_token_type, status, charging_state, \
33    started_at, ended_at, meter_start_wh, meter_stop_wh, stop_reason, reservation_id, remote_start_id, last_seq_no";
34
35/// OCPP 1.6 StartTransaction: the database id becomes the integer transactionId.
36pub async fn start_v16(
37    db: &Db,
38    station_id: i64,
39    connector_id: i32,
40    id_tag: &str,
41    meter_start: i64,
42    at: Ts,
43    reservation_id: Option<i32>,
44) -> sqlx::Result<i64> {
45    sqlx::query_scalar(
46        "WITH n AS (SELECT nextval(pg_get_serial_sequence('transactions', 'id')) AS id) \
47         INSERT INTO transactions (id, station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, status, started_at, \
48           meter_start_wh, reservation_id, last_event_at) OVERRIDING SYSTEM VALUE \
49         SELECT n.id, $1, '1.6', n.id::text, $2, 1, $3, 'Active', $4, $5, $6, $4 FROM n RETURNING id",
50    )
51    .bind(station_id)
52    .bind(connector_id)
53    .bind(id_tag)
54    .bind(at)
55    .bind(meter_start as f64)
56    .bind(reservation_id)
57    .fetch_one(db)
58    .await
59}
60
61/// OCPP 1.6 StopTransaction. Returns `false` if the transaction is unknown.
62pub async fn stop_v16(
63    db: &Db,
64    station_id: i64,
65    transaction_id: i64,
66    meter_stop: i64,
67    at: Ts,
68    reason: &str,
69) -> sqlx::Result<bool> {
70    Ok(sqlx::query(
71        "UPDATE transactions SET status = 'Ended', ended_at = $3, meter_stop_wh = $4, stop_reason = $5, last_event_at = $3, updated_at = now() \
72         WHERE station_id = $1 AND id = $2",
73    )
74    .bind(station_id)
75    .bind(transaction_id)
76    .bind(at)
77    .bind(meter_stop as f64)
78    .bind(reason)
79    .execute(db)
80    .await?
81    .rows_affected()
82        > 0)
83}
84
85/// Version-neutral view of an OCPP 2.x TransactionEvent.
86#[derive(Debug, Default)]
87pub struct TxEvent<'a> {
88    pub version: &'a str,
89    pub uid: &'a str,
90    /// `Started`, `Updated` or `Ended`.
91    pub event_type: &'a str,
92    pub at: Option<Ts>,
93    pub seq_no: i64,
94    pub evse_number: Option<i32>,
95    pub connector_number: Option<i32>,
96    pub id_token: Option<&'a str>,
97    pub id_token_type: Option<&'a str>,
98    pub charging_state: Option<&'a str>,
99    pub stop_reason: Option<&'a str>,
100    pub reservation_id: Option<i32>,
101    pub remote_start_id: Option<i32>,
102    /// Energy.Active.Import.Register in Wh found in the event's meter values.
103    pub energy_wh: Option<f64>,
104}
105
106/// Creates or updates the transaction for a TransactionEvent. Events can
107/// arrive out of order (offline queues), so every field is merged, never reset.
108pub async fn apply_event(db: &Db, station_id: i64, e: &TxEvent<'_>) -> sqlx::Result<i64> {
109    let started = e.event_type == "Started";
110    let ended = e.event_type == "Ended";
111    sqlx::query_scalar(
112        "INSERT INTO transactions (station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, id_token_type, status, \
113           charging_state, started_at, ended_at, meter_start_wh, meter_stop_wh, stop_reason, reservation_id, remote_start_id, last_seq_no, last_event_at) \
114         VALUES ($1, $2, $3, $4, $5, $6, $7, CASE WHEN $8 THEN 'Ended' ELSE 'Active' END, $9, CASE WHEN $10 THEN $11 END, CASE WHEN $8 THEN $11 END, \
115           CASE WHEN $10 THEN $12 END, CASE WHEN $8 THEN $12 END, $13, $14, $15, $16, $11) \
116         ON CONFLICT (station_id, transaction_uid) DO UPDATE SET \
117           evse_number = COALESCE(transactions.evse_number, EXCLUDED.evse_number), \
118           connector_number = COALESCE(EXCLUDED.connector_number, transactions.connector_number), \
119           id_token = COALESCE(EXCLUDED.id_token, transactions.id_token), \
120           id_token_type = COALESCE(EXCLUDED.id_token_type, transactions.id_token_type), \
121           status = CASE WHEN $8 OR transactions.status = 'Ended' THEN 'Ended' ELSE 'Active' END, \
122           charging_state = CASE WHEN transactions.last_seq_no IS NULL OR $16 >= transactions.last_seq_no \
123                            THEN COALESCE(EXCLUDED.charging_state, transactions.charging_state) ELSE transactions.charging_state END, \
124           started_at = COALESCE(transactions.started_at, EXCLUDED.started_at), \
125           ended_at = COALESCE(EXCLUDED.ended_at, transactions.ended_at), \
126           meter_start_wh = COALESCE(transactions.meter_start_wh, EXCLUDED.meter_start_wh), \
127           meter_stop_wh = COALESCE(EXCLUDED.meter_stop_wh, transactions.meter_stop_wh), \
128           stop_reason = COALESCE(EXCLUDED.stop_reason, transactions.stop_reason), \
129           reservation_id = COALESCE(EXCLUDED.reservation_id, transactions.reservation_id), \
130           remote_start_id = COALESCE(EXCLUDED.remote_start_id, transactions.remote_start_id), \
131           last_seq_no = GREATEST(transactions.last_seq_no, EXCLUDED.last_seq_no), \
132           last_event_at = GREATEST(transactions.last_event_at, EXCLUDED.last_event_at), \
133           updated_at = now() \
134         RETURNING id",
135    )
136    .bind(station_id)
137    .bind(e.version)
138    .bind(e.uid)
139    .bind(e.evse_number)
140    .bind(e.connector_number)
141    .bind(e.id_token)
142    .bind(e.id_token_type)
143    .bind(ended)
144    .bind(e.charging_state)
145    .bind(started)
146    .bind(e.at)
147    .bind(e.energy_wh)
148    .bind(e.stop_reason)
149    .bind(e.reservation_id)
150    .bind(e.remote_start_id)
151    .bind(e.seq_no.clamp(0, i32::MAX as i64) as i32)
152    .fetch_one(db)
153    .await
154}
155
156pub async fn find_id_by_uid(db: &Db, station_id: i64, uid: &str) -> sqlx::Result<Option<i64>> {
157    sqlx::query_scalar("SELECT id FROM transactions WHERE station_id = $1 AND transaction_uid = $2")
158        .bind(station_id)
159        .bind(uid)
160        .fetch_optional(db)
161        .await
162}
163
164pub async fn exists_v16(db: &Db, station_id: i64, id: i64) -> sqlx::Result<bool> {
165    Ok(sqlx::query_scalar::<_, i64>(
166        "SELECT id FROM transactions WHERE station_id = $1 AND id = $2",
167    )
168    .bind(station_id)
169    .bind(id)
170    .fetch_optional(db)
171    .await?
172    .is_some())
173}
174
175pub async fn insert_meter_values(
176    db: &Db,
177    station_id: i64,
178    transaction_id: Option<i64>,
179    evse_number: i32,
180    sampled_at: Ts,
181    sampled: &serde_json::Value,
182) -> sqlx::Result<()> {
183    sqlx::query("INSERT INTO meter_values (station_id, transaction_id, evse_number, sampled_at, sampled_values) VALUES ($1, $2, $3, $4, $5)")
184        .bind(station_id)
185        .bind(transaction_id)
186        .bind(evse_number)
187        .bind(sampled_at)
188        .bind(sqlx::types::Json(sampled))
189        .execute(db)
190        .await?;
191    Ok(())
192}
193
194pub async fn get(db: &Db, id: i64) -> sqlx::Result<Option<Transaction>> {
195    sqlx::query_as(&format!("SELECT {COLS} FROM transactions WHERE id = $1"))
196        .bind(id)
197        .fetch_optional(db)
198        .await
199}
200
201pub async fn list_for_station(
202    db: &Db,
203    station_id: i64,
204    active_only: bool,
205    limit: i64,
206) -> sqlx::Result<Vec<Transaction>> {
207    sqlx::query_as(&format!(
208        "SELECT {COLS} FROM transactions WHERE station_id = $1 AND (NOT $2 OR status = 'Active') ORDER BY COALESCE(started_at, created_at) DESC LIMIT $3"
209    ))
210    .bind(station_id)
211    .bind(active_only)
212    .bind(limit)
213    .fetch_all(db)
214    .await
215}
216
217#[derive(Debug, Serialize, sqlx::FromRow)]
218pub struct MeterRow {
219    pub evse_number: i32,
220    pub sampled_at: Ts,
221    pub sampled_values: serde_json::Value,
222}
223
224pub async fn meter_values(db: &Db, transaction_id: i64) -> sqlx::Result<Vec<MeterRow>> {
225    sqlx::query_as("SELECT evse_number, sampled_at, sampled_values FROM meter_values WHERE transaction_id = $1 ORDER BY sampled_at")
226        .bind(transaction_id)
227        .fetch_all(db)
228        .await
229}
230
231/// Active transaction on an EVSE (1.6: connector): its OCPP id and start time.
232pub async fn active_on_evse(
233    db: &Db,
234    station_id: i64,
235    evse_number: i64,
236) -> sqlx::Result<Option<(String, Option<Ts>)>> {
237    sqlx::query_as(
238        "SELECT transaction_uid, started_at FROM transactions WHERE station_id = $1 AND evse_number = $2 AND status = 'Active' \
239         ORDER BY COALESCE(started_at, created_at) DESC LIMIT 1",
240    )
241    .bind(station_id)
242    .bind(evse_number.clamp(i32::MIN as i64, i32::MAX as i64) as i32)
243    .fetch_optional(db)
244    .await
245}
246
247/// Active transaction by OCPP transaction id: (db id, EVSE, start).
248pub async fn active_by_uid(
249    db: &Db,
250    station_id: i64,
251    uid: &str,
252) -> sqlx::Result<Option<(i64, Option<i32>, Option<Ts>)>> {
253    sqlx::query_as("SELECT id, evse_number, started_at FROM transactions WHERE station_id = $1 AND transaction_uid = $2 AND status = 'Active'")
254        .bind(station_id)
255        .bind(uid)
256        .fetch_optional(db)
257        .await
258}
ocpp_service/db/transactions.rs

ocpp_service/db/
transactions.rs
