codetics-ocpp-service 0.1.0

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

//! Charging transactions and meter values.

use serde::Serialize;

use super::{Db, Ts};

#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct Transaction {
    pub id: i64,
    pub station_id: i64,
    pub ocpp_version: String,
    pub transaction_uid: String,
    pub evse_number: Option<i32>,
    pub connector_number: Option<i32>,
    pub id_token: Option<String>,
    pub id_token_type: Option<String>,
    pub status: String,
    pub charging_state: Option<String>,
    pub started_at: Option<Ts>,
    pub ended_at: Option<Ts>,
    pub meter_start_wh: Option<f64>,
    pub meter_stop_wh: Option<f64>,
    pub stop_reason: Option<String>,
    pub reservation_id: Option<i32>,
    pub remote_start_id: Option<i32>,
    pub last_seq_no: Option<i32>,
}

const COLS: &str = "id, station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, id_token_type, status, charging_state, \
    started_at, ended_at, meter_start_wh, meter_stop_wh, stop_reason, reservation_id, remote_start_id, last_seq_no";

/// OCPP 1.6 StartTransaction: the database id becomes the integer transactionId.
pub async fn start_v16(
    db: &Db,
    station_id: i64,
    connector_id: i32,
    id_tag: &str,
    meter_start: i64,
    at: Ts,
    reservation_id: Option<i32>,
) -> sqlx::Result<i64> {
    sqlx::query_scalar(
        "WITH n AS (SELECT nextval(pg_get_serial_sequence('transactions', 'id')) AS id) \
         INSERT INTO transactions (id, station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, status, started_at, \
           meter_start_wh, reservation_id, last_event_at) OVERRIDING SYSTEM VALUE \
         SELECT n.id, $1, '1.6', n.id::text, $2, 1, $3, 'Active', $4, $5, $6, $4 FROM n RETURNING id",
    )
    .bind(station_id)
    .bind(connector_id)
    .bind(id_tag)
    .bind(at)
    .bind(meter_start as f64)
    .bind(reservation_id)
    .fetch_one(db)
    .await
}

/// OCPP 1.6 StopTransaction. Returns `false` if the transaction is unknown.
pub async fn stop_v16(
    db: &Db,
    station_id: i64,
    transaction_id: i64,
    meter_stop: i64,
    at: Ts,
    reason: &str,
) -> sqlx::Result<bool> {
    Ok(sqlx::query(
        "UPDATE transactions SET status = 'Ended', ended_at = $3, meter_stop_wh = $4, stop_reason = $5, last_event_at = $3, updated_at = now() \
         WHERE station_id = $1 AND id = $2",
    )
    .bind(station_id)
    .bind(transaction_id)
    .bind(at)
    .bind(meter_stop as f64)
    .bind(reason)
    .execute(db)
    .await?
    .rows_affected()
        > 0)
}

/// Version-neutral view of an OCPP 2.x TransactionEvent.
#[derive(Debug, Default)]
pub struct TxEvent<'a> {
    pub version: &'a str,
    pub uid: &'a str,
    /// `Started`, `Updated` or `Ended`.
    pub event_type: &'a str,
    pub at: Option<Ts>,
    pub seq_no: i64,
    pub evse_number: Option<i32>,
    pub connector_number: Option<i32>,
    pub id_token: Option<&'a str>,
    pub id_token_type: Option<&'a str>,
    pub charging_state: Option<&'a str>,
    pub stop_reason: Option<&'a str>,
    pub reservation_id: Option<i32>,
    pub remote_start_id: Option<i32>,
    /// Energy.Active.Import.Register in Wh found in the event's meter values.
    pub energy_wh: Option<f64>,
}

/// Creates or updates the transaction for a TransactionEvent. Events can
/// arrive out of order (offline queues), so every field is merged, never reset.
pub async fn apply_event(db: &Db, station_id: i64, e: &TxEvent<'_>) -> sqlx::Result<i64> {
    let started = e.event_type == "Started";
    let ended = e.event_type == "Ended";
    sqlx::query_scalar(
        "INSERT INTO transactions (station_id, ocpp_version, transaction_uid, evse_number, connector_number, id_token, id_token_type, status, \
           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) \
         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, \
           CASE WHEN $10 THEN $12 END, CASE WHEN $8 THEN $12 END, $13, $14, $15, $16, $11) \
         ON CONFLICT (station_id, transaction_uid) DO UPDATE SET \
           evse_number = COALESCE(transactions.evse_number, EXCLUDED.evse_number), \
           connector_number = COALESCE(EXCLUDED.connector_number, transactions.connector_number), \
           id_token = COALESCE(EXCLUDED.id_token, transactions.id_token), \
           id_token_type = COALESCE(EXCLUDED.id_token_type, transactions.id_token_type), \
           status = CASE WHEN $8 OR transactions.status = 'Ended' THEN 'Ended' ELSE 'Active' END, \
           charging_state = CASE WHEN transactions.last_seq_no IS NULL OR $16 >= transactions.last_seq_no \
                            THEN COALESCE(EXCLUDED.charging_state, transactions.charging_state) ELSE transactions.charging_state END, \
           started_at = COALESCE(transactions.started_at, EXCLUDED.started_at), \
           ended_at = COALESCE(EXCLUDED.ended_at, transactions.ended_at), \
           meter_start_wh = COALESCE(transactions.meter_start_wh, EXCLUDED.meter_start_wh), \
           meter_stop_wh = COALESCE(EXCLUDED.meter_stop_wh, transactions.meter_stop_wh), \
           stop_reason = COALESCE(EXCLUDED.stop_reason, transactions.stop_reason), \
           reservation_id = COALESCE(EXCLUDED.reservation_id, transactions.reservation_id), \
           remote_start_id = COALESCE(EXCLUDED.remote_start_id, transactions.remote_start_id), \
           last_seq_no = GREATEST(transactions.last_seq_no, EXCLUDED.last_seq_no), \
           last_event_at = GREATEST(transactions.last_event_at, EXCLUDED.last_event_at), \
           updated_at = now() \
         RETURNING id",
    )
    .bind(station_id)
    .bind(e.version)
    .bind(e.uid)
    .bind(e.evse_number)
    .bind(e.connector_number)
    .bind(e.id_token)
    .bind(e.id_token_type)
    .bind(ended)
    .bind(e.charging_state)
    .bind(started)
    .bind(e.at)
    .bind(e.energy_wh)
    .bind(e.stop_reason)
    .bind(e.reservation_id)
    .bind(e.remote_start_id)
    .bind(e.seq_no.clamp(0, i32::MAX as i64) as i32)
    .fetch_one(db)
    .await
}

