codetics-ocpp-service 0.1.0

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

//! OCPP 2.1 CSMS behaviour.

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

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

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_token_info(a: &db::tokens::AuthResult) -> IdTokenInfo {
    let status = match a.status {
        "Accepted" => AuthorizationStatusEnum::Accepted,
        "Blocked" => AuthorizationStatusEnum::Blocked,
        "Expired" => AuthorizationStatusEnum::Expired,
        "Unknown" => AuthorizationStatusEnum::Unknown,
        _ => AuthorizationStatusEnum::Invalid,
    };
    IdTokenInfo {
        custom_data: None,
        status,
        cache_expiry_date_time: a.expires_at.map(|t| t.0),
        charging_priority: None,
        language1: None,
        evse_id: None,
        group_id_token: a.group_id.as_ref().map(|g| IdToken {
            custom_data: None,
            additional_info: None,
            id_token: g.clone(),
            r#type: "Central".to_owned(),
        }),
        language2: None,
        personal_message: None,
    }
}

/// Value of a measurand in base units (Wh, W, A), from the last meter value
/// containing it: the phase-less value if present, otherwise the sum of L1–L3.
fn measurand(values: &[MeterValue], m: MeasurandEnum) -> Option<f64> {
    let scale = |s: &SampledValue| {
        let (unit, mult) = s
            .unit_of_measure
            .as_ref()
            .map(|u| (u.unit.as_deref().unwrap_or("Wh"), u.multiplier.unwrap_or(0)))
            .unwrap_or(("Wh", 0));
        let v = s.value * 10f64.powi(mult as i32);
        if unit.starts_with('k') { v * 1000.0 } else { v }
    };
    for mv in values.iter().rev() {
        // An absent measurand means Energy.Active.Import.Register.
        let matching: Vec<&SampledValue> = mv
            .sampled_value
            .iter()
            .filter(|s| {
                s.measurand
                    .unwrap_or(MeasurandEnum::EnergyActiveImportRegister)
                    == m
            })
            .collect();
        if let Some(total) = matching.iter().find(|s| s.phase.is_none()) {
            return Some(scale(total));
        }
        let phases: Vec<f64> = matching
            .iter()
            .filter(|s| matches!(s.phase, Some(PhaseEnum::L1 | PhaseEnum::L2 | PhaseEnum::L3)))
            .map(|s| scale(s))
            .collect();
        if !phases.is_empty() {
            return Some(phases.iter().sum());
        }
    }
    None
}

fn energy_wh(values: &[MeterValue]) -> Option<f64> {
    measurand(values, MeasurandEnum::EnergyActiveImportRegister)
}

fn message(text: String) -> MessageContent {
    let content: String = text.chars().take(512).collect();
    MessageContent {
        format: MessageFormatEnum::UTF8,
        language: None,
        content,
        custom_data: None,
    }
}

fn additional_info<'a>(token: Option<&'a IdToken>, kind: &str) -> Option<&'a str> {
    token?
        .additional_info
        .as_ref()?
        .iter()
        .find(|a| a.r#type == kind)
        .map(|a| a.additional_id_token.as_str())
}

