mako-wim 0.13.0

//! WiM Messstellenbetrieb — MSB change workflow (PIDs 55039, 55042, 55051, 55168).
//!
//! Covers the process by which an incoming metering point operator
//! (neuer Messstellenbetreiber, nMSB) initiates a change of the MSB at a
//! delivery point (Messlokation, MeLo) by sending a UTILMD message to the
//! grid operator (Netzbetreiber, NB). The NB validates the message and
//! dispatches an APERAK within **5 Werktage** (business days).
//!
//! # Regulatory basis
//!
//! - **MsbG** — Messstellenbetriebsgesetz (governing smart meter rollout)
//! - **BDEW WiM** — Wechselprozesse im Messwesen Strom
//! - **BNetzA BK6-18-032** — ruling governing WiM timeline obligations
//! - **UTILMD S2.x** — EDI@Energy message format for metering processes
//! - **APERAK 2.x** — Application error acknowledgement (**5 Werktage** Frist)
//!
//! # Frist comparison
//!
//! | Process family | APERAK Frist | Calculation |
//! |---|---|---|
//! | GPKE Lieferbeginn | 24 h wall-clock | `fristen::add_hours(24)` |
//! | WiM Gerätewechsel | **5 Werktage** | `fristen::add_werktage(5, BdewMaKo)` |
//! | GeLi Gas Anmeldung | **10 Werktage** | `fristen::add_werktage(10, BdewMaKo)` |

use std::collections::HashMap;

use mako_engine::types::Pruefidentifikator;
use mako_engine::{
    envelope::EventEnvelope,
    error::WorkflowError,
    fristen::{
        APERAK_STROM_WINDOW_LABEL, HolidayCalendar, aperak_strom_due_at, deadline_at_werktage,
    },
    ids::DeadlineId,
    outbox::PendingOutbox,
    projection::Projection,
    types::{DeviceId, MarktpartnerCode, MeLo, MessageRef},
    workflow::{CommandPayload, EventPayload, PendingDeadline, Workflow, WorkflowOutput},
};
use time::OffsetDateTime;

/// Stable workflow name used as the `WorkflowId.name` and in the `ProcessRegistry`.
pub const WORKFLOW_NAME: &str = "wim-device-change";

/// Deadline label for the 5-Werktage APERAK response window (WiM BK6-18-032).
///
/// Register a `Deadline` with this label immediately after `ValidationPassed`:
///
/// ```rust,ignore
/// let due = mako_engine::fristen::deadline_at_werktage(
///     received_at, 5, HolidayCalendar::BdewMaKo,
/// );
/// let deadline = Deadline::new(process.stream_id().clone(), ..., APERAK_WINDOW_LABEL, due);
/// deadline_store.register(&deadline).await?;
/// ```
pub const APERAK_WINDOW_LABEL: &str = "wim-aperak-5-werktage";

/// Prüfidentifikatoren that carry a WiM MSB-Wechsel UTILMD.
///
/// Directions are per *Anwendungsübersicht der Prüfidentifikatoren* 4.0 and the
/// BK6-24-174 WiM Teil 1 Lesefassung. Note that they are **not** uniformly
/// "MSB → NB" — 55039 never reaches the NB at all, and 55168 addresses the gMSB:
///
/// | PID   | Process                            | Von  | An   | Kap.  |
/// |-------|------------------------------------|------|------|-------|
/// | 55039 | Kündigung MSB                      | MSBN | MSBA | 2.2   |
/// | 55042 | Anmeldung MSB                      | MSBN | NB   | 2.3   |
/// | 55051 | Ende MSB (Abmeldung)               | MSBA | NB   | 2.4   |
/// | 55168 | Verpflichtungsanfrage / Aufforderung | NB | gMSB | 2.4   |
///
/// The Kündigung (55039) runs on the **contract layer** between the two MSB and
/// is explicitly *non-constitutive*: BK6-24-174 Kap. 2.1.3 states that a switch
/// is effected solely by the successful Anmeldung MSBN → NB. Never gate 55042 on
/// a 55040 Bestätigung — they are independent channels.
///
/// Used both to validate inbound UTILMD and to constrain the outbound
/// [`DeviceChangeCommand::InitiateDeviceChange`] order.
pub const DEVICE_CHANGE_PIDS: &[u32] = &[55_039, 55_042, 55_051, 55_168];

/// Antwortfrist in Werktagen for the counterparty's business response.
///
/// **These differ per process** — a single flat window would fire early for the
/// Kündigung and late for the Abmeldung. From BK6-24-174 WiM Teil 1
/// ("Unverzüglich, jedoch spätester ÜT ist der *n*. WT nach dem ÜT von Nr. 1"):
///
/// | Request | Antwort | Frist | Fundstelle |
/// |---------|---------|-------|------------|
/// | 55039   | 55040/55041 | **3 WT** | Kap. 2.2.2 Nr. 2 |
/// | 55042   | 55043/55044 | **5 WT** | Kap. 2.3.2 Nr. 2 |
/// | 55051   | 55052/55053 | **7 WT** | Kap. 2.4.2 Nr. 2 |
/// | 55168   | 55169/55170 | **1 WT** | Kap. 2.4.2 Nr. 4 |
///
/// Distinct from the APERAK window, which is **45 minutes** for UTILMD in Strom
/// (APERAK AHB §2.4.1) — see [`APERAK_WINDOW_LABEL`].
///
/// Returns `None` when `request_pid` is not a WiM MSB-Wechsel request.
#[must_use]
pub const fn antwort_frist_werktage(request_pid: u32) -> Option<u32> {
    match request_pid {
        55_039 => Some(3),
        55_042 => Some(5),
        55_051 => Some(7),
        55_168 => Some(1),
        _ => None,
    }
}

/// Deadline label for the 5-Werktage counterparty response window on an
/// **outbound** MSB-Wechsel order (WiM BK6-24-174).
///
/// Registered by the caller alongside [`DeviceChangeCommand::InitiateDeviceChange`].
/// Distinct from [`APERAK_WINDOW_LABEL`], which tracks *our* obligation to
/// acknowledge an inbound message; this one tracks *their* obligation to answer ours.
pub const AUFTRAG_ANTWORT_WINDOW_LABEL: &str = "wim-device-change-antwort-5-werktage";

