ocpp_service/tariff/

mod.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)

//! Tariff and cost (OCPP 2.0.1 block I, OCPP 2.1 block I).
//!
//! OCPP 1.6 has no tariff or cost messages; nothing here applies to it.
//!
//! Tariffs use the OCPP 2.1 `TariffType` structure. Per station, `cost_mode` selects:
//! - `Central`: the CSMS calculates costs (2.0.1 I01–I03; 2.1 stations without
//!   local calculation). The driver sees the tariff text in
//!   `IdTokenInfo.personalMessage` and costs in `TransactionEventResponse.totalCost`.
//! - `Local` (2.1 only): the station calculates costs from tariffs it receives
//!   (SetDefaultTariff, AuthorizeResponse.tariff, ChangeTransactionTariff) and
//!   reports `costDetails`; the CSMS recalculates them for verification and
//!   falls back to its own calculation when the station could not calculate.

pub mod cost;

use ocpp_core::{DateTime, OcppVersion};
use ocpp21::Tariff;
use serde::Serialize;
use serde_json::{Value, json};

use crate::calls;
use crate::db::{self, Ts, stations::Station, tariffs::TxCostInfo};
use crate::error::AppError;
use crate::state::App;

pub fn decode(v: &Value) -> Result<Tariff, String> {
    ocpp_core::decode_value(v.clone()).map_err(|e| e.to_string())
}

fn valid_tod(s: &str) -> bool {
    let b = s.as_bytes();
    b.len() == 5
        && b[2] == b':'
        && s[..2].parse::<u32>().is_ok_and(|h| h < 24)
        && s[3..].parse::<u32>().is_ok_and(|m| m < 60)
}

fn valid_date(s: &str) -> bool {
    DateTime::parse(&format!("{s}T00:00:00Z")).is_ok() && s.len() == 10
}

/// startTimeOfDay, endTimeOfDay, validFromDate, validToDate of a condition.
type CalendarFields<'a> = (
    Option<&'a str>,
    Option<&'a str>,
    Option<&'a str>,
    Option<&'a str>,
);

/// Checks a tariff beyond its schema: currency, time and date formats, at least one price.
pub fn validate(t: &Tariff) -> Result<(), String> {
    if !(t.currency.len() == 3 && t.currency.chars().all(|c| c.is_ascii_uppercase())) {
        return Err("currency must be an ISO 4217 code (e.g. EUR)".into());
    }
    if t.energy.is_none()
        && t.charging_time.is_none()
        && t.idle_time.is_none()
        && t.fixed_fee.is_none()
        && t.reservation_time.is_none()
        && t.reservation_fixed.is_none()
    {
        return Err("a tariff needs at least one of energy, chargingTime, idleTime, fixedFee, reservationTime, reservationFixed".into());
    }
    let mut checks: Vec<CalendarFields> = Vec::new();
    if let Some(e) = &t.energy {
        checks.extend(
            e.prices
                .iter()
                .filter_map(|p| p.conditions.as_ref())
                .map(|c| {
                    (
                        c.start_time_of_day.as_deref(),
                        c.end_time_of_day.as_deref(),
                        c.valid_from_date.as_deref(),
                        c.valid_to_date.as_deref(),
                    )
                }),
        );
    }
    for tt in t
        .charging_time
        .iter()
        .chain(&t.idle_time)
        .chain(&t.reservation_time)
    {
        checks.extend(
            tt.prices
                .iter()
                .filter_map(|p| p.conditions.as_ref())
                .map(|c| {
                    (
                        c.start_time_of_day.as_deref(),
                        c.end_time_of_day.as_deref(),
                        c.valid_from_date.as_deref(),
                        c.valid_to_date.as_deref(),
                    )
                }),
        );
    }
    for f in t.fixed_fee.iter().chain(&t.reservation_fixed) {
        checks.extend(
            f.prices
                .iter()
                .filter_map(|p| p.conditions.as_ref())
                .map(|c| {
                    (
                        c.start_time_of_day.as_deref(),
                        c.end_time_of_day.as_deref(),
                        c.valid_from_date.as_deref(),
                        c.valid_to_date.as_deref(),
                    )
                }),
        );
    }
    for (s, e, from, to) in checks {
        if s.is_some_and(|x| !valid_tod(x)) || e.is_some_and(|x| !valid_tod(x)) {
            return Err("startTimeOfDay/endTimeOfDay must be hh:mm (24h)".into());
        }
        if from.is_some_and(|x| !valid_date(x)) || to.is_some_and(|x| !valid_date(x)) {
            return Err("validFromDate/validToDate must be YYYY-MM-DD".into());
        }
    }
    Ok(())
}

