codetics-ocpp-service 0.1.0

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

//! Charging stations, EVSEs and connectors.

use evse_id::EvseId;
use serde::{Deserialize, Serialize};

use super::{Db, Ts};

#[derive(Debug, Clone, sqlx::FromRow, Serialize)]
pub struct Station {
    pub id: i64,
    pub ocpp_identity: String,
    pub evse_id: String,
    pub name: String,
    pub latitude: Option<f64>,
    pub longitude: Option<f64>,
    pub address_line1: Option<String>,
    pub address_line2: Option<String>,
    pub postal_code: Option<String>,
    pub city: Option<String>,
    pub region: Option<String>,
    pub country_code: Option<String>,
    #[serde(skip)]
    pub auth_password_hash: Option<String>,
    pub allowed_versions: Option<Vec<String>>,
    pub boot_policy: String,
    pub registration_status: String,
    pub vendor: Option<String>,
    pub model: Option<String>,
    pub serial_number: Option<String>,
    pub firmware_version: Option<String>,
    pub iccid: Option<String>,
    pub imsi: Option<String>,
    pub meter_type: Option<String>,
    pub meter_serial_number: Option<String>,
    pub last_boot_at: Option<Ts>,
    pub last_boot_reason: Option<String>,
    pub status: Option<String>,
    pub error_code: Option<String>,
    pub status_at: Option<Ts>,
    pub ocpp_version: Option<String>,
    pub connected: bool,
    pub connected_node: Option<String>,
    pub connected_at: Option<Ts>,
    pub disconnected_at: Option<Ts>,
    pub last_heartbeat_at: Option<Ts>,
    pub last_seen_at: Option<Ts>,
    pub security_profile: Option<i16>,
    pub timezone: Option<String>,
    pub cost_mode: String,
    pub created_at: Ts,
    pub updated_at: Ts,
}

impl Station {
    pub fn has_password(&self) -> bool {
        self.auth_password_hash.is_some()
    }
}

const COLS: &str = "id, ocpp_identity, evse_id, name, latitude, longitude, address_line1, address_line2, postal_code, city, region, country_code, \
    auth_password_hash, allowed_versions, boot_policy, registration_status, vendor, model, serial_number, firmware_version, iccid, imsi, \
    meter_type, meter_serial_number, last_boot_at, last_boot_reason, status, error_code, status_at, ocpp_version, connected, connected_node, \
    connected_at, disconnected_at, last_heartbeat_at, last_seen_at, security_profile, timezone, cost_mode, created_at, updated_at";

#[derive(Debug, Clone, Default, Deserialize)]
pub struct StationInput {
    pub ocpp_identity: Option<String>,
    pub name: Option<String>,
    /// Body of the station e-mobility ID (without type and check characters).
    /// Generated from the database id when absent.
    pub evse_id_body: Option<String>,
    pub latitude: Option<f64>,
    pub longitude: Option<f64>,
    pub address_line1: Option<String>,
    pub address_line2: Option<String>,
    pub postal_code: Option<String>,
    pub city: Option<String>,
    pub region: Option<String>,
    pub country_code: Option<String>,
    pub allowed_versions: Option<Vec<String>>,
    pub boot_policy: Option<String>,
    /// OCPP Security Profile 0–3 enforced when the station connects.
    pub security_profile: Option<i16>,
    /// IANA time zone, e.g. `Europe/Paris`.
    pub timezone: Option<String>,
    /// `None`, `Central` or `Local` (OCPP 2.x tariff and cost).
    pub cost_mode: Option<String>,
}

pub async fn get_by_identity(db: &Db, identity: &str) -> sqlx::Result<Option<Station>> {
    sqlx::query_as(&format!(
        "SELECT {COLS} FROM charging_stations WHERE ocpp_identity = $1"
    ))
    .bind(identity)
    .fetch_optional(db)
    .await
}

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

