codetics-ocpp-service 0.1.0

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

//! Charging profiles and external charging limits.

use serde::Serialize;
use serde_json::Value;

use super::{Db, Ts};

#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct ProfileRow {
    pub id: i64,
    pub ocpp_version: String,
    pub profile_id: i32,
    pub evse_number: i32,
    pub purpose: String,
    pub stack_level: i32,
    pub kind: String,
    pub transaction_uid: Option<String>,
    pub valid_from: Option<Ts>,
    pub valid_to: Option<Ts>,
    pub source: String,
    pub status: String,
    pub status_info: Option<Value>,
    pub profile: Value,
    pub dyn_update: Option<Value>,
    pub dyn_updated_at: Option<Ts>,
    pub created_at: Ts,
    pub updated_at: Ts,
}

const COLS: &str = "id, ocpp_version, profile_id, evse_number, purpose, stack_level, kind, transaction_uid, valid_from, valid_to, source, status, \
    status_info, profile, dyn_update, dyn_updated_at, created_at, updated_at";

/// Normalized columns of a profile row.
pub struct NewProfile<'a> {
    pub ocpp_version: &'a str,
    pub profile_id: i64,
    pub evse_number: i64,
    pub purpose: &'a str,
    pub stack_level: i64,
    pub kind: &'a str,
    pub transaction_uid: Option<&'a str>,
    pub valid_from: Option<Ts>,
    pub valid_to: Option<Ts>,
    pub source: &'a str,
    pub profile: &'a Value,
}

fn i32_of(v: i64) -> i32 {
    v.clamp(i32::MIN as i64, i32::MAX as i64) as i32
}

/// Records a profile that is being sent (status Pending). A previous Pending
/// row with the same id (an unanswered attempt) is discarded.
pub async fn insert_pending(db: &Db, station_id: i64, p: &NewProfile<'_>) -> sqlx::Result<i64> {
    let mut tx = db.begin().await?;
    sqlx::query("DELETE FROM charging_profiles WHERE station_id = $1 AND profile_id = $2 AND status = 'Pending'")
        .bind(station_id)
        .bind(i32_of(p.profile_id))
        .execute(&mut *tx)
        .await?;
    let id = insert(&mut tx, station_id, p, "Pending").await?;
    tx.commit().await?;
    Ok(id)
}

async fn insert(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    station_id: i64,
    p: &NewProfile<'_>,
    status: &str,
) -> sqlx::Result<i64> {
    sqlx::query_scalar(
        "INSERT INTO charging_profiles (station_id, ocpp_version, profile_id, evse_number, purpose, stack_level, kind, transaction_uid, \
           valid_from, valid_to, source, status, profile) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING id",
    )
    .bind(station_id)
    .bind(p.ocpp_version)
    .bind(i32_of(p.profile_id))
    .bind(i32_of(p.evse_number))
    .bind(p.purpose)
    .bind(i32_of(p.stack_level))
    .bind(p.kind)
    .bind(p.transaction_uid)
    .bind(p.valid_from)
    .bind(p.valid_to)
    .bind(p.source)
    .bind(status)
    .bind(sqlx::types::Json(p.profile))
    .fetch_one(&mut **tx)
    .await
}

/// The station accepted the profile: it replaces any installed profile with
/// the same id and, when `replace_same_slot`, the one with the same
/// EVSE + purpose + stack level (and transaction for TxProfiles), as the
/// station itself does (1.6 §3.13.2, §5.16.3).
pub async fn mark_installed(
    db: &Db,
    station_id: i64,
    row_id: i64,
    replace_same_slot: bool,
) -> sqlx::Result<()> {
    let mut tx = db.begin().await?;
    sqlx::query(
        "UPDATE charging_profiles o SET status = 'Cleared', updated_at = now() FROM charging_profiles n \
         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 \
           ($3 AND o.evse_number = n.evse_number AND o.purpose = n.purpose AND o.stack_level = n.stack_level \
            AND o.transaction_uid IS NOT DISTINCT FROM n.transaction_uid))",
    )
    .bind(station_id)
    .bind(row_id)
    .bind(replace_same_slot)
    .execute(&mut *tx)
    .await?;
    sqlx::query(
        "UPDATE charging_profiles SET status = 'Installed', updated_at = now() WHERE id = $1",
    )
    .bind(row_id)
    .execute(&mut *tx)
    .await?;
    tx.commit().await
}

pub async fn mark_rejected(db: &Db, row_id: i64, status_info: Option<&Value>) -> sqlx::Result<()> {
    sqlx::query("UPDATE charging_profiles SET status = 'Rejected', status_info = $2, updated_at = now() WHERE id = $1")
        .bind(row_id)
        .bind(status_info.map(sqlx::types::Json))
        .execute(db)
        .await?;
    Ok(())
}

/// Removes the Pending row of a profile that could not be delivered.
pub async fn discard_pending(db: &Db, row_id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM charging_profiles WHERE id = $1 AND status = 'Pending'")
        .bind(row_id)
        .execute(db)
        .await?;
    Ok(())
}

pub async fn list(db: &Db, station_id: i64, status: Option<&str>) -> sqlx::Result<Vec<ProfileRow>> {
    sqlx::query_as(&format!(
        "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"
    ))
    .bind(station_id)
    .bind(status)
    .fetch_all(db)
    .await
}

pub async fn installed(db: &Db, station_id: i64) -> sqlx::Result<Vec<ProfileRow>> {
    list(db, station_id, Some("Installed")).await
}