/// Response Prüfidentifikatoren for the WiM MSB-Wechsel, as
/// `(antwort_pid, request_pid, is_confirmed)`.
///
/// | Request | Bestätigung | Ablehnung |
/// |---------|-------------|-----------|
/// | 55039   | 55040       | 55041     |
/// | 55042   | 55043       | 55044     |
/// | 55051   | 55052       | 55053     |
/// | 55168   | 55169       | 55170     |
///
/// These close an order opened with [`DeviceChangeCommand::InitiateDeviceChange`].
///
/// The UTILMD AHB Strom **does** define all twelve as full Anwendungsfälle —
/// Kap. 10.1 (Kündigung), 10.2 (Anmeldung), 10.3 (Verpflichtungsanfrage),
/// 10.4 (Beendigung), each as one table with a column per Prüfidentifikator.
/// The response PIDs additionally carry `SG4 STS+E01` (Status der Antwort) and
/// resolve their Ablehnungsgründe through EBD codes (55040→`E_0200`,
/// 55043/55044→`E_0201`, 55052/55053→`E_0202`, 55169/55170→`E_0240`).
///
/// **However**, `crates/edi-energy/src/generated/` currently emits rule packs for
/// the four *request* PIDs only, so `validate()` on an inbound response yields
/// `ProfileNotFound`. The adapter therefore treats a missing profile as
/// "not validated" rather than "invalid", and the Bestätigung/Ablehnung decision
/// rides on the PID. Regenerating the profiles from AHB Kap. 10 would let these
/// be schema-validated too — tracked as a codegen follow-up, not a spec limit.
pub const DEVICE_CHANGE_ANTWORT_PIDS: &[(u32, u32, bool)] = &[
    (55_040, 55_039, true),
    (55_041, 55_039, false),
    (55_043, 55_042, true),
    (55_044, 55_042, false),
    (55_052, 55_051, true),
    (55_053, 55_051, false),
    (55_169, 55_168, true),
    (55_170, 55_168, false),
];

/// Resolve a response PID to `(request_pid, is_confirmed)`.
///
/// Returns `None` when `pid` is not a WiM MSB-Wechsel response.
#[must_use]
pub fn antwort_pid_meaning(pid: u32) -> Option<(u32, bool)> {
    DEVICE_CHANGE_ANTWORT_PIDS
        .iter()
        .find(|(antwort, _, _)| *antwort == pid)
        .map(|(_, request, confirmed)| (*request, *confirmed))
}

/// WiM Strom IFTSTA Prüfidentifikatoren (PIDs 21007, 21009–21015, 21018, 21029–21032).
///
/// These status messages are part of the WiM MSB-Wechsel (WiM Strom Teil 1)
/// process. All are routed to `"wim-device-change"` for correlation.
///
/// Per IFTSTA AHB these PIDs are "WiM / Statusmeldung MSB-Wechsel nach MsbG".
///
/// | PID   | Beschreibung | Richtung |
/// |-------|---|---|
/// | 21007 | Statusmeldung NB→LF / NB→MSBA | WiM Strom Teil 1 / WiM Gas |
/// | 21009 | Statusmeldung MSB-Wechsel nach MsbG an LF | NB → LF |
/// | 21010 | Statusmeldung MSB-Wechsel nach MsbG an NB | MSB alt → NB |
/// | 21011 | Statusmeldung MSB-Wechsel nach MsbG an NB | MSB neu → NB |
/// | 21012 | Statusmeldung MSB-Wechsel nach MsbG an BKV | NB → BKV |
/// | 21013 | Statusmeldung MSB-Wechsel nach MsbG an ÜNB | NB → ÜNB |
/// | 21015 | Statusmeldung Einbau iMS | wMSB → gMSB |
/// | 21018 | Statusmeldung Anforderung Datenzugang | MSB → LF |
/// | 21029 | Vorabinformation | wMSB → NB |
/// | 21030 | iMS-Ersteinbauzustand | wMSB → gMSB |
/// | 21031 | Bestandssituation / Eigenausbau iMS | wMSB → gMSB |
/// | 21032 | Antwort auf das Angebot | LF → MSB |
pub const IFTSTA_PIDS: &[u32] = &[
    21_007, 21_009, 21_010, 21_011, 21_012, 21_013, 21_015, 21_018, 21_029, 21_030, 21_031, 21_032,
];

// ── Domain events ─────────────────────────────────────────────────────────────

/// Events emitted by the WiM Gerätewechsel workflow.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", content = "data")]
pub enum DeviceChangeEvent {
    /// Process initiated by a valid UTILMD Anmeldung Messstellenbetrieb.
    Initiated {
        /// Messlokation EIC code.
        melo_id: MeLo,
        /// GLN of the incoming Messstellenbetreiber.
        incoming_msb: MarktpartnerCode,
        /// GLN of the grid operator (Netzbetreiber).
        grid_operator: MarktpartnerCode,
        /// Physical device identifier.
        device_id: DeviceId,
        /// EDIFACT document date (YYYYMMDD).
        document_date: String,
        /// EDIFACT message reference (UNH/BGM).
        message_ref: MessageRef,
        /// BDEW Prüfidentifikator.
        pruefidentifikator: Pruefidentifikator,
    },
    /// EDIFACT message passed profile validation (no rule violations).
    ValidationPassed {
        /// Reference of the validated message.
        message_ref: MessageRef,
    },
    /// A positive or negative APERAK was dispatched within 5 Werktage.
    AperakDispatched {
        /// `true` for positive (accepted), `false` for negative (rejected).
        positive: bool,
        /// Rejection reason (only set when `positive = false`).
        reason: Option<String>,
    },
    /// Meter device physically changed; new MSB is active.
    Completed {
        /// Physical device identifier confirmed at completion.
        device_id: DeviceId,
    },
    /// Process was rejected and closed.
    Rejected {
        /// Human-readable rejection reason.
        reason: String,
    },
    /// A registered deadline expired before the process completed.
    DeadlineExpired {
        /// Unique ID of the expired deadline.
        deadline_id: DeadlineId,
        /// Label identifying the deadline type.
        label: Box<str>,
    },
    /// Received an IFTSTA WiM status message (PIDs 21009–21018).
    ///
    /// WiM IFTSTA messages are informational: they notify the parties of
    /// process-status updates and Vollzugsmeldungen without driving a state
    /// transition. Recorded in the event log for audit purposes.
    IftstaStatusReceived {
        /// IFTSTA Prüfidentifikator (21009–21018).
        pid: Pruefidentifikator,
        /// Sender party code (GLN).
        sender: MarktpartnerCode,
        /// Receiver party code (GLN).
        receiver: MarktpartnerCode,
        /// EDIFACT message reference.
        message_ref: MessageRef,
    },
    /// An **outbound** MSB-Wechsel order was dispatched by this party.
    ///
    /// Distinct from [`Self::Initiated`], which records an inbound UTILMD we
    /// *received*. Conflating the two would make the event log claim we were
    /// sent a message we in fact sent — the audit trail must record direction.
    AuftragGesendet {
        /// Messlokation the order applies to.
        melo_id: MeLo,
        /// GLN of this party (the order sender).
        sender: MarktpartnerCode,
        /// GLN of the counterparty (NB or nMSB, depending on PID).
        receiver: MarktpartnerCode,
        /// Requested execution date (YYYYMMDD, German local time).
        process_date: String,
        /// EDIFACT message reference of the outbound UTILMD.
        message_ref: MessageRef,
        /// Prüfidentifikator (55039, 55042, 55051, or 55168).
        pruefidentifikator: Pruefidentifikator,
    },
    /// The counterparty answered our outbound order (Bestätigung or Ablehnung).
    ///
    /// Closes the loop opened by [`Self::AuftragGesendet`] and absorbs the
    /// 5-Werktage response deadline.
    AntwortEmpfangen {
        /// Response Prüfidentifikator — see [`DEVICE_CHANGE_ANTWORT_PIDS`].
        pruefidentifikator: Pruefidentifikator,
        /// GLN of the answering counterparty.
        sender: MarktpartnerCode,
        /// EDIFACT message reference of the inbound response.
        message_ref: MessageRef,
        /// `true` for a Bestätigung, `false` for an Ablehnung.
        is_confirmed: bool,
        /// Rejection reason, when the counterparty supplied one.
        reason: Option<String>,
    },
}