/// Creates a station. The `S` e-mobility ID is derived from the body given
/// or, by default, from the generated database id.
pub async fn create(
    db: &Db,
    input: &StationInput,
    identity: &str,
    name: &str,
) -> Result<Station, crate::error::AppError> {
    let mut tx = db.begin().await?;
    let id: i64 = sqlx::query_scalar(
        "INSERT INTO charging_stations (ocpp_identity, evse_id, name, latitude, longitude, address_line1, address_line2, postal_code, city, region, \
         country_code, allowed_versions, boot_policy, security_profile, timezone, cost_mode) \
         VALUES ($1, 'tmp-' || gen_random_uuid(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, COALESCE($12, 'Accepted'), $13, $14, COALESCE($15, 'None')) RETURNING id",
    )
    .bind(identity)
    .bind(name)
    .bind(input.latitude)
    .bind(input.longitude)
    .bind(&input.address_line1)
    .bind(&input.address_line2)
    .bind(&input.postal_code)
    .bind(&input.city)
    .bind(&input.region)
    .bind(input.country_code.as_ref().map(|c| c.to_ascii_uppercase()))
    .bind(&input.allowed_versions)
    .bind(&input.boot_policy)
    .bind(input.security_profile)
    .bind(&input.timezone)
    .bind(&input.cost_mode)
    .fetch_one(&mut *tx)
    .await?;
    let evse_id = match &input.evse_id_body {
        Some(body) => EvseId::generate(evse_id::IdType::Station, body)
            .map_err(|e| crate::error::AppError::BadRequest(format!("evse_id_body: {e}")))?,
        None => EvseId::station(id as u64),
    };
    sqlx::query("UPDATE charging_stations SET evse_id = $2 WHERE id = $1")
        .bind(id)
        .bind(evse_id.as_str())
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(get(db, id).await?.expect("just inserted"))
}

pub async fn update(db: &Db, id: i64, input: &StationInput) -> sqlx::Result<Option<Station>> {
    sqlx::query_as(&format!(
        "UPDATE charging_stations SET \
           name = COALESCE($2, name), latitude = COALESCE($3, latitude), longitude = COALESCE($4, longitude), \
           address_line1 = COALESCE($5, address_line1), address_line2 = COALESCE($6, address_line2), postal_code = COALESCE($7, postal_code), \
           city = COALESCE($8, city), region = COALESCE($9, region), country_code = COALESCE($10, country_code), \
           allowed_versions = COALESCE($11, allowed_versions), boot_policy = COALESCE($12, boot_policy), \
           security_profile = COALESCE($13, security_profile), timezone = COALESCE($14, timezone), cost_mode = COALESCE($15, cost_mode), updated_at = now() \
         WHERE id = $1 RETURNING {COLS}"
    ))
    .bind(id)
    .bind(&input.name)
    .bind(input.latitude)
    .bind(input.longitude)
    .bind(&input.address_line1)
    .bind(&input.address_line2)
    .bind(&input.postal_code)
    .bind(&input.city)
    .bind(&input.region)
    .bind(input.country_code.as_ref().map(|c| c.to_ascii_uppercase()))
    .bind(&input.allowed_versions)
    .bind(&input.boot_policy)
    .bind(input.security_profile)
    .bind(&input.timezone)
    .bind(&input.cost_mode)
    .fetch_optional(db)
    .await
}

pub async fn delete(db: &Db, id: i64) -> sqlx::Result<bool> {
    Ok(sqlx::query("DELETE FROM charging_stations WHERE id = $1")
        .bind(id)
        .execute(db)
        .await?
        .rows_affected()
        > 0)
}

pub async fn set_password_hash(db: &Db, id: i64, hash: Option<&str>) -> sqlx::Result<bool> {
    Ok(sqlx::query(
        "UPDATE charging_stations SET auth_password_hash = $2, updated_at = now() WHERE id = $1",
    )
    .bind(id)
    .bind(hash)
    .execute(db)
    .await?
    .rows_affected()
        > 0)
}