/// Human-readable tariff text (I01): the tariff description, or a summary.
pub fn text(t: &Tariff) -> String {
    if let Some(d) = t.description.as_ref().and_then(|d| d.first()) {
        return d.content.clone();
    }
    let mut parts = Vec::new();
    let c = &t.currency;
    let suffix = |has_cond: bool| if has_cond { " (conditions apply)" } else { "" };
    if let Some(f) = t.fixed_fee.as_ref().and_then(|f| f.prices.first()) {
        parts.push(format!(
            "{:.2} {c} start fee{}",
            f.price_fixed,
            suffix(f.conditions.is_some())
        ));
    }
    if let Some(e) = t.energy.as_ref().and_then(|e| e.prices.first()) {
        parts.push(format!(
            "{:.2} {c}/kWh{}",
            e.price_kwh,
            suffix(e.conditions.is_some())
        ));
    }
    if let Some(p) = t.charging_time.as_ref().and_then(|x| x.prices.first()) {
        parts.push(format!(
            "{:.2} {c}/min charging{}",
            p.price_minute,
            suffix(p.conditions.is_some())
        ));
    }
    if let Some(p) = t.idle_time.as_ref().and_then(|x| x.prices.first()) {
        parts.push(format!(
            "{:.2} {c}/min idle{}",
            p.price_minute,
            suffix(p.conditions.is_some())
        ));
    }
    parts.join(", ")
}

pub async fn load(app: &App, tariff_id: &str) -> Result<Option<Tariff>, AppError> {
    Ok(db::tariffs::get(&app.db, tariff_id)
        .await?
        .and_then(|row| decode(&row.tariff).ok()))
}

/// Tariff for a driver at an EVSE: the driver-specific tariff, else the default one.
pub async fn applicable(
    app: &App,
    station_id: i64,
    evse: i32,
    id_token: Option<(&str, Option<&str>)>,
    at: DateTime,
) -> Result<Option<Tariff>, AppError> {
    if let Some((token, ty)) = id_token
        && let Some(id) = db::tariffs::driver_tariff(&app.db, token, ty).await?
        && let Some(t) = load(app, &id).await?
    {
        return Ok(Some(t));
    }
    match db::tariffs::default_for(&app.db, station_id, evse, Ts(at)).await? {
        Some(id) => load(app, &id).await,
        None => Ok(None),
    }
}

pub async fn driver_tariff(
    app: &App,
    token: &str,
    ty: Option<&str>,
) -> Result<Option<Tariff>, AppError> {
    match db::tariffs::driver_tariff(&app.db, token, ty).await? {
        Some(id) => load(app, &id).await,
        None => Ok(None),
    }
}

/// What the station should be told at Authorize, depending on its cost mode.
pub struct AuthorizeExtras {
    /// I01: tariff text for `IdTokenInfo.personalMessage` (central cost).
    pub personal_message: Option<String>,
    /// I08 (2.1 local cost): driver-specific tariff for `AuthorizeResponse.tariff`, without validFrom.
    pub tariff: Option<Tariff>,
}

pub async fn on_authorize(
    app: &App,
    station_id: i64,
    version: OcppVersion,
    token: &str,
    ty: Option<&str>,
) -> Result<AuthorizeExtras, AppError> {
    let none = AuthorizeExtras {
        personal_message: None,
        tariff: None,
    };
    let Some(station) = db::stations::get(&app.db, station_id).await? else {
        return Ok(none);
    };
    match (station.cost_mode.as_str(), version) {
        ("Central", OcppVersion::V201 | OcppVersion::V21) => {
            let t = applicable(app, station_id, 0, Some((token, ty)), DateTime::now()).await?;
            Ok(AuthorizeExtras {
                personal_message: t.as_ref().map(text),
                tariff: None,
            })
        }
        ("Local", OcppVersion::V21) => {
            let t = driver_tariff(app, token, ty).await?.map(|mut t| {
                t.valid_from = None; // I08.FR.09
                t
            });
            Ok(AuthorizeExtras {
                personal_message: None,
                tariff: t,
            })
        }
        _ => Ok(none),
    }
}