impl EventPayload for DeviceChangeEvent {
    fn event_type(&self) -> &'static str {
        match self {
            Self::AuftragGesendet { .. } => "WimDeviceChangeAuftragGesendet",
            Self::AntwortEmpfangen { .. } => "WimDeviceChangeAntwortEmpfangen",
            Self::Initiated { .. } => "WimDeviceChangeInitiated",
            Self::ValidationPassed { .. } => "WimDeviceChangeValidationPassed",
            Self::AperakDispatched { .. } => "WimDeviceChangeAperakDispatched",
            Self::Completed { .. } => "WimDeviceChangeCompleted",
            Self::Rejected { .. } => "WimDeviceChangeRejected",
            Self::DeadlineExpired { .. } => "WimDeviceChangeDeadlineExpired",
            Self::IftstaStatusReceived { .. } => "WimDeviceChangeIftstaStatusReceived",
        }
    }
    // schema_version defaults to 1; increment and add an upcast arm on next
    // backward-incompatible payload layout change.
}

// ── Domain state ──────────────────────────────────────────────────────────────

/// Business data set at `Initiated` time and carried through every later state.
///
/// All fields are structurally guaranteed to be present once the process moves
/// past `New` — no `unwrap()` required downstream.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DeviceChangeData {
    /// EIC/MeLo code for the metering location.
    pub melo_id: MeLo,
    /// Market partner code (GLN) of the incoming MSB.
    pub incoming_msb: MarktpartnerCode,
    /// Market partner code (GLN) of the grid operator.
    pub grid_operator: MarktpartnerCode,
    /// Device identifier.
    pub device_id: DeviceId,
    /// EDIFACT document date string from the UTILMD.
    pub document_date: String,
    /// BDEW Prüfidentifikator.
    pub pruefidentifikator: Pruefidentifikator,
    /// Original UTILMD message reference, preserved for APERAK construction.
    /// `None` only for processes initiated before this field was added (old snapshots).
    #[serde(default)]
    pub message_ref: Option<MessageRef>,
}

/// Current state of a WiM Gerätewechsel process stream.
///
/// Modelled as an enum-per-variant to eliminate all `Option`-unwraps:
/// each variant carries exactly the data that is structurally available at
/// that stage. Invalid states are unrepresentable.
///
/// # Lifecycle
///
/// ```text
/// New → Initiated → ValidationPassed → AperakSent → Completed
///                                    ↘ Rejected
///     ↘ Rejected (failed validation at Initiated step)
/// ```
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "status", content = "data")]
#[derive(Default)]
pub enum DeviceChangeState {
    /// No events yet; stream exists but process has not started.
    #[default]
    New,
    /// Outbound MSB-Wechsel order dispatched; awaiting the counterparty's answer.
    AuftragGesendet(DeviceChangeData),
    /// Counterparty confirmed our outbound order; awaiting the physical device swap.
    AuftragBestaetigt(DeviceChangeData),
    /// UTILMD received and `Initiated` event applied.
    Initiated(DeviceChangeData),
    /// EDIFACT validation passed; APERAK not yet dispatched.
    ValidationPassed(DeviceChangeData),
    /// Positive APERAK dispatched; awaiting physical device swap.
    AperakSent(DeviceChangeData),
    /// Device physically changed; new MSB is active.
    Completed(DeviceChangeData),
    /// Process rejected (validation failure or negative APERAK).
    Rejected {
        /// Human-readable rejection reason.
        reason: String,
    },
}

impl DeviceChangeState {
    /// Stable string label for the current variant.
    #[must_use]
    pub fn status_str(&self) -> &'static str {
        match self {
            Self::New => "New",
            Self::AuftragGesendet(_) => "AuftragGesendet",
            Self::AuftragBestaetigt(_) => "AuftragBestaetigt",
            Self::Initiated(_) => "Initiated",
            Self::ValidationPassed(_) => "ValidationPassed",
            Self::AperakSent(_) => "AperakSent",
            Self::Completed(_) => "Completed",
            Self::Rejected { .. } => "Rejected",
        }
    }
}

// ── Domain commands ───────────────────────────────────────────────────────────