#[derive(Debug, Clone, Default, Deserialize)]
pub struct SearchParams {
    /// Free text over name and postal address (accent- and case-insensitive).
    pub q: Option<String>,
    pub lat: Option<f64>,
    pub lon: Option<f64>,
    /// Radius around (lat, lon), default 10 km.
    pub radius_km: Option<f64>,
    pub city: Option<String>,
    pub postal_code: Option<String>,
    pub country_code: Option<String>,
    pub connected: Option<bool>,
    pub limit: Option<i64>,
    pub offset: Option<i64>,
}

#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct SearchHit {
    #[sqlx(flatten)]
    #[serde(flatten)]
    pub station: Station,
    /// Distance in km when searching around a position.
    pub distance_km: Option<f64>,
}

pub async fn search(db: &Db, p: &SearchParams) -> sqlx::Result<Vec<SearchHit>> {
    let radius = p.radius_km.unwrap_or(10.0).clamp(0.01, 20_000.0);
    // Bounding box (uses the (latitude, longitude) index), then exact haversine distance.
    let (lat_delta, lon_delta) = match p.lat {
        Some(lat) => (
            radius / 111.32,
            radius / (111.32 * lat.to_radians().cos().abs().max(0.01)),
        ),
        None => (0.0, 0.0),
    };
    sqlx::query_as(&format!(
        "SELECT * FROM ( \
           SELECT {COLS}, \
             CASE WHEN $3::float8 IS NULL OR latitude IS NULL THEN NULL ELSE \
               6371.0 * 2 * asin(sqrt(power(sin(radians(latitude - $3) / 2), 2) + \
               cos(radians($3)) * cos(radians(latitude)) * power(sin(radians(longitude - $4) / 2), 2))) END AS distance_km, \
             CASE WHEN $1::text IS NULL THEN 0 ELSE similarity(address_search, lower(ocpp_immutable_unaccent($1))) END AS score \
           FROM charging_stations \
           WHERE ($1::text IS NULL OR address_search ILIKE '%' || lower(ocpp_immutable_unaccent($1)) || '%' \
                  OR address_search % lower(ocpp_immutable_unaccent($1))) \
             AND ($3::float8 IS NULL OR (latitude BETWEEN $3 - $5 AND $3 + $5 AND longitude BETWEEN $4 - $6 AND $4 + $6)) \
             AND ($7::text IS NULL OR lower(ocpp_immutable_unaccent(city)) = lower(ocpp_immutable_unaccent($7))) \
             AND ($8::text IS NULL OR postal_code = $8) \
             AND ($9::text IS NULL OR country_code = upper($9)) \
             AND ($10::bool IS NULL OR connected = $10) \
         ) s \
         WHERE $3::float8 IS NULL OR distance_km <= $2 \
         ORDER BY distance_km ASC NULLS LAST, score DESC, name ASC \
         LIMIT $11 OFFSET $12"
    ))
    .bind(p.q.as_deref().map(str::trim).filter(|s| !s.is_empty()))
    .bind(radius)
    .bind(p.lat.filter(|_| p.lon.is_some()))
    .bind(p.lon)
    .bind(lat_delta)
    .bind(lon_delta)
    .bind(&p.city)
    .bind(&p.postal_code)
    .bind(&p.country_code)
    .bind(p.connected)
    .bind(p.limit.unwrap_or(50).clamp(1, 500))
    .bind(p.offset.unwrap_or(0).max(0))
    .fetch_all(db)
    .await
}

// --- Connection state --------------------------------------------------------

pub async fn mark_connected(
    db: &Db,
    id: i64,
    node: &str,
    connection_id: &str,
    version: &str,
) -> sqlx::Result<()> {
    sqlx::query(
        "UPDATE charging_stations SET connected = true, connected_node = $2, connection_id = $3, ocpp_version = $4, \
         connected_at = now(), last_seen_at = now(), updated_at = now() WHERE id = $1",
    )
    .bind(id)
    .bind(node)
    .bind(connection_id)
    .bind(version)
    .execute(db)
    .await?;
    Ok(())
}