/// Version-neutral content of a 2.x TransactionEvent relevant for cost.
#[derive(Debug, Default)]
pub struct TxEventCost<'a> {
    pub event_type: &'a str,
    pub at: DateTime,
    pub seq_no: i64,
    pub charging_state: Option<&'a str>,
    pub energy_wh: Option<f64>,
    pub power_w: Option<f64>,
    pub current_a: Option<f64>,
    /// 2.1 transactionInfo.tariffId.
    pub tariff_id: Option<&'a str>,
    /// 2.1 costDetails (local cost calculation).
    pub cost_details: Option<Value>,
    pub payment_brand: Option<&'a str>,
    pub payment_recognition: Option<&'a str>,
}

fn round2(v: f64) -> f64 {
    (v * 100.0).round() / 100.0
}

fn zone_for(app: &App, station: &Station) -> std::sync::Arc<crate::tz::Zone> {
    app.zones.get(
        &app.cfg.zoneinfo_dir,
        station
            .timezone
            .as_deref()
            .unwrap_or(&app.cfg.default_timezone),
    )
}

/// Calculates the cost of a transaction from its samples and tariff(s).
pub async fn calculate(
    app: &App,
    station: &Station,
    tx: &TxCostInfo,
    final_: bool,
) -> Result<Option<ocpp21::CostDetails>, AppError> {
    let Some(first_id) = &tx.tariff_id else {
        return Ok(None);
    };
    let Some(first) = load(app, first_id).await? else {
        return Ok(None);
    };
    let rows = db::tariffs::samples(&app.db, tx.id).await?;
    let start = tx
        .started_at
        .map(|t| t.0)
        .or(rows.first().map(|r| r.at.0))
        .unwrap_or_else(DateTime::now);
    let end = if final_ {
        tx.ended_at
            .or(tx.last_event_at)
            .map(|t| t.0)
            .unwrap_or_else(DateTime::now)
    } else {
        DateTime::now().max(start)
    };
    // Tariff changes reported by the station (2.1 TariffChanged) apply from that moment on (I11.FR.08).
    let mut tariffs = vec![(start, first)];
    let mut current_id = first_id.clone();
    for r in &rows {
        if let Some(id) = &r.tariff_id
            && *id != current_id
            && let Some(t) = load(app, id).await?
        {
            tariffs.push((r.at.0, t));
            current_id = id.clone();
        }
    }
    let samples: Vec<cost::Sample> = rows
        .iter()
        .map(|r| cost::Sample {
            at: r.at.0,
            energy_wh: r.energy_wh,
            power_w: r.power_w,
            current_a: r.current_a,
            charging_state: r.charging_state.clone(),
        })
        .collect();
    let zone = zone_for(app, station);
    let input = cost::Input {
        tariffs: &tariffs,
        start,
        end: end.max(start),
        samples: &samples,
        zone: &zone,
        evse_kind: None,
        payment_brand: tx.payment_brand.as_deref(),
        payment_recognition: tx.payment_recognition.as_deref(),
        reserved_at: tx.reserved_at.map(|t| t.0),
        final_,
    };
    Ok(Some(cost::calculate(&input)))
}

