codetics-ocpp-service 0.1.0

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

//! Reservations (OCPP 1.6 §3.11, §5.1, §5.13; OCPP 2.x H01–H04).
//!
//! The CSMS assigns reservation ids, sends ReserveNow / CancelReservation and
//! follows the life cycle from the station's messages: StartTransaction /
//! TransactionEvent with reservationId (used), ReservationStatusUpdate
//! (2.x: expired / removed), Faulted / Unavailable connectors (1.6), and
//! expiry.

use ocpp_core::{DateTime, OcppVersion};
use serde::Deserialize;
use serde_json::{Value, json};

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

#[derive(Debug, Deserialize)]
pub struct ReserveRequest {
    /// EVSE (2.x) / connector (1.6) to reserve; absent = any (1.6 connector 0,
    /// requires ReserveConnectorZeroSupported; 2.x requires ReservationNonEvseSpecific).
    pub evse_id: Option<i64>,
    /// 2.x only: reserve any EVSE with this connector type (e.g. `cCCS2`).
    pub connector_type: Option<String>,
    pub id_token: String,
    /// 2.x IdTokenType, default `Central`.
    pub id_token_type: Option<String>,
    /// 1.6 parentIdTag / 2.x groupIdToken.
    pub group_id_token: Option<String>,
    pub expires_at: DateTime,
}

pub async fn reserve(
    app: &App,
    station: &Station,
    version: OcppVersion,
    req: &ReserveRequest,
) -> Result<(Reservation, Value), AppError> {
    if req.expires_at <= DateTime::now() {
        return Err(AppError::BadRequest(
            "expires_at must be in the future".into(),
        ));
    }
    if req.evse_id.is_some_and(|e| e < 0) {
        return Err(AppError::BadRequest("evse_id must be positive".into()));
    }
    if version == OcppVersion::V16 && req.connector_type.is_some() {
        return Err(AppError::BadRequest(
            "OCPP 1.6 cannot reserve by connector type".into(),
        ));
    }
    let evse = req.evse_id.filter(|e| *e > 0).map(|e| e as i32);
    let token_type = (version != OcppVersion::V16).then(|| {
        req.id_token_type
            .clone()
            .unwrap_or_else(|| "Central".into())
    });
    let reservation = db::reservations::insert(
        &app.db,
        station.id,
        &db::reservations::NewReservation {
            ocpp_version: version.as_str(),
            evse_number: evse,
            connector_type: req.connector_type.as_deref(),
            id_token: &req.id_token,
            id_token_type: token_type.as_deref(),
            group_id_token: req.group_id_token.as_deref(),
            expires_at: Ts(req.expires_at),
        },
    )
    .await?;
    let payload = match version {
        OcppVersion::V16 => json!({
            "connectorId": evse.unwrap_or(0),
            "expiryDate": req.expires_at,
            "idTag": req.id_token,
            "parentIdTag": req.group_id_token,
            "reservationId": reservation.id,
        }),
        _ => json!({
            "id": reservation.id,
            "expiryDateTime": req.expires_at,
            "connectorType": req.connector_type,
            "idToken": { "idToken": req.id_token, "type": token_type },
            "evseId": evse,
            "groupIdToken": req.group_id_token.as_ref().map(|g| json!({ "idToken": g, "type": "Central" })),
        }),
    };
    let payload = crate::smart::strip_nulls(payload);
    let response =
        match calls::call(app, &station.ocpp_identity, version, "ReserveNow", payload).await {
            Ok(r) => r,
            Err(e) => {
                db::reservations::set_status(
                    &app.db,
                    station.id,
                    reservation.id,
                    "Failed",
                    &["Requested"],
                )
                .await?;
                return Err(e);
            }
        };
    let status = match calls::status_of(&response) {
        s @ ("Accepted" | "Faulted" | "Occupied" | "Rejected" | "Unavailable") => s,
        _ => "Rejected",
    };
    db::reservations::set_status(&app.db, station.id, reservation.id, status, &["Requested"])
        .await?;
    tracing::info!(station = %station.ocpp_identity, reservation = reservation.id, %status, "ReserveNow");
    let reservation = db::reservations::get(&app.db, station.id, reservation.id)
        .await?
        .expect("just inserted");
    Ok((reservation, response))
}

pub async fn cancel(
    app: &App,
    station: &Station,
    version: OcppVersion,
    id: i32,
) -> Result<(Reservation, Value), AppError> {
    let r = db::reservations::get(&app.db, station.id, id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("reservation {id} not found")))?;
    if r.status != "Accepted" {
        return Err(AppError::Conflict(format!(
            "reservation {id} is not active (status {})",
            r.status
        )));
    }
    let response = calls::call(
        app,
        &station.ocpp_identity,
        version,
        "CancelReservation",
        json!({ "reservationId": id }),
    )
    .await?;
    // Rejected: the station has no such reservation any more (expired, used or removed).
    let status = if calls::status_of(&response) == "Accepted" {
        "Cancelled"
    } else {
        "Removed"
    };
    db::reservations::set_status(&app.db, station.id, id, status, &["Accepted"]).await?;
    let r = db::reservations::get(&app.db, station.id, id)
        .await?
        .expect("exists");
    Ok((r, response))
}

/// The station started a transaction for a reservation.
pub async fn on_used(
    app: &App,
    station_id: i64,
    reservation_id: i64,
    transaction_id: i64,
) -> Result<(), AppError> {
    let Ok(id) = i32::try_from(reservation_id) else {
        return Ok(());
    };
    if !db::reservations::mark_used(&app.db, station_id, id, transaction_id).await? {
        tracing::warn!(
            station_id,
            reservation = id,
            "transaction refers to an unknown or ended reservation"
        );
    }
    Ok(())
}

/// OCPP 2.x ReservationStatusUpdate (H04.FR.01, H01.FR.16/17).
pub async fn on_status_update(
    app: &App,
    station_id: i64,
    reservation_id: i64,
    status: &str,
) -> Result<(), AppError> {
    let Ok(id) = i32::try_from(reservation_id) else {
        return Ok(());
    };
    let status = if status == "Expired" {
        "Expired"
    } else {
        "Removed"
    };
    db::reservations::set_status(&app.db, station_id, id, status, &["Accepted", "Requested"])
        .await?;
    Ok(())
}

/// Periodically closes reservations past their expiry time.
pub fn spawn_sweeper(app: App, mut shutdown: tokio::sync::watch::Receiver<bool>) {
    tokio::spawn(async move {
        let mut tick = tokio::time::interval(app.cfg.reservation_sweep_interval);
        loop {
            tokio::select! {
                _ = shutdown.changed() => return,
                _ = tick.tick() => match db::reservations::expire_due(&app.db).await {
                    Ok(0) => {}
                    Ok(n) => tracing::info!(count = n, "reservations expired"),
                    Err(e) => tracing::warn!(error = %e, "reservation sweep failed"),
                },
            }
        }
    });
}