/// Handles a CALL from a 2.1 Charging Station 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 cs = &r.charging_station;
            let info = BootInfo {
                vendor: &cs.vendor_name,
                model: &cs.model,
                serial_number: cs.serial_number.as_deref(),
                firmware_version: cs.firmware_version.as_deref(),
                iccid: cs.modem.as_ref().and_then(|m| m.iccid.as_deref()),
                imsi: cs.modem.as_ref().and_then(|m| m.imsi.as_deref()),
                meter_type: None,
                meter_serial_number: None,
                reason: Some(r.reason.as_str()),
            };
            let status = match ctx.boot(&info).await? {
                super::Registration::Pending => RegistrationStatusEnum::Pending,
                super::Registration::Rejected => RegistrationStatusEnum::Rejected,
                _ => RegistrationStatusEnum::Accepted,
            };
            json(BootNotificationResponse {
                custom_data: None,
                current_time: StationCtx::now(),
                interval: ctx.heartbeat_interval(),
                status,
                status_info: None,
            })
        }
        StationCall::Heartbeat(_) => {
            ctx.heartbeat().await?;
            json(HeartbeatResponse {
                custom_data: None,
                current_time: StationCtx::now(),
            })
        }
        StationCall::StatusNotification(r) => {
            ctx.status(&StatusUpdate {
                evse_number: i32_of(r.evse_id),
                connector_number: i32_of(r.connector_id),
                status: r.connector_status.as_str(),
                error_code: None,
                info: None,
                vendor_error_code: None,
                at: Ts(r.timestamp),
            })
            .await?;
            json(StatusNotificationResponse::default())
        }
        StationCall::Authorize(r) => {
            let ty = r.id_token.r#type.as_str();
            let a = ctx.authorize(&r.id_token.id_token, Some(ty)).await?;
            let mut info = id_token_info(&a);
            let mut tariff = None;
            if a.status == "Accepted" {
                // I01 (central cost): tariff text; I08 (local cost): driver-specific tariff.
                let extras = crate::tariff::on_authorize(
                    &ctx.app,
                    ctx.station_id,
                    OcppVersion::V21,
                    &r.id_token.id_token,
                    Some(ty),
                )
                .await
                .map_err(internal)?;
                info.personal_message = extras.personal_message.map(message);
                tariff = extras.tariff;
            }
            json(AuthorizeResponse {
                id_token_info: info,
                certificate_status: None,
                allowed_energy_transfer: None,
                tariff,
                custom_data: None,
            })
        }
        StationCall::TransactionEvent(r) => {
            let ti = &r.transaction_info;
            let event = TxEvent {
                version: "2.1",
                uid: &ti.transaction_id,
                event_type: r.event_type.as_str(),
                at: Some(Ts(r.timestamp)),
                seq_no: r.seq_no,
                evse_number: r.evse.as_ref().map(|e| i32_of(e.id)),
                connector_number: r.evse.as_ref().and_then(|e| e.connector_id).map(i32_of),
                id_token: r.id_token.as_ref().map(|t| t.id_token.as_str()),
                id_token_type: r.id_token.as_ref().map(|t| t.r#type.as_str()),
                charging_state: ti.charging_state.map(|s| s.as_str()),
                stop_reason: ti.stopped_reason.map(|s| s.as_str()),
                reservation_id: r.reservation_id.map(i32_of),
                remote_start_id: ti.remote_start_id.map(i32_of),
                energy_wh: r.meter_value.as_deref().and_then(energy_wh),
            };
            let tx_id = db::transactions::apply_event(&ctx.app.db, ctx.station_id, &event)
                .await
                .map_err(internal)?;
            if let Some(reservation) = r.reservation_id {
                // H01.FR.15: the reservation is used by this transaction.
                crate::reservation::on_used(&ctx.app, ctx.station_id, reservation, tx_id)
                    .await
                    .map_err(internal)?;
            }
            if r.event_type == TransactionEventEnum::Ended {
                // A TxProfile ends with its transaction.
                db::charging::clear_for_transaction(
                    &ctx.app.db,
                    ctx.station_id,
                    &ti.transaction_id,
                )
                .await
                .map_err(internal)?;
            }
            let evse = event.evse_number.unwrap_or(0);
            for mv in r.meter_value.iter().flatten() {
                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
                ctx.meter_values(Some(tx_id), evse, Ts(mv.timestamp), &sampled)
                    .await?;
            }
            tracing::debug!(station = %ctx.identity, transaction = %ti.transaction_id, event = r.event_type.as_str(), seq = r.seq_no, "TransactionEvent");
            let id_token_info = match &r.id_token {
                Some(t) => Some(id_token_info(
                    &ctx.authorize(&t.id_token, Some(t.r#type.as_str())).await?,
                )),
                None => None,
            };
            let values = r.meter_value.as_deref().unwrap_or_default();
            let cost_event = crate::tariff::TxEventCost {
                event_type: r.event_type.as_str(),
                at: r.timestamp,
                seq_no: r.seq_no,
                charging_state: ti.charging_state.map(|s| s.as_str()),
                energy_wh: energy_wh(values),
                power_w: measurand(values, MeasurandEnum::PowerActiveImport),
                current_a: measurand(values, MeasurandEnum::CurrentImport),
                tariff_id: ti.tariff_id.as_deref(),
                cost_details: r
                    .cost_details
                    .as_ref()
                    .map(serde_json::to_value)
                    .transpose()
                    .map_err(internal)?,
                payment_brand: additional_info(r.id_token.as_ref(), "PaymentBrand"),
                payment_recognition: additional_info(r.id_token.as_ref(), "PaymentRecognition"),
            };
            // I02/I03: running and final cost.
            let total_cost =
                crate::tariff::on_transaction_event(&ctx.app, ctx.station_id, tx_id, &cost_event)
                    .await
                    .map_err(internal)?;
            json(TransactionEventResponse {
                id_token_info,
                total_cost,
                ..Default::default()
            })
        }
        StationCall::MeterValues(r) => {
            for mv in &r.meter_value {
                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
                ctx.meter_values(None, i32_of(r.evse_id), Ts(mv.timestamp), &sampled)
                    .await?;
            }
            json(MeterValuesResponse::default())
        }
        StationCall::DataTransfer(r) => {
            ctx.log_event(action, payload).await;
            tracing::info!(station = %ctx.identity, vendor_id = %r.vendor_id, message_id = ?r.message_id, "DataTransfer");
            json(DataTransferResponse {
                custom_data: None,
                status: DataTransferStatusEnum::UnknownVendorId,
                status_info: None,
                data: None,
            })
        }
        StationCall::FirmwareStatusNotification(r) => {
            ctx.log_event(action, payload).await;
            if r.status == FirmwareStatusEnum::Installed {
                tracing::info!(station = %ctx.identity, "firmware installed");
            }
            json(FirmwareStatusNotificationResponse::default())
        }
        // No certificate authority is wired in: certificate requests are declined explicitly.
        StationCall::SignCertificate(_) => {
            ctx.log_event(action, payload).await;
            json(SignCertificateResponse {
                custom_data: None,
                status: GenericStatusEnum::Rejected,
                status_info: None,
            })
        }
        StationCall::Get15118EVCertificate(_) => {
            ctx.log_event(action, payload).await;
            json(Get15118EVCertificateResponse {
                custom_data: None,
                status: Iso15118EVCertificateStatusEnum::Failed,
                status_info: None,
                exi_response: String::new(),
                remaining_contracts: None,
            })
        }
        StationCall::GetCertificateStatus(_) => json(GetCertificateStatusResponse {
            custom_data: None,
            status: GetCertificateStatusEnum::Failed,
            status_info: None,
            ocsp_result: None,
        }),
        StationCall::NotifyEVChargingNeeds(r) => {
            ctx.log_event(action, payload).await;
            let n = &r.charging_needs;
            let needs = crate::smart::EvNeeds {
                evse: r.evse_id,
                bidirectional: {
                    let mode = n.requested_energy_transfer.as_str();
                    mode.contains("BPT") || mode.contains("DER")
                },
                ac_max_current: n
                    .ac_charging_parameters
                    .as_ref()
                    .map(|ac| ac.ev_max_current),
                dc_max_power: n
                    .dc_charging_parameters
                    .as_ref()
                    .map(|d| {
                        d.ev_max_power
                            .unwrap_or(d.ev_max_current * d.ev_max_voltage)
                    })
                    .or_else(|| {
                        n.v2x_charging_parameters
                            .as_ref()
                            .and_then(|x| x.max_charge_power)
                    }),
                departure: n.departure_time,
                max_schedule_tuples: r.max_schedule_tuples,
            };
            let status = crate::smart::on_ev_charging_needs(
                &ctx.app,
                ctx.station_id,
                OcppVersion::V21,
                needs,
            )
            .await
            .map_err(internal)?;
            let status = match status {
                "Accepted" => NotifyEVChargingNeedsStatusEnum::Accepted,
                _ => NotifyEVChargingNeedsStatusEnum::Rejected,
            };
            json(NotifyEVChargingNeedsResponse {
                custom_data: None,
                status,
                status_info: None,
            })
        }
        StationCall::NotifyEVChargingSchedule(_) => {
            ctx.log_event(action, payload).await;
            json(NotifyEVChargingScheduleResponse {
                custom_data: None,
                status: GenericStatusEnum::Accepted,
                status_info: None,
            })
        }
        // Notifications: stored in the event log, acknowledged with an empty response.
        StationCall::ClearedChargingLimit(r) => {
            db::charging::delete_external_limits(
                &ctx.app.db,
                ctx.station_id,
                r.charging_limit_source.as_str(),
                r.evse_id,
            )
            .await
            .map_err(internal)?;
            ack(
                ctx,
                action,
                payload,
                ClearedChargingLimitResponse::default(),
            )
            .await
        }
        StationCall::LogStatusNotification(_) => {
            ack(
                ctx,
                action,
                payload,
                LogStatusNotificationResponse::default(),
            )
            .await
        }
        StationCall::NotifyChargingLimit(r) => {
            let schedules = r
                .charging_schedule
                .as_ref()
                .map(serde_json::to_value)
                .transpose()
                .map_err(internal)?;
            db::charging::upsert_external_limit(
                &ctx.app.db,
                ctx.station_id,
                r.evse_id.unwrap_or(0),
                r.charging_limit.charging_limit_source.as_str(),
                r.charging_limit.is_grid_critical,
                r.charging_limit.is_local_generation,
                schedules.as_ref(),
            )
            .await
            .map_err(internal)?;
            ack(ctx, action, payload, NotifyChargingLimitResponse::default()).await
        }
        StationCall::NotifyCustomerInformation(_) => {
            ack(
                ctx,
                action,
                payload,
                NotifyCustomerInformationResponse::default(),
            )
            .await
        }
        StationCall::NotifyDisplayMessages(_) => {
            ack(
                ctx,
                action,
                payload,
                NotifyDisplayMessagesResponse::default(),
            )
            .await
        }
        StationCall::NotifyEvent(_) => {
            ack(ctx, action, payload, NotifyEventResponse::default()).await
        }
        StationCall::NotifyMonitoringReport(_) => {
            ack(
                ctx,
                action,
                payload,
                NotifyMonitoringReportResponse::default(),
            )
            .await
        }
        StationCall::NotifyReport(_) => {
            ack(ctx, action, payload, NotifyReportResponse::default()).await
        }
        StationCall::PublishFirmwareStatusNotification(_) => {
            ack(
                ctx,
                action,
                payload,
                PublishFirmwareStatusNotificationResponse::default(),
            )
            .await
        }
        StationCall::ReportChargingProfiles(r) => {
            let profiles: Vec<Value> = r
                .charging_profile
                .iter()
                .filter_map(|p| serde_json::to_value(p).ok())
                .collect();
            crate::smart::record_reported(
                &ctx.app,
                ctx.station_id,
                OcppVersion::V21,
                r.evse_id,
                r.charging_limit_source.as_str(),
                &profiles,
            )
            .await
            .map_err(internal)?;
            ack(
                ctx,
                action,
                payload,
                ReportChargingProfilesResponse::default(),
            )
            .await
        }
        StationCall::ReservationStatusUpdate(r) => {
            crate::reservation::on_status_update(
                &ctx.app,
                ctx.station_id,
                r.reservation_id,
                r.reservation_update_status.as_str(),
            )
            .await
            .map_err(internal)?;
            ack(
                ctx,
                action,
                payload,
                ReservationStatusUpdateResponse::default(),
            )
            .await
        }
        StationCall::SecurityEventNotification(_) => {
            ack(
                ctx,
                action,
                payload,
                SecurityEventNotificationResponse::default(),
            )
            .await
        }
        // OCPP 2.1 additions.
        StationCall::BatterySwap(_) => {
            ack(ctx, action, payload, BatterySwapResponse::default()).await
        }
        StationCall::NotifyDERAlarm(_) => {
            ack(ctx, action, payload, NotifyDERAlarmResponse::default()).await
        }
        StationCall::NotifyDERStartStop(_) => {
            ack(ctx, action, payload, NotifyDERStartStopResponse::default()).await
        }
        StationCall::NotifyPriorityCharging(_) => {
            ack(
                ctx,
                action,
                payload,
                NotifyPriorityChargingResponse::default(),
            )
            .await
        }
        StationCall::NotifySettlement(_) => {
            ack(ctx, action, payload, NotifySettlementResponse::default()).await
        }
        StationCall::ReportDERControl(_) => {
            ack(ctx, action, payload, ReportDERControlResponse::default()).await
        }
        StationCall::OpenPeriodicEventStream(_) => {
            ctx.log_event(action, payload).await;
            json(OpenPeriodicEventStreamResponse {
                status: GenericStatusEnum::Accepted,
                status_info: None,
                custom_data: None,
            })
        }
        StationCall::ClosePeriodicEventStream(_) => {
            ack(
                ctx,
                action,
                payload,
                ClosePeriodicEventStreamResponse::default(),
            )
            .await
        }
        StationCall::PullDynamicScheduleUpdate(r) => {
            // K28.FR.07 / K28.FR.12
            let update =
                crate::smart::pull_dynamic_update(&ctx.app, ctx.station_id, r.charging_profile_id)
                    .await
                    .map_err(internal)?;
            let update =
                update.and_then(|u| ocpp_core::decode_value::<ChargingScheduleUpdate>(u).ok());
            json(match update {
                Some(u) => PullDynamicScheduleUpdateResponse {
                    schedule_update: Some(u),
                    status: ChargingProfileStatusEnum::Accepted,
                    status_info: None,
                    custom_data: None,
                },
                None => PullDynamicScheduleUpdateResponse {
                    schedule_update: None,
                    status: ChargingProfileStatusEnum::Rejected,
                    status_info: Some(StatusInfo {
                        reason_code: "InvalidProfile".into(),
                        additional_info: None,
                        custom_data: None,
                    }),
                    custom_data: None,
                },
            })
        }
        StationCall::VatNumberValidation(r) => {
            ctx.log_event(action, payload).await;
            // No VAT registry is connected: report the number as not validated.
            json(VatNumberValidationResponse {
                company: None,
                status_info: None,
                vat_number: r.vat_number,
                evse_id: r.evse_id,
                status: GenericStatusEnum::Rejected,
                custom_data: None,
            })
        }
        // No certificate authority / OCSP responder is connected.
        StationCall::GetCertificateChainStatus(_) => Err(RpcError::not_supported(action)),
    }
}

/// Handles an unconfirmed SEND message (OCPP 2.1 message type 6). No reply is sent.
pub async fn handle_send(ctx: &mut StationCtx, action: &str, payload: &str) {
    match StationSend::decode(action, payload) {
        Ok(StationSend::NotifyPeriodicEventStream(s)) => {
            // Values are offsets from `basetime`; stored as received.
            let at = Ts(s.basetime);
            let data =
                serde_json::json!({ "streamId": s.id, "pending": s.pending, "data": s.data });
            if let Err(e) = ctx.meter_values(None, 0, at, &data).await {
                tracing::warn!(station = %ctx.identity, error = %e, "failed to store periodic event stream data");
            }
        }
        Err(e) => {
            tracing::warn!(station = %ctx.identity, action, error = %e, "invalid SEND message ignored")
        }
    }
}

async fn ack<T: serde::Serialize>(
    ctx: &StationCtx,
    action: &str,
    payload: &str,
    response: T,
) -> Result<Value, RpcError> {
    ctx.log_event(action, payload).await;
    json(response)
}

/// 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}"),
        )
    })?;
    ocpp21::validate_response(a, payload)
}