/// Marks the station offline unless a newer connection took over meanwhile.
pub async fn mark_disconnected(db: &Db, id: i64, connection_id: &str) -> sqlx::Result<()> {
    sqlx::query(
        "UPDATE charging_stations SET connected = false, connected_node = NULL, connection_id = NULL, disconnected_at = now(), updated_at = now() \
         WHERE id = $1 AND connection_id = $2",
    )
    .bind(id)
    .bind(connection_id)
    .execute(db)
    .await?;
    Ok(())
}

/// On startup: connections recorded for this node do not exist any more.
pub async fn reset_node(db: &Db, node: &str) -> sqlx::Result<u64> {
    Ok(sqlx::query(
        "UPDATE charging_stations SET connected = false, connected_node = NULL, connection_id = NULL, disconnected_at = now() \
         WHERE connected AND connected_node = $1",
    )
    .bind(node)
    .execute(db)
    .await?
    .rows_affected())
}

pub async fn touch(db: &Db, id: i64, heartbeat: bool) -> sqlx::Result<()> {
    let sql = if heartbeat {
        "UPDATE charging_stations SET last_heartbeat_at = now(), last_seen_at = now() WHERE id = $1"
    } else {
        "UPDATE charging_stations SET last_seen_at = now() WHERE id = $1"
    };
    sqlx::query(sql).bind(id).execute(db).await?;
    Ok(())
}

#[derive(Debug, Default)]
pub struct BootInfo<'a> {
    pub vendor: &'a str,
    pub model: &'a str,
    pub serial_number: Option<&'a str>,
    pub firmware_version: Option<&'a str>,
    pub iccid: Option<&'a str>,
    pub imsi: Option<&'a str>,
    pub meter_type: Option<&'a str>,
    pub meter_serial_number: Option<&'a str>,
    pub reason: Option<&'a str>,
}

pub async fn record_boot(
    db: &Db,
    id: i64,
    b: &BootInfo<'_>,
    registration_status: &str,
) -> sqlx::Result<()> {
    sqlx::query(
        "UPDATE charging_stations SET vendor = $2, model = $3, serial_number = $4, firmware_version = $5, iccid = $6, imsi = $7, \
         meter_type = $8, meter_serial_number = $9, last_boot_reason = $10, registration_status = $11, last_boot_at = now(), \
         last_seen_at = now(), updated_at = now() WHERE id = $1",
    )
    .bind(id)
    .bind(b.vendor)
    .bind(b.model)
    .bind(b.serial_number)
    .bind(b.firmware_version)
    .bind(b.iccid)
    .bind(b.imsi)
    .bind(b.meter_type)
    .bind(b.meter_serial_number)
    .bind(b.reason)
    .bind(registration_status)
    .execute(db)
    .await?;
    Ok(())
}

pub async fn set_firmware_version(db: &Db, id: i64, firmware: &str) -> sqlx::Result<()> {
    sqlx::query(
        "UPDATE charging_stations SET firmware_version = $2, updated_at = now() WHERE id = $1",
    )
    .bind(id)
    .bind(firmware)
    .execute(db)
    .await?;
    Ok(())
}

// --- EVSEs and connectors ----------------------------------------------------

#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct Evse {
    pub evse_number: i32,
    pub evse_id: String,
    #[sqlx(json)]
    pub connectors: Vec<Connector>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Connector {
    pub connector_number: i32,
    pub status: String,
    pub error_code: Option<String>,
    pub info: Option<String>,
    pub vendor_error_code: Option<String>,
    pub status_at: String,
}