/// Records a TransactionEvent for cost purposes and returns the `totalCost`
/// to put in the TransactionEventResponse, if any (I02, I03, I12).
pub async fn on_transaction_event(
    app: &App,
    station_id: i64,
    transaction_id: i64,
    e: &TxEventCost<'_>,
) -> Result<Option<f64>, AppError> {
    db::tariffs::insert_sample(
        &app.db,
        transaction_id,
        &db::tariffs::NewSample {
            at: Some(Ts(e.at)),
            seq_no: Some(e.seq_no.clamp(0, i32::MAX as i64) as i32),
            event_type: Some(e.event_type),
            charging_state: e.charging_state,
            energy_wh: e.energy_wh,
            power_w: e.power_w,
            current_a: e.current_a,
            tariff_id: e.tariff_id,
        },
    )
    .await?;
    if e.payment_brand.is_some() || e.payment_recognition.is_some() {
        db::tariffs::set_payment(
            &app.db,
            transaction_id,
            e.payment_brand,
            e.payment_recognition,
        )
        .await?;
    }
    let Some(station) = db::stations::get(&app.db, station_id).await? else {
        return Ok(None);
    };
    if station.cost_mode == "None" {
        return Ok(None);
    }
    let Some(tx) = db::tariffs::tx_info(&app.db, transaction_id).await? else {
        return Ok(None);
    };

    // The transaction's tariff, fixed at its start (I07.FR.24).
    if tx.tariff_id.is_none() {
        let tariff = match e.tariff_id {
            Some(id) => load(app, id).await?,
            None => {
                let token = tx
                    .id_token
                    .as_deref()
                    .map(|t| (t, tx.id_token_type.as_deref()));
                applicable(
                    app,
                    station_id,
                    tx.evse_number.unwrap_or(0),
                    token,
                    tx.started_at.map(|t| t.0).unwrap_or(e.at),
                )
                .await?
            }
        };
        if let Some(t) = tariff {
            db::tariffs::set_tx_tariff_once(&app.db, transaction_id, &t.tariff_id, &t.currency)
                .await?;
        }
    }
    let ended = e.event_type == "Ended";
    if let Some(details) = &e.cost_details {
        db::tariffs::set_station_cost(&app.db, transaction_id, details).await?;
    }
    let tx = db::tariffs::tx_info(&app.db, transaction_id)
        .await?
        .expect("exists");

    // CSMS calculation (central cost, verification of local cost).
    let csms = calculate(app, &station, &tx, ended).await?;
    if let Some(c) = &csms {
        let v = serde_json::to_value(c).map_err(|e| AppError::Internal(e.to_string()))?;
        db::tariffs::set_cost(
            &app.db,
            transaction_id,
            &v,
            &c.total_cost.currency,
            c.total_cost.total.excl_tax,
            c.total_cost.total.incl_tax,
        )
        .await?;
    }
    // Total of the station when it calculated successfully (local cost), else the CSMS one.
    let station_total = e
        .cost_details
        .as_ref()
        .filter(|d| d.get("failureToCalculate") != Some(&Value::Bool(true)))
        .and_then(|d| {
            d["totalCost"]["total"]["inclTax"]
                .as_f64()
                .or(d["totalCost"]["total"]["exclTax"].as_f64())
        });
    if let (Some(st), Some(c)) = (station_total, &csms)
        && let Some(own) = c.total_cost.total.incl_tax
        && (st - own).abs() > 0.01
    {
        tracing::warn!(station = %station.ocpp_identity, transaction = %tx.transaction_uid, station_total = st, csms_total = own, "station cost differs from the CSMS calculation");
    }
    let total = station_total.or_else(|| csms.as_ref().and_then(|c| c.total_cost.total.incl_tax));
    Ok(match (ended, station.cost_mode.as_str()) {
        (true, _) => total.map(round2),
        // I02 (alternative scenario): running cost in the response to Updated events.
        (false, "Central") if e.event_type == "Updated" => total.map(round2),
        _ => None,
    })
}

/// Cost of a transaction now (running) or final, plus what the station reported.
#[derive(Debug, Serialize)]
pub struct CostReport {
    pub transaction_id: i64,
    pub tariff_id: Option<String>,
    pub final_: bool,
    pub csms: Option<ocpp21::CostDetails>,
    pub station: Option<Value>,
}

pub async fn report(app: &App, transaction_id: i64) -> Result<CostReport, AppError> {
    let tx = db::tariffs::tx_info(&app.db, transaction_id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("transaction {transaction_id} not found")))?;
    let station = db::stations::get(&app.db, tx.station_id)
        .await?
        .ok_or_else(|| AppError::NotFound("station".into()))?;
    let final_ = tx.status == "Ended";
    let csms = calculate(app, &station, &tx, final_).await?;
    Ok(CostReport {
        transaction_id,
        tariff_id: tx.tariff_id.clone(),
        final_,
        csms,
        station: tx.station_cost_details,
    })
}