/// Commands for the WiM Gerätewechsel workflow.
///
/// **All domain values must be pre-extracted by the transport layer** before
/// constructing a command. `Workflow::handle()` is pure — no I/O, no EDIFACT
/// parsing, no external calls. See the crate-level doc for a construction
/// example.
#[derive(Clone)]
pub enum DeviceChangeCommand {
    /// ERP instructs this party to **send** a WiM MSB-Wechsel order.
    ///
    /// Emits [`DeviceChangeEvent::AuftragGesendet`] plus a `UTILMD` outbox entry.
    /// The caller registers the counterparty-response deadline
    /// ([`AUFTRAG_ANTWORT_WINDOW_LABEL`]) alongside it, sized per process via
    /// [`antwort_frist_werktage`] — 3 / 5 / 7 / 1 WT, **not** one flat window.
    ///
    /// Direction is per-PID and not a uniform "MSB → NB" split — see
    /// [`DEVICE_CHANGE_PIDS`] for the table. In particular 55039 never reaches
    /// the NB (MSBN → MSBA) and 55168 addresses the gMSB.
    ///
    /// The command itself is role-agnostic; `makod`'s command API enforces which
    /// Marktrolle may issue which PID.
    InitiateDeviceChange {
        /// Prüfidentifikator; must be one of [`DEVICE_CHANGE_PIDS`].
        pid: Pruefidentifikator,
        /// GLN of this party (order sender).
        sender: MarktpartnerCode,
        /// GLN of the counterparty (order receiver).
        receiver: MarktpartnerCode,
        /// Messlokation the order applies to.
        melo_id: MeLo,
        /// Requested execution date (YYYYMMDD, German local time).
        process_date: String,
        /// EDIFACT message reference of the outbound UTILMD.
        message_ref: MessageRef,
    },
    /// Inbound Bestätigung / Ablehnung answering our outbound order.
    ///
    /// Emits [`DeviceChangeEvent::AntwortEmpfangen`] and closes the
    /// [`AUFTRAG_ANTWORT_WINDOW_LABEL`] deadline by leaving `AuftragGesendet`.
    ReceiveAntwort {
        /// Response Prüfidentifikator — see [`DEVICE_CHANGE_ANTWORT_PIDS`].
        pid: Pruefidentifikator,
        /// GLN of the answering counterparty.
        sender: MarktpartnerCode,
        /// EDIFACT message reference of the inbound response.
        message_ref: MessageRef,
        /// Rejection reason, when the counterparty supplied one.
        reason: Option<String>,
    },
    /// Inbound UTILMD accepted from the AS4 layer. Domain fields extracted and
    /// validation performed by the caller before constructing this command.
    ReceiveUtilmd {
        /// BDEW Prüfidentifikator.
        pid: Pruefidentifikator,
        /// GLN of the message sender (nMSB).
        sender: MarktpartnerCode,
        /// GLN of the message receiver (NB).
        receiver: MarktpartnerCode,
        /// Messlokation EIC code.
        melo_id: MeLo,
        /// Physical device identifier.
        device_id: DeviceId,
        /// EDIFACT document date (YYYYMMDD).
        document_date: String,
        /// EDIFACT message reference.
        message_ref: MessageRef,
        /// `true` if `msg.validate()` returned a report with no errors.
        validation_passed: bool,
        /// Human-readable validation issue strings for the `Rejected` event.
        validation_errors: Vec<String>,
        /// UTC wall-clock time when the inbound UTILMD was received.
        ///
        /// Used to compute the APERAK 45-minute sending deadline
        /// (APERAK AHB 1.0 §2.4.1) and the 5-Werktage process deadline
        /// (WiM BK6-24-174 §2a) that are registered atomically with
        /// the `Initiated` event.
        received_at: OffsetDateTime,
    },
    /// Inbound iMS Universalbestellprozess order received via REST
    /// (BDEW API-Webdienste Strom, valid 2026-01-29+, PIDs 11021–11023).
    ///
    /// Used when the Netzbetreiber orders an iMS installation from the MSB
    /// through the REST channel rather than via EDIFACT/AS4. The caller is
    /// responsible for validating the request before constructing this command.
    ReceiveRestOrder {
        /// REST transaction UUID (idempotency key; carried through to events).
        tx_id: String,
        /// 13-digit GLN of the Netzbetreiber (order sender).
        sender_mp_id: MarktpartnerCode,
        /// EIC of the Messlokation at which the device should be installed.
        melo_id: MeLo,
        /// Requested device category (e.g. `"iMSys"`, `"mME"`, `"mME+KME"`).
        device_category: String,
        /// Requested installation / process date (ISO 8601 date string).
        process_date: String,
    },
    /// Dispatch a positive or negative APERAK.
    ///
    /// **BDEW WiM / BNetzA BK6-18-032**: APERAK must be sent within
    /// **5 Werktage** of receiving the UTILMD (not wall-clock hours).
    /// Use `fristen::add_werktage(5, HolidayCalendar::BdewMaKo)` to compute
    /// the deadline.
    DispatchAperak {
        /// `true` for positive APERAK, `false` for negative.
        positive: bool,
        /// Rejection reason (required when `positive = false`).
        reason: Option<String>,
    },
    /// Mark the device change as completed once the physical swap is confirmed.
    Complete {
        /// Physical device identifier confirmed at completion.
        device_id: DeviceId,
    },
    /// A registered deadline fired and was dispatched by the scheduler.
    ///
    /// Transitions the process to `Rejected` unless it has already reached
    /// a terminal state (`Completed` or `Rejected`), in which case this is a no-op.
    TimeoutExpired {
        /// Unique ID of the expired deadline.
        deadline_id: DeadlineId,
        /// Label identifying the deadline type.
        label: Box<str>,
    },
    /// Received an IFTSTA WiM status message (PIDs 21009–21018).
    ///
    /// Constructed by the IFTSTA adapter in `makod` when an inbound AS4
    /// IFTSTA message with a WiM PID arrives, or via the
    /// `"wim.iftsta.empfangen"` REST command.
    ReceiveIftsta {
        /// IFTSTA Prüfidentifikator (21009–21018).
        pid: Pruefidentifikator,
        /// Sender party code (GLN).
        sender: MarktpartnerCode,
        /// Receiver party code (GLN).
        receiver: MarktpartnerCode,
        /// EDIFACT message reference.
        message_ref: MessageRef,
        /// Whether the IFTSTA message passed AHB validation.
        validation_passed: bool,
        /// Validation errors collected by the AHB validator.
        validation_errors: Vec<String>,
    },
}

impl CommandPayload for DeviceChangeCommand {}

// ── Workflow ──────────────────────────────────────────────────────────────────

/// WiM Messstellenbetrieb (PIDs 55039, 55042, 55051, 55168) workflow.
///
/// Implements the BDEW WiM process for change and management of meter operators
/// (MSB) at a Messlokation. The grid operator receives inbound UTILMD messages
/// from the MSBN and must respond with an APERAK within **5 Werktage**.
///
/// Spawn via [`mako_engine::process::Process`]:
/// ```rust,ignore
/// let process = ctx.spawn::<WimDeviceChangeWorkflow>(
///     tenant_id,
///     WorkflowId::new("wim-device-change", "FV2025-10-01"),
/// );
/// ```
pub struct WimDeviceChangeWorkflow;

impl Workflow for WimDeviceChangeWorkflow {
    type State = DeviceChangeState;
    type Event = DeviceChangeEvent;
    type Command = DeviceChangeCommand;

