ocpp_service/db/
tariffs.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Tariffs, tariff assignments, transaction samples and costs.
5
6use serde::Serialize;
7use serde_json::Value;
8
9use super::{Db, Ts};
10
11#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
12pub struct TariffRow {
13    pub tariff_id: String,
14    pub currency: String,
15    pub tariff: Value,
16    pub created_at: Ts,
17}
18
19pub enum Upsert {
20    Created(TariffRow),
21    Unchanged(TariffRow),
22    /// The id exists with different content (I08.FR.08: a tariff id never changes content).
23    Conflict,
24}
25
26pub async fn upsert(
27    db: &Db,
28    tariff_id: &str,
29    currency: &str,
30    tariff: &Value,
31) -> sqlx::Result<Upsert> {
32    let created: Option<TariffRow> = sqlx::query_as(
33        "INSERT INTO tariffs (tariff_id, currency, tariff) VALUES ($1, $2, $3) ON CONFLICT (tariff_id) DO NOTHING RETURNING tariff_id, currency, tariff, created_at",
34    )
35    .bind(tariff_id)
36    .bind(currency)
37    .bind(sqlx::types::Json(tariff))
38    .fetch_optional(db)
39    .await?;
40    if let Some(row) = created {
41        return Ok(Upsert::Created(row));
42    }
43    let existing = get(db, tariff_id).await?.expect("conflicting row exists");
44    Ok(if &existing.tariff == tariff {
45        Upsert::Unchanged(existing)
46    } else {
47        Upsert::Conflict
48    })
49}
50
51pub async fn get(db: &Db, tariff_id: &str) -> sqlx::Result<Option<TariffRow>> {
52    sqlx::query_as(
53        "SELECT tariff_id, currency, tariff, created_at FROM tariffs WHERE tariff_id = $1",
54    )
55    .bind(tariff_id)
56    .fetch_optional(db)
57    .await
58}
59
60pub async fn list(db: &Db) -> sqlx::Result<Vec<TariffRow>> {
61    sqlx::query_as("SELECT tariff_id, currency, tariff, created_at FROM tariffs ORDER BY tariff_id")
62        .fetch_all(db)
63        .await
64}
65
66pub async fn delete(db: &Db, tariff_id: &str) -> sqlx::Result<bool> {
67    Ok(sqlx::query("DELETE FROM tariffs WHERE tariff_id = $1")
68        .bind(tariff_id)
69        .execute(db)
70        .await?
71        .rows_affected()
72        > 0)
73}
74
75#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
76pub struct Assignment {
77    pub evse_number: i32,
78    pub tariff_id: String,
79    pub valid_from: Option<Ts>,
80    pub install_status: Option<String>,
81    pub updated_at: Ts,
82}
83
84/// Assigns a default tariff to an EVSE (0 = all EVSEs). A tariff that is valid
85/// now replaces the currently valid ones; a future one is kept next to them
86/// until it becomes valid (I07.FR.20–23).
87pub async fn assign_default(
88    db: &Db,
89    station_id: i64,
90    evse: i32,
91    tariff_id: &str,
92    valid_from: Option<Ts>,
93    install_status: Option<&str>,
94) -> sqlx::Result<()> {
95    let mut tx = db.begin().await?;
96    let now = Ts(ocpp_core::DateTime::now());
97    if valid_from.is_none_or(|v| v <= now) {
98        sqlx::query("DELETE FROM station_tariffs WHERE station_id = $1 AND evse_number = $2 AND tariff_id <> $3 AND (valid_from IS NULL OR valid_from <= $4)")
99            .bind(station_id)
100            .bind(evse)
101            .bind(tariff_id)
102            .bind(now)
103            .execute(&mut *tx)
104            .await?;
105    }
106    sqlx::query(
107        "INSERT INTO station_tariffs (station_id, evse_number, tariff_id, valid_from, install_status) VALUES ($1, $2, $3, $4, $5) \
108         ON CONFLICT (station_id, evse_number, tariff_id) DO UPDATE SET valid_from = EXCLUDED.valid_from, install_status = EXCLUDED.install_status, updated_at = now()",
109    )
110    .bind(station_id)
111    .bind(evse)
112    .bind(tariff_id)
113    .bind(valid_from)
114    .bind(install_status)
115    .execute(&mut *tx)
116    .await?;
117    tx.commit().await
118}
119
120pub async fn remove_default(
121    db: &Db,
122    station_id: i64,
123    tariff_id: &str,
124    evse: Option<i32>,
125) -> sqlx::Result<u64> {
126    Ok(sqlx::query("DELETE FROM station_tariffs WHERE station_id = $1 AND tariff_id = $2 AND ($3::int IS NULL OR evse_number = $3)")
127        .bind(station_id)
128        .bind(tariff_id)
129        .bind(evse)
130        .execute(db)
131        .await?
132        .rows_affected())
133}
134
135pub async fn defaults(db: &Db, station_id: i64) -> sqlx::Result<Vec<Assignment>> {
136    sqlx::query_as("SELECT evse_number, tariff_id, valid_from, install_status, updated_at FROM station_tariffs WHERE station_id = $1 ORDER BY evse_number, valid_from NULLS FIRST")
137        .bind(station_id)
138        .fetch_all(db)
139        .await
140}
141
142/// Default tariff valid at `at` for an EVSE: EVSE-specific before EVSE 0, latest validFrom first.
143pub async fn default_for(
144    db: &Db,
145    station_id: i64,
146    evse: i32,
147    at: Ts,
148) -> sqlx::Result<Option<String>> {
149    sqlx::query_scalar(
150        "SELECT tariff_id FROM station_tariffs WHERE station_id = $1 AND evse_number IN ($2, 0) AND (valid_from IS NULL OR valid_from <= $3) \
151         AND (install_status IS NULL OR install_status = 'Accepted') ORDER BY (evse_number = $2) DESC, valid_from DESC NULLS LAST LIMIT 1",
152    )
153    .bind(station_id)
154    .bind(evse)
155    .bind(at)
156    .fetch_optional(db)
157    .await
158}
159
160/// Driver-specific tariff of an id token (1.6-style case-insensitive match, any type).
161pub async fn driver_tariff(
162    db: &Db,
163    id_token: &str,
164    token_type: Option<&str>,
165) -> sqlx::Result<Option<String>> {
166    sqlx::query_scalar(
167        "SELECT tariff_id FROM id_tokens WHERE tariff_id IS NOT NULL AND lower(id_token) = lower($1) AND ($2::text IS NULL OR token_type IN ($2, '*')) \
168         ORDER BY (token_type = '*') ASC LIMIT 1",
169    )
170    .bind(id_token)
171    .bind(token_type)
172    .fetch_optional(db)
173    .await
174}
175
176#[derive(Debug, Clone, sqlx::FromRow)]
177pub struct TxCostInfo {
178    pub id: i64,
179    pub station_id: i64,
180    pub transaction_uid: String,
181    pub status: String,
182    pub evse_number: Option<i32>,
183    pub id_token: Option<String>,
184    pub id_token_type: Option<String>,
185    pub started_at: Option<Ts>,
186    pub ended_at: Option<Ts>,
187    pub last_event_at: Option<Ts>,
188    pub tariff_id: Option<String>,
189    pub payment_brand: Option<String>,
190    pub payment_recognition: Option<String>,
191    pub reserved_at: Option<Ts>,
192    pub cost_details: Option<Value>,
193    pub station_cost_details: Option<Value>,
194    pub currency: Option<String>,
195    pub cost_excl_tax: Option<f64>,
196    pub cost_incl_tax: Option<f64>,
197}
198
199pub async fn tx_info(db: &Db, transaction_id: i64) -> sqlx::Result<Option<TxCostInfo>> {
200    sqlx::query_as(
201        "SELECT t.id, t.station_id, t.transaction_uid, t.status, t.evse_number, t.id_token, t.id_token_type, t.started_at, t.ended_at, t.last_event_at, \
202           t.tariff_id, t.payment_brand, t.payment_recognition, r.created_at AS reserved_at, t.cost_details, t.station_cost_details, t.currency, \
203           t.cost_excl_tax, t.cost_incl_tax \
204         FROM transactions t LEFT JOIN reservations r ON r.id = t.reservation_db_id WHERE t.id = $1",
205    )
206    .bind(transaction_id)
207    .fetch_optional(db)
208    .await
209}
210
211/// Sets the tariff of a transaction once (the tariff does not change during a transaction, I07.FR.24).
212pub async fn set_tx_tariff_once(
213    db: &Db,
214    transaction_id: i64,
215    tariff_id: &str,
216    currency: &str,
217) -> sqlx::Result<()> {
218    sqlx::query(
219        "UPDATE transactions SET tariff_id = $2, currency = $3 WHERE id = $1 AND tariff_id IS NULL",
220    )
221    .bind(transaction_id)
222    .bind(tariff_id)
223    .bind(currency)
224    .execute(db)
225    .await?;
226    Ok(())
227}
228
229pub async fn set_payment(
230    db: &Db,
231    transaction_id: i64,
232    brand: Option<&str>,
233    recognition: Option<&str>,
234) -> sqlx::Result<()> {
235    sqlx::query("UPDATE transactions SET payment_brand = COALESCE($2, payment_brand), payment_recognition = COALESCE($3, payment_recognition) WHERE id = $1")
236        .bind(transaction_id)
237        .bind(brand)
238        .bind(recognition)
239        .execute(db)
240        .await?;
241    Ok(())
242}
243
244pub async fn set_cost(
245    db: &Db,
246    transaction_id: i64,
247    details: &Value,
248    currency: &str,
249    excl: Option<f64>,
250    incl: Option<f64>,
251) -> sqlx::Result<()> {
252    sqlx::query("UPDATE transactions SET cost_details = $2, currency = $3, cost_excl_tax = $4, cost_incl_tax = $5 WHERE id = $1")
253        .bind(transaction_id)
254        .bind(sqlx::types::Json(details))
255        .bind(currency)
256        .bind(excl)
257        .bind(incl)
258        .execute(db)
259        .await?;
260    Ok(())
261}
262
263/// Cost details reported by a station doing local cost calculation (I12).
264pub async fn set_station_cost(db: &Db, transaction_id: i64, details: &Value) -> sqlx::Result<()> {
265    sqlx::query("UPDATE transactions SET station_cost_details = $2 WHERE id = $1")
266        .bind(transaction_id)
267        .bind(sqlx::types::Json(details))
268        .execute(db)
269        .await?;
270    Ok(())
271}
272
273#[derive(Debug, Clone, Default)]
274pub struct NewSample<'a> {
275    pub at: Option<Ts>,
276    pub seq_no: Option<i32>,
277    pub event_type: Option<&'a str>,
278    pub charging_state: Option<&'a str>,
279    pub energy_wh: Option<f64>,
280    pub power_w: Option<f64>,
281    pub current_a: Option<f64>,
282    pub tariff_id: Option<&'a str>,
283}
284
285pub async fn insert_sample(db: &Db, transaction_id: i64, s: &NewSample<'_>) -> sqlx::Result<()> {
286    sqlx::query(
287        "INSERT INTO transaction_samples (transaction_id, at, seq_no, event_type, charging_state, energy_wh, power_w, current_a, tariff_id) \
288         VALUES ($1, COALESCE($2, now()), $3, $4, $5, $6, $7, $8, $9)",
289    )
290    .bind(transaction_id)
291    .bind(s.at)
292    .bind(s.seq_no)
293    .bind(s.event_type)
294    .bind(s.charging_state)
295    .bind(s.energy_wh)
296    .bind(s.power_w)
297    .bind(s.current_a)
298    .bind(s.tariff_id)
299    .execute(db)
300    .await?;
301    Ok(())
302}
303
304#[derive(Debug, Clone, sqlx::FromRow)]
305pub struct SampleRow {
306    pub at: Ts,
307    pub charging_state: Option<String>,
308    pub energy_wh: Option<f64>,
309    pub power_w: Option<f64>,
310    pub current_a: Option<f64>,
311    pub tariff_id: Option<String>,
312}
313
314pub async fn samples(db: &Db, transaction_id: i64) -> sqlx::Result<Vec<SampleRow>> {
315    sqlx::query_as("SELECT at, charging_state, energy_wh, power_w, current_a, tariff_id FROM transaction_samples WHERE transaction_id = $1 ORDER BY at, seq_no NULLS FIRST")
316        .bind(transaction_id)
317        .fetch_all(db)
318        .await
319}
ocpp_service/db/tariffs.rs

ocpp_service/db/
tariffs.rs