/// Assigns a default tariff; for 2.1 local-cost stations it is sent with SetDefaultTariff (I07).
pub async fn set_default(
    app: &App,
    station: &Station,
    evse: i64,
    tariff_id: &str,
) -> Result<Value, AppError> {
    let row = db::tariffs::get(&app.db, tariff_id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("tariff {tariff_id} not found")))?;
    let tariff = decode(&row.tariff).map_err(AppError::Internal)?;
    let evse32 = evse.clamp(0, i32::MAX as i64) as i32;
    let valid_from = tariff.valid_from.map(Ts);
    if station.cost_mode == "Local" {
        let version = calls::connected_version(app, &station.ocpp_identity).await?;
        if version != OcppVersion::V21 {
            return Err(AppError::BadRequest(
                "local cost calculation (SetDefaultTariff) requires OCPP 2.1".into(),
            ));
        }
        let response = calls::call(
            app,
            &station.ocpp_identity,
            version,
            "SetDefaultTariff",
            json!({ "evseId": evse, "tariff": row.tariff }),
        )
        .await?;
        let status = calls::status_of(&response).to_owned();
        if status == "Accepted" {
            db::tariffs::assign_default(
                &app.db,
                station.id,
                evse32,
                tariff_id,
                valid_from,
                Some(&status),
            )
            .await?;
        }
        return Ok(json!({ "status": status, "response": response }));
    }
    db::tariffs::assign_default(&app.db, station.id, evse32, tariff_id, valid_from, None).await?;
    Ok(json!({ "status": "Accepted" }))
}

/// Removes a default tariff; for 2.1 local-cost stations with ClearTariffs (I10).
pub async fn clear_default(
    app: &App,
    station: &Station,
    tariff_id: &str,
    evse: Option<i64>,
) -> Result<Value, AppError> {
    let evse32 = evse.map(|e| e.clamp(0, i32::MAX as i64) as i32);
    if station.cost_mode == "Local" {
        let version = calls::connected_version(app, &station.ocpp_identity).await?;
        if version == OcppVersion::V21 {
            let payload =
                crate::smart::strip_nulls(json!({ "tariffIds": [tariff_id], "evseId": evse }));
            let response = calls::call(
                app,
                &station.ocpp_identity,
                version,
                "ClearTariffs",
                payload,
            )
            .await?;
            let cleared = response["clearTariffsResult"].as_array().is_some_and(|r| {
                r.iter()
                    .any(|x| x["status"] == "Accepted" || x["status"] == "NoTariff")
            });
            if cleared {
                db::tariffs::remove_default(&app.db, station.id, tariff_id, evse32).await?;
            }
            return Ok(response);
        }
    }
    let n = db::tariffs::remove_default(&app.db, station.id, tariff_id, evse32).await?;
    Ok(json!({ "removed": n }))
}

/// I11: changes the tariff of an ongoing transaction (2.1 local cost calculation).
pub async fn change_transaction_tariff(
    app: &App,
    transaction_id: i64,
    tariff_id: &str,
) -> Result<Value, AppError> {
    let tx = db::tariffs::tx_info(&app.db, transaction_id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("transaction {transaction_id} not found")))?;
    if tx.status != "Active" {
        return Err(AppError::Conflict(
            "the transaction is not active (I11.FR.04)".into(),
        ));
    }
    let station = db::stations::get(&app.db, tx.station_id)
        .await?
        .ok_or_else(|| AppError::NotFound("station".into()))?;
    let version = calls::connected_version(app, &station.ocpp_identity).await?;
    if version != OcppVersion::V21 {
        return Err(AppError::BadRequest(
            "ChangeTransactionTariff requires OCPP 2.1".into(),
        ));
    }
    let row = db::tariffs::get(&app.db, tariff_id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("tariff {tariff_id} not found")))?;
    if let Some(cur) = &tx.currency
        && *cur != row.currency
    {
        return Err(AppError::Conflict(format!(
            "the currency cannot change during a transaction ({cur} → {}) (I11.FR.05)",
            row.currency
        )));
    }
    calls::call(
        app,
        &station.ocpp_identity,
        version,
        "ChangeTransactionTariff",
        json!({ "transactionId": tx.transaction_uid, "tariff": row.tariff }),
    )
    .await
}