    /// Deadline compensation for the WiM Gerätewechsel 5-Werktage APERAK window.
    ///
    /// | Label | State guard | Command emitted | BNetzA rule |
    /// |---|---|---|---|
    /// | `"wim-aperak-5-werktage"` | `Initiated` or `ValidationPassed` | `TimeoutExpired` | BK6-18-032 — 5 Werktage APERAK Frist |
    fn on_deadline(
        deadline: &mako_engine::deadline::Deadline,
        state: &Self::State,
    ) -> Option<Self::Command> {
        match (deadline.label(), state) {
            (
                APERAK_WINDOW_LABEL,
                DeviceChangeState::Initiated(_) | DeviceChangeState::ValidationPassed(_),
            ) => Some(DeviceChangeCommand::TimeoutExpired {
                deadline_id: deadline.deadline_id(),
                label: deadline.label().into(),
            }),
            // Counterparty missed the 5-Werktage answer window on our outbound order.
            (AUFTRAG_ANTWORT_WINDOW_LABEL, DeviceChangeState::AuftragGesendet(_)) => {
                Some(DeviceChangeCommand::TimeoutExpired {
                    deadline_id: deadline.deadline_id(),
                    label: deadline.label().into(),
                })
            }
            _ => None,
        }
    }

    fn apply(state: Self::State, event: &Self::Event) -> Self::State {
        match event {
            DeviceChangeEvent::AuftragGesendet {
                melo_id,
                sender,
                receiver,
                process_date,
                message_ref,
                pruefidentifikator,
            } => DeviceChangeState::AuftragGesendet(DeviceChangeData {
                melo_id: melo_id.clone(),
                // On an outbound order this party is the sender; `incoming_msb`
                // and `grid_operator` are populated by PID direction so the
                // projection stays meaningful either way.
                incoming_msb: sender.clone(),
                grid_operator: receiver.clone(),
                device_id: DeviceId::new(""),
                document_date: process_date.clone(),
                pruefidentifikator: *pruefidentifikator,
                message_ref: Some(message_ref.clone()),
            }),
            DeviceChangeEvent::AntwortEmpfangen {
                is_confirmed,
                reason,
                ..
            } => match state {
                DeviceChangeState::AuftragGesendet(data) => {
                    if *is_confirmed {
                        DeviceChangeState::AuftragBestaetigt(data)
                    } else {
                        DeviceChangeState::Rejected {
                            reason: reason
                                .clone()
                                .unwrap_or_else(|| "Auftrag vom Marktpartner abgelehnt".to_owned()),
                        }
                    }
                }
                other => other,
            },
            DeviceChangeEvent::Initiated {
                melo_id,
                incoming_msb,
                grid_operator,
                device_id,
                document_date,
                message_ref,
                pruefidentifikator,
            } => DeviceChangeState::Initiated(DeviceChangeData {
                melo_id: melo_id.clone(),
                incoming_msb: incoming_msb.clone(),
                grid_operator: grid_operator.clone(),
                device_id: device_id.clone(),
                document_date: document_date.clone(),
                pruefidentifikator: *pruefidentifikator,
                message_ref: Some(message_ref.clone()),
            }),
            DeviceChangeEvent::ValidationPassed { .. } => {
                if let DeviceChangeState::Initiated(data) = state {
                    DeviceChangeState::ValidationPassed(data)
                } else {
                    state
                }
            }
            DeviceChangeEvent::AperakDispatched { positive, .. } => match state {
                DeviceChangeState::ValidationPassed(data) => {
                    if *positive {
                        DeviceChangeState::AperakSent(data)
                    } else {
                        DeviceChangeState::Rejected {
                            reason: "negative APERAK".to_owned(),
                        }
                    }
                }
                _ => state,
            },
            DeviceChangeEvent::Completed { device_id } => match state {
                DeviceChangeState::AperakSent(mut data)
                | DeviceChangeState::AuftragBestaetigt(mut data) => {
                    data.device_id = device_id.clone();
                    DeviceChangeState::Completed(data)
                }
                other => other,
            },
            DeviceChangeEvent::Rejected { reason } => DeviceChangeState::Rejected {
                reason: reason.clone(),
            },
            DeviceChangeEvent::DeadlineExpired { label, .. } => match state {
                DeviceChangeState::Completed(_) | DeviceChangeState::Rejected { .. } => state,
                _ => DeviceChangeState::Rejected {
                    reason: format!("deadline expired: {label}"),
                },
            },

            // Informational WiM IFTSTA status messages do not change state.
            DeviceChangeEvent::IftstaStatusReceived { .. } => state,
        }
    }

