codetics-ocpp-service 0.1.0

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

//! OCPP 1.6 Central System behaviour.

use ocpp_core::{ErrorCode, RpcError};
use ocpp16::*;
use serde_json::Value;

use super::{StationCtx, i32_of, internal};
use crate::db::{self, Ts, stations::BootInfo, stations::StatusUpdate};

pub fn decode_error(e: DecodeError) -> RpcError {
    match e {
        DecodeError::UnknownAction(a) => RpcError::not_implemented(&a),
        DecodeError::WrongDirection(a) => RpcError::not_supported(a.as_str()),
        DecodeError::Invalid(v) => v.into(),
    }
}

fn json<T: serde::Serialize>(v: T) -> Result<Value, RpcError> {
    serde_json::to_value(v).map_err(internal)
}

fn id_tag_info(a: &db::tokens::AuthResult) -> IdTagInfo {
    let status = match a.status {
        "Accepted" => AuthorizationStatus::Accepted,
        "Blocked" => AuthorizationStatus::Blocked,
        "Expired" => AuthorizationStatus::Expired,
        _ => AuthorizationStatus::Invalid,
    };
    IdTagInfo {
        status,
        expiry_date: a.expires_at.map(|t| t.0),
        parent_id_tag: a.group_id.clone(),
    }
}