/// Creates the EVSE (with its `E` e-mobility ID) if it does not exist yet.
pub async fn ensure_evse(
    db: &Db,
    station_id: i64,
    station_evse_id: &str,
    evse_number: i32,
) -> Result<String, crate::error::AppError> {
    if let Some(id) = sqlx::query_scalar::<_, String>(
        "SELECT evse_id FROM evses WHERE station_id = $1 AND evse_number = $2",
    )
    .bind(station_id)
    .bind(evse_number)
    .fetch_optional(db)
    .await?
    {
        return Ok(id);
    }
    let station = EvseId::parse(station_evse_id).map_err(|e| {
        crate::error::AppError::Internal(format!("stored station evse_id invalid: {e}"))
    })?;
    let evse = EvseId::evse_of(&station, evse_number as u32)
        .map_err(|e| crate::error::AppError::BadRequest(e.to_string()))?;
    sqlx::query("INSERT INTO evses (station_id, evse_number, evse_id) VALUES ($1, $2, $3) ON CONFLICT (station_id, evse_number) DO NOTHING")
        .bind(station_id)
        .bind(evse_number)
        .bind(evse.as_str())
        .execute(db)
        .await?;
    Ok(
        sqlx::query_scalar("SELECT evse_id FROM evses WHERE station_id = $1 AND evse_number = $2")
            .bind(station_id)
            .bind(evse_number)
            .fetch_one(db)
            .await?,
    )
}

pub struct StatusUpdate<'a> {
    pub evse_number: i32,
    pub connector_number: i32,
    pub status: &'a str,
    pub error_code: Option<&'a str>,
    pub info: Option<&'a str>,
    pub vendor_error_code: Option<&'a str>,
    pub at: Ts,
}

pub async fn upsert_connector_status(
    db: &Db,
    station_id: i64,
    u: &StatusUpdate<'_>,
) -> sqlx::Result<()> {
    // Out-of-order (older) notifications do not overwrite newer state.
    sqlx::query(
        "INSERT INTO connectors (station_id, evse_number, connector_number, status, error_code, info, vendor_error_code, status_at) \
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8) \
         ON CONFLICT (station_id, evse_number, connector_number) DO UPDATE SET status = EXCLUDED.status, error_code = EXCLUDED.error_code, \
           info = EXCLUDED.info, vendor_error_code = EXCLUDED.vendor_error_code, status_at = EXCLUDED.status_at, updated_at = now() \
         WHERE connectors.status_at <= EXCLUDED.status_at",
    )
    .bind(station_id)
    .bind(u.evse_number)
    .bind(u.connector_number)
    .bind(u.status)
    .bind(u.error_code)
    .bind(u.info)
    .bind(u.vendor_error_code)
    .bind(u.at)
    .execute(db)
    .await?;
    Ok(())
}

pub async fn set_station_status(
    db: &Db,
    station_id: i64,
    status: &str,
    error_code: Option<&str>,
    at: Ts,
) -> sqlx::Result<()> {
    sqlx::query(
        "UPDATE charging_stations SET status = $2, error_code = $3, status_at = $4, updated_at = now() \
         WHERE id = $1 AND (status_at IS NULL OR status_at <= $4)",
    )
    .bind(station_id)
    .bind(status)
    .bind(error_code)
    .bind(at)
    .execute(db)
    .await?;
    Ok(())
}

pub async fn list_evses(db: &Db, station_id: i64) -> sqlx::Result<Vec<Evse>> {
    sqlx::query_as(
        "SELECT e.evse_number, e.evse_id, COALESCE(json_agg(json_build_object( \
             'connector_number', c.connector_number, 'status', c.status, 'error_code', c.error_code, 'info', c.info, \
             'vendor_error_code', c.vendor_error_code, 'status_at', to_char(c.status_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"')) \
             ORDER BY c.connector_number) FILTER (WHERE c.connector_number IS NOT NULL), '[]') AS connectors \
         FROM evses e LEFT JOIN connectors c ON c.station_id = e.station_id AND c.evse_number = e.evse_number \
         WHERE e.station_id = $1 GROUP BY e.evse_number, e.evse_id ORDER BY e.evse_number",
    )
    .bind(station_id)
    .fetch_all(db)
    .await
}