    fn handle(
        state: &Self::State,
        command: Self::Command,
    ) -> Result<WorkflowOutput<Self::Event>, WorkflowError> {
        match command {
            DeviceChangeCommand::InitiateDeviceChange {
                pid,
                sender,
                receiver,
                melo_id,
                process_date,
                message_ref,
            } => {
                if !matches!(state, DeviceChangeState::New) {
                    return Err(WorkflowError::invalid_state("New", state.status_str()));
                }
                if !DEVICE_CHANGE_PIDS.contains(&pid.as_u32()) {
                    return Err(WorkflowError::rejected(format!(
                        "expected a WiM MSB-Wechsel PID (55039, 55042, 55051, 55168), got {pid}",
                    )));
                }

                // Key set required by `edifact_renderer::render_utilmd` for a WiM
                // UTILMD: pid, sender, receiver, melo, process_date.
                let outbox = PendingOutbox::new(
                    "UTILMD",
                    receiver.as_str(),
                    serde_json::json!({
                        "direction":    "outbound",
                        "pid":          pid.as_u32(),
                        "sender":       sender.as_str(),
                        "receiver":     receiver.as_str(),
                        "melo":         melo_id.as_str(),
                        "process_date": process_date,
                        "message_ref":  message_ref.as_str(),
                    }),
                )
                .caused_by(0);

                let event = DeviceChangeEvent::AuftragGesendet {
                    melo_id,
                    sender,
                    receiver,
                    process_date,
                    message_ref,
                    pruefidentifikator: pid,
                };
                Ok(WorkflowOutput::with_outbox(vec![event], vec![outbox]))
            }

            DeviceChangeCommand::ReceiveAntwort {
                pid,
                sender,
                message_ref,
                reason,
            } => {
                let DeviceChangeState::AuftragGesendet(data) = state else {
                    return Err(WorkflowError::invalid_state(
                        "AuftragGesendet",
                        state.status_str(),
                    ));
                };

                let Some((request_pid, is_confirmed)) = antwort_pid_meaning(pid.as_u32()) else {
                    return Err(WorkflowError::rejected(format!(
                        "PID {pid} is not a WiM MSB-Wechsel Antwort (expected one of \
                         55040, 55041, 55043, 55044, 55052, 55053, 55169, 55170)",
                    )));
                };

                // The answer must belong to the order we actually sent. Without this
                // check a 55043 (Anmeldung confirmed) could silently close a 55039
                // (Kündigung) order and the audit trail would claim the wrong process
                // completed.
                if request_pid != data.pruefidentifikator.as_u32() {
                    return Err(WorkflowError::rejected(format!(
                        "Antwort PID {pid} answers request {request_pid}, but this process \
                         sent {}",
                        data.pruefidentifikator,
                    )));
                }

                Ok(vec![DeviceChangeEvent::AntwortEmpfangen {
                    pruefidentifikator: pid,
                    sender,
                    message_ref,
                    is_confirmed,
                    reason,
                }]
                .into())
            }

            DeviceChangeCommand::ReceiveRestOrder {
                tx_id,
                sender_mp_id,
                melo_id,
                device_category,
                process_date,
            } => {
                if !matches!(state, DeviceChangeState::New) {
                    return Err(WorkflowError::invalid_state("New", state.status_str()));
                }
                // PID 55042 (WiM MSB Anmeldung Strom) is the canonical EDIFACT process
                // identifier for iMSys Universalbestellprozess Anmeldung regardless of
                // transport channel.  REST (API-Webdienste Strom) and EDIFACT (UTILMD)
                // both initiate the same underlying MaKo process; 55042 keeps the audit
                // trail consistent and avoids phantom PIDs in the event store.
                let pid = Pruefidentifikator::new(55_042).map_err(|e| {
                    WorkflowError::rejected(format!(
                        "constant PID 55042 (WiM Anmeldung MSB) invalid: {e}"
                    ))
                })?;
                // REST orders carry no EDIFACT device ID; use the tx_id as a
                // provisional placeholder until the MSB assigns a device EIC.
                let device_id = DeviceId::new(&*tx_id);
                let message_ref = MessageRef::new(&*tx_id);
                Ok(vec![
                    DeviceChangeEvent::Initiated {
                        melo_id,
                        incoming_msb: sender_mp_id,
                        // REST orders target the MSB (self); grid_operator is
                        // not known at this point — carry device_category in
                        // document_date for now (process_date holds the date).
                        grid_operator: MarktpartnerCode::new(""),
                        device_id,
                        document_date: format!("{process_date}|category={device_category}"),
                        message_ref: message_ref.clone(),
                        pruefidentifikator: pid,
                    },
                    // REST-sourced orders are structurally valid by definition
                    // (the HTTP layer validated the JSON payload); emit
                    // ValidationPassed immediately.
                    DeviceChangeEvent::ValidationPassed { message_ref },
                ]
                .into())
            }

            DeviceChangeCommand::ReceiveUtilmd {
                pid,
                sender,
                receiver,
                melo_id,
                device_id,
                document_date,
                message_ref,
                validation_passed,
                validation_errors,
                received_at,
            } => {
                if !matches!(state, DeviceChangeState::New) {
                    return Err(WorkflowError::invalid_state("New", state.status_str()));
                }
                // PID guard: reject any PID not in the WiM MSB-Wechsel family.
                // Only PIDs 55039, 55042, 55051, 55168 are registered by WimModule;
                // this guard is defence-in-depth for direct callers.
                let valid_pids = [55_039_u32, 55_042, 55_051, 55_168];
                if !valid_pids.contains(&pid.as_u32()) {
                    return Err(WorkflowError::rejected(format!(
                        "PID {} is not a WiM Messstellenbetrieb PID (expected 55039, 55042, 55051, or 55168)",
                        pid.as_u32()
                    )));
                }
                // Clone before move for APERAK emission in the validation-failed path.
                let sender_mp_id = sender.clone();
                let receiver_gln = receiver.clone();

                let mut events = vec![DeviceChangeEvent::Initiated {
                    melo_id,
                    incoming_msb: sender,
                    grid_operator: receiver,
                    device_id,
                    document_date,
                    message_ref: message_ref.clone(),
                    pruefidentifikator: pid,
                }];
                if validation_passed {
                    events.push(DeviceChangeEvent::ValidationPassed { message_ref });
                    // WiM Ger\u00e4tewechsel: the APERAK is dispatched by the ERP within 5 Werktage
                    // (BK6-24-174 \u00a72a) \u2014 NOT auto-emitted here. DispatchAperak is the single
                    // APERAK decision point for both positive (BGM+312) and negative (BGM+313).
                    //
                    // Register TWO deadlines atomically with the events:
                    //   1. APERAK Strom *sending* deadline (APERAK AHB 1.0 \u00a72.4.1):
                    //      weekday = 45 min; Saturday = Sunday noon.
                    //      Label: APERAK_STROM_WINDOW_LABEL
                    //   2. WiM 5-Werktage *process response* deadline (BK6-24-174 \u00a72a):
                    //      the NB must issue the positive/negative APERAK within 5 WT.
                    //      Label: APERAK_WINDOW_LABEL ("wim-aperak-5-werktage")
                    let aperak_send_dl = PendingDeadline::new(
                        APERAK_STROM_WINDOW_LABEL,
                        aperak_strom_due_at(received_at),
                    );
                    let process_dl = PendingDeadline::new(
                        APERAK_WINDOW_LABEL,
                        deadline_at_werktage(received_at, 5, HolidayCalendar::BdewMaKo),
                    );
                    Ok(WorkflowOutput::with_outbox_and_deadlines(
                        events,
                        vec![],
                        vec![aperak_send_dl, process_dl],
                    ))
                } else {
                    let reason = validation_errors.join("; ");
                    events.push(DeviceChangeEvent::Rejected {
                        reason: reason.clone(),
                    });
                    // F-035: APERAK BGM+313 \u2014 mandatory per APERAK AHB 1.0 \u00a72.1.1.
                    // Validation failed \u2192 APERAK sent immediately: register the 45-min
                    // *sending* deadline so the OutboxWorker is monitored (APERAK AHB 1.0 \u00a72.4.1).
                    let aperak_send_dl = PendingDeadline::new(
                        APERAK_STROM_WINDOW_LABEL,
                        aperak_strom_due_at(received_at),
                    );
                    let outbox = vec![
                        PendingOutbox::new(
                            "APERAK",
                            sender_mp_id.as_str(),
                            serde_json::json!({
                                "sender":     receiver_gln.as_str(),
                                "receiver":   sender_mp_id.as_str(),
                                "pid":        29001_u32,
                                "error_code": mako_engine::erc::codes::Z29,
                                "reason":     reason,
                            }),
                        )
                        .caused_by(0),
                    ];
                    Ok(WorkflowOutput::with_outbox_and_deadlines(
                        events,
                        outbox,
                        vec![aperak_send_dl],
                    ))
                }
            }

            DeviceChangeCommand::DispatchAperak { positive, reason } => {
                let data = match state {
                    DeviceChangeState::ValidationPassed(d) => d,
                    _ => {
                        return Err(WorkflowError::invalid_state(
                            "ValidationPassed",
                            state.status_str(),
                        ));
                    }
                };
                // Always enqueue an APERAK outbox entry so the ERP layer sees the
                // business decision.  The renderer/outbox worker translates the
                // domain payload into the wire APERAK:
                //   positive = true  → BGM+312 (Bestätigung Gerätewechsel, within 5 WT)
                //   positive = false → BGM+313 (Ablehnung, within 5 WT)
                // Both polarities share the same domain payload schema so the ERP
                // can record the outcome uniformly (APERAK AHB 1.0 §2.4).
                // `sender` = grid_operator: the NB sends the APERAK to the incoming MSB.
                let mut aperak_payload = serde_json::json!({
                    "sender":   data.grid_operator.as_str(),
                    "pid":      data.pruefidentifikator.as_u32(),
                    "melo":     data.melo_id.as_str(),
                    "positive": positive,
                });
                if let Some(ref mr) = data.message_ref {
                    aperak_payload["orig_message_ref"] =
                        serde_json::Value::String(mr.as_str().to_owned());
                }
                if let Some(ref r) = reason {
                    aperak_payload["reason"] = serde_json::Value::String(r.clone());
                }
                let outbox_entry =
                    PendingOutbox::new("APERAK", data.incoming_msb.as_str(), aperak_payload)
                        .caused_by(0);
                Ok(WorkflowOutput::with_outbox(
                    vec![DeviceChangeEvent::AperakDispatched { positive, reason }],
                    vec![outbox_entry],
                ))
            }

            DeviceChangeCommand::Complete { device_id } => {
                // Reachable from both directions: `AperakSent` closes an inbound
                // order we acknowledged, `AuftragBestaetigt` closes an outbound
                // order the counterparty confirmed.
                if !matches!(
                    state,
                    DeviceChangeState::AperakSent(_) | DeviceChangeState::AuftragBestaetigt(_)
                ) {
                    return Err(WorkflowError::invalid_state(
                        "AperakSent or AuftragBestaetigt",
                        state.status_str(),
                    ));
                }
                Ok(vec![DeviceChangeEvent::Completed { device_id }].into())
            }

            DeviceChangeCommand::TimeoutExpired { deadline_id, label } => {
                if matches!(
                    state,
                    DeviceChangeState::Completed(_) | DeviceChangeState::Rejected { .. }
                ) {
                    return Ok(WorkflowOutput::events(vec![]));
                }
                Ok(vec![DeviceChangeEvent::DeadlineExpired { deadline_id, label }].into())
            }

            DeviceChangeCommand::ReceiveIftsta {
                pid,
                sender,
                receiver,
                message_ref,
                ..
            } => {
                // WiM IFTSTA messages are informational. Accept them in any
                // state (the process may already be completed when a late
                // Vollzugsmeldung arrives) and record for audit purposes.
                Ok(vec![DeviceChangeEvent::IftstaStatusReceived {
                    pid,
                    sender,
                    receiver,
                    message_ref,
                }]
                .into())
            }
        }
    }
}