pub async fn find_id_by_uid(db: &Db, station_id: i64, uid: &str) -> sqlx::Result<Option<i64>> {
    sqlx::query_scalar("SELECT id FROM transactions WHERE station_id = $1 AND transaction_uid = $2")
        .bind(station_id)
        .bind(uid)
        .fetch_optional(db)
        .await
}

pub async fn exists_v16(db: &Db, station_id: i64, id: i64) -> sqlx::Result<bool> {
    Ok(sqlx::query_scalar::<_, i64>(
        "SELECT id FROM transactions WHERE station_id = $1 AND id = $2",
    )
    .bind(station_id)
    .bind(id)
    .fetch_optional(db)
    .await?
    .is_some())
}

pub async fn insert_meter_values(
    db: &Db,
    station_id: i64,
    transaction_id: Option<i64>,
    evse_number: i32,
    sampled_at: Ts,
    sampled: &serde_json::Value,
) -> sqlx::Result<()> {
    sqlx::query("INSERT INTO meter_values (station_id, transaction_id, evse_number, sampled_at, sampled_values) VALUES ($1, $2, $3, $4, $5)")
        .bind(station_id)
        .bind(transaction_id)
        .bind(evse_number)
        .bind(sampled_at)
        .bind(sqlx::types::Json(sampled))
        .execute(db)
        .await?;
    Ok(())
}

pub async fn get(db: &Db, id: i64) -> sqlx::Result<Option<Transaction>> {
    sqlx::query_as(&format!("SELECT {COLS} FROM transactions WHERE id = $1"))
        .bind(id)
        .fetch_optional(db)
        .await
}

pub async fn list_for_station(
    db: &Db,
    station_id: i64,
    active_only: bool,
    limit: i64,
) -> sqlx::Result<Vec<Transaction>> {
    sqlx::query_as(&format!(
        "SELECT {COLS} FROM transactions WHERE station_id = $1 AND (NOT $2 OR status = 'Active') ORDER BY COALESCE(started_at, created_at) DESC LIMIT $3"
    ))
    .bind(station_id)
    .bind(active_only)
    .bind(limit)
    .fetch_all(db)
    .await
}

#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct MeterRow {
    pub evse_number: i32,
    pub sampled_at: Ts,
    pub sampled_values: serde_json::Value,
}

pub async fn meter_values(db: &Db, transaction_id: i64) -> sqlx::Result<Vec<MeterRow>> {
    sqlx::query_as("SELECT evse_number, sampled_at, sampled_values FROM meter_values WHERE transaction_id = $1 ORDER BY sampled_at")
        .bind(transaction_id)
        .fetch_all(db)
        .await
}

/// Active transaction on an EVSE (1.6: connector): its OCPP id and start time.
pub async fn active_on_evse(
    db: &Db,
    station_id: i64,
    evse_number: i64,
) -> sqlx::Result<Option<(String, Option<Ts>)>> {
    sqlx::query_as(
        "SELECT transaction_uid, started_at FROM transactions WHERE station_id = $1 AND evse_number = $2 AND status = 'Active' \
         ORDER BY COALESCE(started_at, created_at) DESC LIMIT 1",
    )
    .bind(station_id)
    .bind(evse_number.clamp(i32::MIN as i64, i32::MAX as i64) as i32)
    .fetch_optional(db)
    .await
}

/// Active transaction by OCPP transaction id: (db id, EVSE, start).
pub async fn active_by_uid(
    db: &Db,
    station_id: i64,
    uid: &str,
) -> sqlx::Result<Option<(i64, Option<i32>, Option<Ts>)>> {
    sqlx::query_as("SELECT id, evse_number, started_at FROM transactions WHERE station_id = $1 AND transaction_uid = $2 AND status = 'Active'")
        .bind(station_id)
        .bind(uid)
        .fetch_optional(db)
        .await
}