pub async fn find_installed(
    db: &Db,
    station_id: i64,
    profile_id: i64,
) -> sqlx::Result<Option<ProfileRow>> {
    sqlx::query_as(&format!("SELECT {COLS} FROM charging_profiles WHERE station_id = $1 AND profile_id = $2 AND status = 'Installed'"))
        .bind(station_id)
        .bind(i32_of(profile_id))
        .fetch_optional(db)
        .await
}

/// Criteria of a ClearChargingProfile request (all optional, combined with AND).
#[derive(Debug, Default)]
pub struct ClearCriteria<'a> {
    pub profile_id: Option<i64>,
    pub evse_number: Option<i64>,
    pub purpose: Option<&'a str>,
    pub stack_level: Option<i64>,
}

pub async fn clear(db: &Db, station_id: i64, c: &ClearCriteria<'_>) -> sqlx::Result<u64> {
    Ok(sqlx::query(
        "UPDATE charging_profiles SET status = 'Cleared', updated_at = now() WHERE station_id = $1 AND status = 'Installed' AND source = 'CSO' \
         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) \
         AND ($5::int IS NULL OR stack_level = $5)",
    )
    .bind(station_id)
    .bind(c.profile_id.map(i32_of))
    .bind(c.evse_number.map(i32_of))
    .bind(c.purpose)
    .bind(c.stack_level.map(i32_of))
    .execute(db)
    .await?
    .rows_affected())
}

/// A TxProfile ends with its transaction (1.6 §3.13.1, 2.x K §3.2).
pub async fn clear_for_transaction(
    db: &Db,
    station_id: i64,
    transaction_uid: &str,
) -> sqlx::Result<u64> {
    Ok(sqlx::query(
        "UPDATE charging_profiles SET status = 'Cleared', updated_at = now() \
         WHERE station_id = $1 AND status = 'Installed' AND purpose IN ('TxProfile', 'PriorityCharging') AND transaction_uid = $2",
    )
    .bind(station_id)
    .bind(transaction_uid)
    .execute(db)
    .await?
    .rows_affected())
}

/// Stores a profile reported by the station (ReportChargingProfiles) as installed.
pub async fn upsert_reported(db: &Db, station_id: i64, p: &NewProfile<'_>) -> sqlx::Result<()> {
    let mut tx = db.begin().await?;
    sqlx::query("UPDATE charging_profiles SET status = 'Cleared', updated_at = now() WHERE station_id = $1 AND profile_id = $2 AND status IN ('Installed', 'Pending')")
        .bind(station_id)
        .bind(i32_of(p.profile_id))
        .execute(&mut *tx)
        .await?;
    insert(&mut tx, station_id, p, "Installed").await?;
    tx.commit().await
}

pub async fn set_dyn_update(db: &Db, row_id: i64, update: &Value) -> sqlx::Result<()> {
    sqlx::query("UPDATE charging_profiles SET dyn_update = $2, dyn_updated_at = now(), updated_at = now() WHERE id = $1")
        .bind(row_id)
        .bind(sqlx::types::Json(update))
        .execute(db)
        .await?;
    Ok(())
}

/// Next free profile id for CSMS-generated profiles. Starts high to stay clear of ids
/// that stations generate for external constraints (2.1 K01.FR.80).
pub async fn next_profile_id(db: &Db, station_id: i64) -> sqlx::Result<i64> {
    sqlx::query_scalar("SELECT GREATEST(COALESCE(MAX(profile_id), 0) + 1, 100000)::bigint FROM charging_profiles WHERE station_id = $1")
        .bind(station_id)
        .fetch_one(db)
        .await
}

#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct ExternalLimit {
    pub evse_number: i32,
    pub source: String,
    pub is_grid_critical: Option<bool>,
    pub is_local_generation: Option<bool>,
    pub schedules: Option<Value>,
    pub received_at: Ts,
}

pub async fn upsert_external_limit(
    db: &Db,
    station_id: i64,
    evse_number: i64,
    source: &str,
    grid_critical: Option<bool>,
    local_generation: Option<bool>,
    schedules: Option<&Value>,
) -> sqlx::Result<()> {
    sqlx::query(
        "INSERT INTO external_charging_limits (station_id, evse_number, source, is_grid_critical, is_local_generation, schedules, received_at) \
         VALUES ($1, $2, $3, $4, $5, $6, now()) ON CONFLICT (station_id, evse_number, source) DO UPDATE SET \
         is_grid_critical = EXCLUDED.is_grid_critical, is_local_generation = EXCLUDED.is_local_generation, schedules = EXCLUDED.schedules, received_at = now()",
    )
    .bind(station_id)
    .bind(i32_of(evse_number))
    .bind(source)
    .bind(grid_critical)
    .bind(local_generation)
    .bind(schedules.map(sqlx::types::Json))
    .execute(db)
    .await?;
    Ok(())
}

/// ClearedChargingLimit: without an EVSE, the limits of that source are released everywhere.
pub async fn delete_external_limits(
    db: &Db,
    station_id: i64,
    source: &str,
    evse_number: Option<i64>,
) -> sqlx::Result<u64> {
    Ok(sqlx::query("DELETE FROM external_charging_limits WHERE station_id = $1 AND source = $2 AND ($3::int IS NULL OR evse_number = $3)")
        .bind(station_id)
        .bind(source)
        .bind(evse_number.map(i32_of))
        .execute(db)
        .await?
        .rows_affected())
}

pub async fn external_limits(db: &Db, station_id: i64) -> sqlx::Result<Vec<ExternalLimit>> {
    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")
        .bind(station_id)
        .fetch_all(db)
        .await
}