// ── Read-model projection ─────────────────────────────────────────────────────

/// Read-model record for a single WiM Gerätewechsel process stream.
///
/// Uses a type-state design so field access never requires `Option::unwrap`:
/// the `Active` variant carries all domain fields that are structurally
/// guaranteed to exist once the process moves past `New`.
#[derive(Debug)]
pub enum DeviceChangeRecord {
    /// No `Initiated` event applied yet.
    New {
        /// Total events applied so far (should be 0).
        event_count: usize,
    },
    /// `Initiated` event applied; process fields now available.
    Active {
        /// Current lifecycle stage.
        status: &'static str,
        /// Messlokation EIC code.
        melo_id: MeLo,
        /// GLN of the incoming Messstellenbetreiber.
        incoming_msb: MarktpartnerCode,
        /// GLN of the grid operator.
        grid_operator: MarktpartnerCode,
        /// Physical device identifier (updated on `Completed`).
        device_id: DeviceId,
        /// BDEW Prüfidentifikator.
        pruefidentifikator: Pruefidentifikator,
        /// Total events applied.
        event_count: usize,
    },
}

impl DeviceChangeRecord {
    /// Current lifecycle status label, suitable for logging and serialisation.
    #[must_use]
    pub fn status(&self) -> &'static str {
        match self {
            Self::New { .. } => "New",
            Self::Active { status, .. } => status,
        }
    }

    /// Total events applied to this stream.
    #[must_use]
    pub fn event_count(&self) -> usize {
        match self {
            Self::New { event_count } | Self::Active { event_count, .. } => *event_count,
        }
    }

    /// Domain data for this record if it has been initiated, or `None` if `New`.
    #[must_use]
    pub fn active_data(&self) -> Option<DeviceChangeRecordData<'_>> {
        match self {
            Self::New { .. } => None,
            Self::Active {
                melo_id,
                incoming_msb,
                grid_operator,
                device_id,
                pruefidentifikator,
                ..
            } => Some(DeviceChangeRecordData {
                melo_id,
                incoming_msb,
                grid_operator,
                device_id,
                pruefidentifikator,
            }),
        }
    }
}

/// Borrowed view of the domain fields in an `Active` `DeviceChangeRecord`.
#[derive(Debug, Clone, Copy)]
pub struct DeviceChangeRecordData<'a> {
    /// Messlokation EIC code.
    pub melo_id: &'a MeLo,
    /// GLN of the incoming Messstellenbetreiber.
    pub incoming_msb: &'a MarktpartnerCode,
    /// GLN of the grid operator.
    pub grid_operator: &'a MarktpartnerCode,
    /// Physical device identifier.
    pub device_id: &'a DeviceId,
    /// BDEW Prüfidentifikator.
    pub pruefidentifikator: &'a Pruefidentifikator,
}

impl Default for DeviceChangeRecord {
    fn default() -> Self {
        Self::New { event_count: 0 }
    }
}

/// In-process read model that tracks status across all WiM Gerätewechsel
/// streams. Feed via [`mako_engine::projection::ProjectionRunner`].
#[derive(Debug, Default)]
pub struct DeviceChangeProjection {
    /// Map of stream ID → record.
    pub records: HashMap<String, DeviceChangeRecord>,
    /// Highest event sequence number processed.
    pub last_seq: u64,
}