/// Handles a CALL from a 1.6 Charge Point and returns the CALLRESULT payload.
pub async fn handle(ctx: &mut StationCtx, action: &str, payload: &str) -> Result<Value, RpcError> {
    let call = StationCall::decode(action, payload).map_err(decode_error)?;
    match call {
        StationCall::BootNotification(r) => {
            let info = BootInfo {
                vendor: &r.charge_point_vendor,
                model: &r.charge_point_model,
                serial_number: r
                    .charge_point_serial_number
                    .as_deref()
                    .or(r.charge_box_serial_number.as_deref()),
                firmware_version: r.firmware_version.as_deref(),
                iccid: r.iccid.as_deref(),
                imsi: r.imsi.as_deref(),
                meter_type: r.meter_type.as_deref(),
                meter_serial_number: r.meter_serial_number.as_deref(),
                reason: None,
            };
            let status = match ctx.boot(&info).await? {
                super::Registration::Pending => RegistrationStatus::Pending,
                super::Registration::Rejected => RegistrationStatus::Rejected,
                _ => RegistrationStatus::Accepted,
            };
            json(BootNotificationResponse {
                status,
                current_time: StationCtx::now(),
                interval: ctx.heartbeat_interval(),
            })
        }
        StationCall::Heartbeat(_) => {
            ctx.heartbeat().await?;
            json(HeartbeatResponse {
                current_time: StationCtx::now(),
            })
        }
        StationCall::StatusNotification(r) => {
            let at = Ts(r.timestamp.unwrap_or_else(StationCtx::now));
            let error =
                (r.error_code != ChargePointErrorCode::NoError).then(|| r.error_code.as_str());
            // 1.6 has one connector per connectorId: connectorId N ↦ EVSE N, connector 1.
            ctx.status(&StatusUpdate {
                evse_number: i32_of(r.connector_id),
                connector_number: 1,
                status: r.status.as_str(),
                error_code: error,
                info: r.info.as_deref(),
                vendor_error_code: r.vendor_error_code.as_deref(),
                at,
            })
            .await?;
            // §5.13: a reservation ends when its connector becomes Faulted or Unavailable.
            if matches!(
                r.status,
                ChargePointStatus::Faulted | ChargePointStatus::Unavailable
            ) {
                db::reservations::remove_for_connector(&ctx.app.db, ctx.station_id, r.connector_id)
                    .await
                    .map_err(internal)?;
            }
            json(StatusNotificationResponse {})
        }
        StationCall::Authorize(r) => {
            let a = ctx.authorize(&r.id_tag, None).await?;
            json(AuthorizeResponse {
                id_tag_info: id_tag_info(&a),
            })
        }
        StationCall::StartTransaction(r) => {
            let a = ctx.authorize(&r.id_tag, None).await?;
            // A transactionId is always assigned, even when the idTag is not accepted:
            // the Charge Point then decides based on StopTransactionOnInvalidId (1.6 §4.8).
            let id = db::transactions::start_v16(
                &ctx.app.db,
                ctx.station_id,
                i32_of(r.connector_id),
                &r.id_tag,
                r.meter_start,
                Ts(r.timestamp),
                r.reservation_id.map(i32_of),
            )
            .await
            .map_err(internal)?;
            tracing::info!(station = %ctx.identity, transaction_id = id, connector = r.connector_id, status = a.status, "StartTransaction");
            if let Some(reservation) = r.reservation_id {
                crate::reservation::on_used(&ctx.app, ctx.station_id, reservation, id)
                    .await
                    .map_err(internal)?;
            }
            json(StartTransactionResponse {
                id_tag_info: id_tag_info(&a),
                transaction_id: id,
            })
        }
        StationCall::StopTransaction(r) => {
            let reason = r.reason.map(|x| x.as_str()).unwrap_or("Local");
            let known = db::transactions::stop_v16(
                &ctx.app.db,
                ctx.station_id,
                r.transaction_id,
                r.meter_stop,
                Ts(r.timestamp),
                reason,
            )
            .await
            .map_err(internal)?;
            // §3.13.1: the TxProfile of a transaction is deleted when it stops.
            db::charging::clear_for_transaction(
                &ctx.app.db,
                ctx.station_id,
                &r.transaction_id.to_string(),
            )
            .await
            .map_err(internal)?;
            if !known {
                tracing::warn!(station = %ctx.identity, transaction_id = r.transaction_id, "StopTransaction for unknown transaction");
                ctx.log_event("StopTransaction", payload).await;
            }
            for mv in r.transaction_data.iter().flatten() {
                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
                ctx.meter_values(
                    known.then_some(r.transaction_id),
                    0,
                    Ts(mv.timestamp),
                    &sampled,
                )
                .await?;
            }
            let id_tag_info = match &r.id_tag {
                Some(tag) => Some(id_tag_info(&ctx.authorize(tag, None).await?)),
                None => None,
            };
            json(StopTransactionResponse { id_tag_info })
        }
        StationCall::MeterValues(r) => {
            let tx = match r.transaction_id {
                Some(id)
                    if db::transactions::exists_v16(&ctx.app.db, ctx.station_id, id)
                        .await
                        .map_err(internal)? =>
                {
                    Some(id)
                }
                _ => None,
            };
            for mv in &r.meter_value {
                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
                ctx.meter_values(tx, i32_of(r.connector_id), Ts(mv.timestamp), &sampled)
                    .await?;
            }
            json(MeterValuesResponse {})
        }
        StationCall::DataTransfer(r) => {
            ctx.log_event("DataTransfer", payload).await;
            tracing::info!(station = %ctx.identity, vendor_id = %r.vendor_id, message_id = ?r.message_id, "DataTransfer");
            json(DataTransferResponse {
                status: DataTransferStatus::UnknownVendorId,
                data: None,
            })
        }
        StationCall::DiagnosticsStatusNotification(_) => {
            ctx.log_event(action, payload).await;
            json(DiagnosticsStatusNotificationResponse {})
        }
        StationCall::FirmwareStatusNotification(_) => {
            ctx.log_event(action, payload).await;
            json(FirmwareStatusNotificationResponse {})
        }
    }
}

/// Validates a command payload sent through the API before it is queued.
pub fn validate_command(action: &str, payload: Value) -> Result<Value, RpcError> {
    CsmsCall::from_value(action, payload)
        .map(|c| c.to_payload())
        .map_err(|e| match e {
            DecodeError::Invalid(v) => RpcError::new(v.kind.error_code(), v.message),
            other => RpcError::new(ErrorCode::NotSupported, other.to_string()),
        })
}

/// Validates a CALLRESULT received for one of our CALLs.
pub fn validate_response(action: &str, payload: &str) -> Result<Value, ocpp_core::Violation> {
    let a = Action::from_name(action).ok_or_else(|| {
        ocpp_core::Violation::new(
            ocpp_core::ViolationKind::Format,
            format!("unknown action {action}"),
        )
    })?;
    ocpp16::validate_response(a, payload)
}