impl Projection for DeviceChangeProjection {
    fn name(&self) -> &'static str {
        "DeviceChangeProjection"
    }

    fn handle_event(&mut self, envelope: &EventEnvelope) {
        self.last_seq = self.last_seq.max(envelope.sequence_number);

        let record = self
            .records
            .entry(envelope.stream_id.as_str().to_owned())
            .or_default();

        let Ok(event) = envelope.decode::<DeviceChangeEvent>() else {
            return;
        };

        // Increment event count on every decoded event.
        match record {
            DeviceChangeRecord::New { event_count } => *event_count += 1,
            DeviceChangeRecord::Active { event_count, .. } => *event_count += 1,
        }

        match event {
            DeviceChangeEvent::AuftragGesendet {
                melo_id,
                sender,
                receiver,
                pruefidentifikator,
                ..
            } => {
                let count = record.event_count();
                *record = DeviceChangeRecord::Active {
                    status: "AuftragGesendet",
                    melo_id,
                    incoming_msb: sender,
                    grid_operator: receiver,
                    device_id: DeviceId::new(""),
                    pruefidentifikator,
                    event_count: count,
                };
            }
            DeviceChangeEvent::Initiated {
                melo_id,
                incoming_msb,
                grid_operator,
                device_id,
                pruefidentifikator,
                ..
            } => {
                let count = record.event_count();
                *record = DeviceChangeRecord::Active {
                    status: "Initiated",
                    melo_id,
                    incoming_msb,
                    grid_operator,
                    device_id,
                    pruefidentifikator,
                    event_count: count,
                };
            }
            DeviceChangeEvent::ValidationPassed { .. } => {
                if let DeviceChangeRecord::Active { status, .. } = record {
                    *status = "ValidationPassed";
                }
            }
            DeviceChangeEvent::AntwortEmpfangen { is_confirmed, .. } => {
                if let DeviceChangeRecord::Active { status, .. } = record {
                    *status = if is_confirmed {
                        "AuftragBestaetigt"
                    } else {
                        "Rejected"
                    };
                }
            }
            DeviceChangeEvent::AperakDispatched { positive, .. } => {
                if let DeviceChangeRecord::Active { status, .. } = record {
                    *status = if positive { "AperakSent" } else { "Rejected" };
                }
            }
            DeviceChangeEvent::Completed { device_id } => {
                if let DeviceChangeRecord::Active {
                    status,
                    device_id: d,
                    ..
                } = record
                {
                    *status = "Completed";
                    *d = device_id;
                }
            }
            DeviceChangeEvent::Rejected { .. } => {
                if let DeviceChangeRecord::Active { status, .. } = record {
                    *status = "Rejected";
                }
            }
            DeviceChangeEvent::DeadlineExpired { .. } => {
                if let DeviceChangeRecord::Active { status, .. } = record {
                    *status = "Rejected";
                }
            }
            DeviceChangeEvent::IftstaStatusReceived { .. } => {
                // Informational — does not change the status label.
            }
        }
    }
}

// ── Unit tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    fn make_receive_cmd(pid: u32, validation_passed: bool) -> DeviceChangeCommand {
        DeviceChangeCommand::ReceiveUtilmd {
            pid: Pruefidentifikator::new(pid).expect("test pid must be in range"),
            sender: MarktpartnerCode::new("4012345000023"),
            receiver: MarktpartnerCode::new("9900357000004"),
            melo_id: MeLo::new("DE0000000001234567890000000000001"),
            device_id: DeviceId::new("ZHR-12345678"),
            document_date: "20250115".to_owned(),
            message_ref: MessageRef::new("MSG-WIM-001"),
            validation_passed,
            validation_errors: if validation_passed {
                vec![]
            } else {
                vec!["AHB rule violation".to_owned()]
            },
            received_at: time::OffsetDateTime::now_utc(),
        }
    }

    #[test]
    fn happy_path_new_to_completed() {
        let state = DeviceChangeState::default();

        let events = WimDeviceChangeWorkflow::handle(&state, make_receive_cmd(55042, true))
            .expect("should accept valid PID 55042");
        assert_eq!(events.len(), 2);
        assert!(
            matches!(&events[0], DeviceChangeEvent::Initiated { pruefidentifikator, .. } if pruefidentifikator.as_u32() == 55042)
        );
        assert!(matches!(
            &events[1],
            DeviceChangeEvent::ValidationPassed { .. }
        ));

        let state = events.iter().fold(state, WimDeviceChangeWorkflow::apply);
        assert!(
            matches!(&state, DeviceChangeState::ValidationPassed(_)),
            "expected ValidationPassed, got {}",
            state.status_str()
        );

        let events = WimDeviceChangeWorkflow::handle(
            &state,
            DeviceChangeCommand::DispatchAperak {
                positive: true,
                reason: None,
            },
        )
        .expect("dispatch APERAK");
        let state = events.iter().fold(state, WimDeviceChangeWorkflow::apply);
        assert!(
            matches!(&state, DeviceChangeState::AperakSent(_)),
            "expected AperakSent"
        );

        let events = WimDeviceChangeWorkflow::handle(
            &state,
            DeviceChangeCommand::Complete {
                device_id: DeviceId::new("ZHR-99999999"),
            },
        )
        .expect("complete");
        let state = events.iter().fold(state, WimDeviceChangeWorkflow::apply);
        assert!(
            matches!(&state, DeviceChangeState::Completed(d) if d.device_id == DeviceId::new("ZHR-99999999")),
            "expected Completed with new device_id",
        );
    }

    #[test]
    fn wrong_pid_is_rejected() {
        let state = DeviceChangeState::default();
        let err = WimDeviceChangeWorkflow::handle(&state, make_receive_cmd(55001, true))
            .expect_err("should reject wrong PID");
        let msg = err.to_string();
        assert!(
            msg.contains("55001"),
            "error should mention the supplied PID: {msg}"
        );
    }

    #[test]
    fn validation_failure_rejects_process() {
        let state = DeviceChangeState::default();
        let events = WimDeviceChangeWorkflow::handle(&state, make_receive_cmd(55042, false))
            .expect("should still produce events");
        assert!(matches!(&events[1], DeviceChangeEvent::Rejected { .. }));
        let state = events.iter().fold(state, WimDeviceChangeWorkflow::apply);
        assert!(
            matches!(&state, DeviceChangeState::Rejected { .. }),
            "expected Rejected"
        );
    }

    #[test]
    fn dispatch_aperak_in_wrong_state_is_rejected() {
        // Status is New (not ValidationPassed)
        let state = DeviceChangeState::default();
        let err = WimDeviceChangeWorkflow::handle(
            &state,
            DeviceChangeCommand::DispatchAperak {
                positive: true,
                reason: None,
            },
        )
        .expect_err("should reject dispatch in wrong state");
        assert!(err.to_string().contains("ValidationPassed"), "{err}");
    }
}