Skip to main content

mako_engine/
pid_router.rs

1//! PID-to-workflow routing table.
2//!
3//! Every inbound EDIFACT message carries a `Prüfidentifikator` (PID) that
4//! identifies the MaKo process family and operation. The `PidRouter` maps
5//! numeric PID values to workflow names, enabling dispatchers to instantiate
6//! the correct [`Workflow`] implementation without ad-hoc `match` chains.
7//!
8//! # Mutability contract — build-time only
9//!
10//! `PidRouter` uses `&mut self` for all registrations. In normal engine usage
11//! the router is populated **once** during `EngineBuilder::build()` and is
12//! subsequently sealed inside `EngineContext` behind a shared `&PidRouter`
13//! reference. There is no runtime mutation path — all PIDs must be registered
14//! before the engine starts serving messages.
15//!
16//! This is intentional: mutation after startup would race with concurrent
17//! dispatch calls and require a `RwLock`. The read-only runtime path is
18//! therefore always lock-free.
19//!
20//! # BDEW PID ranges (incomplete — register all PIDs for your process families)
21//!
22//! | Range | Process family |
23//! |---|---|
24//! | 11001–11099 | WiM Gerätewechsel (UTILMD) |
25//! | 13003        | MABIS Bilanzkreisabrechnung (MSCONS) |
26//! | 13002–13028  | Messwerte Gas/Strom/Redispatch (MSCONS) — fragmented across GaBi Gas, Redispatch, GPKE support |
27//! | 17001, 17002, 17009 | WiM Geräteübernahme (ORDERS) |
28//! | 17101–17135 | WiM Stammdaten / Konfiguration (ORDERS) |
29//! | 31001–31002, 31004–31008 | GPKE Netznutzungsabrechnung / MMM-Rechnung (INVOIC) |
30//! | 31003, 31009 | WiM-Rechnung / MSB-Rechnung (INVOIC) — WiM domain |
31//! | 31010 | Kapazitätsrechnung (INVOIC) — Kapazitätsabrechnung Ausspeisepunkte Gas |
32//! | 31011 | Rechnung sonstige Leistung (INVOIC) — AWH Sperrprozesse Gas |
33//! | 33001–33004  | REMADV Bestätigung/Abweisung — paired with INVOIC workflows |
34//! | 37000–37006  | PARTIN Kommunikationsdaten Strom (GPKE Teil 4) |
35//! | 37008–37014  | PARTIN Kommunikationsdaten Gas (GeLi Gas 2.0) |
36//! | 39000–39001 | ORDCHG Stornierung Sperr-/Entsperrauftrag (AWH Sperrprozesse Gas) |
37//! | 39002 | ORDCHG Stornierung Bestellung (WiM Strom Teil 2) |
38//! | 44001–44018 | GeLi Gas Lieferantenwechsel (UTILMD G) |
39//! | 55001–55018 | GPKE Lieferantenwechsel / Kündigung (UTILMD Strom) |
40//! | 55555 | GPKE Teil 4 — Anfrage Daten der individuellen Bestellung (UTILMD Strom) |
41//!
42//! # Usage
43//!
44//! ```rust
45//! use mako_engine::pid_router::PidRouter;
46//!
47//! let mut router = PidRouter::new();
48//! router.register(55001, "GpkeSupplierChange");
49//! router.register(55002, "GpkeSupplierChange"); // Same workflow, different step
50//!
51//! assert_eq!(router.route(55001), Some("GpkeSupplierChange"));
52//! assert_eq!(router.route(99999), None);
53//! assert_eq!(router.len(), 2);
54//! ```
55//!
56//! [`Workflow`]: crate::workflow::Workflow
57
58use std::collections::HashMap;
59
60use crate::types::Sparte;
61
62// ── PidRouter ─────────────────────────────────────────────────────────────────
63
64/// A static mapping from `Prüfidentifikator` (PID) values to workflow names.
65///
66/// Register all PIDs your platform handles before starting the engine. At
67/// runtime, call [`route`] to look up the workflow name for an inbound PID.
68///
69/// The workflow name matches [`WorkflowId::name`] — use it to select the
70/// correct `Workflow` implementation in your message dispatcher.
71///
72/// # Mutability contract
73///
74/// `PidRouter` exposes a `&mut self` API for registrations (`register`). In
75/// the engine this mutability is exercised **only** during
76/// [`EngineBuilder::build`] — after that the router is owned by
77/// [`EngineContext`] and only shared references are available at runtime.
78/// There is no way to mutate the router from an async dispatch handler.
79///
80/// Duplicate registrations silently replace the previous mapping; the last
81/// call wins. Use `cargo xtask validate-pruefids` to detect PID conflicts
82/// between modules before they reach production.
83///
84/// # Building a complete router
85///
86/// In your `main` or integration module, register every PID that the platform
87/// must handle. PIDs not registered will return `None` from [`route`], causing
88/// the dispatcher to dead-letter the message cleanly.
89///
90/// ```rust
91/// use mako_engine::pid_router::PidRouter;
92///
93/// fn build_router() -> PidRouter {
94///     let mut r = PidRouter::new();
95///     // GPKE Lieferantenwechsel (BK6-22-024) — UTILMD
96///     r.register(55001, "GpkeSupplierChange");
97///     r.register(55002, "GpkeSupplierChange");
98///     r.register(55003, "GpkeSupplierChange");
99///     r.register(55004, "GpkeSupplierChange");
100///     r
101/// }
102/// ```
103///
104/// [`route`]: PidRouter::route
105/// [`WorkflowId::name`]: crate::version::WorkflowId::name
106/// [`EngineBuilder::build`]: crate::builder::EngineBuilder::build
107/// [`EngineContext`]: crate::builder::EngineContext
108#[derive(Debug, Default, Clone)]
109pub struct PidRouter {
110    table: HashMap<u32, Box<str>>,
111    /// Commodity-qualified routing table: `(pid, Sparte) → workflow_name`.
112    ///
113    /// Checked first by [`route_with_sparte`]; falls back to the unambiguous
114    /// [`table`] when no commodity-specific entry exists.
115    ///
116    /// Use this for a Prüfidentifikator that two *different process families*
117    /// share across the Sparten. A PID that only differs in Frist or Codeliste
118    /// between Strom and Gas does not belong here: one workflow takes the
119    /// Sparte as an argument instead, which is what the WiM family does.
120    ///
121    /// [`route_with_sparte`]: PidRouter::route_with_sparte
122    /// [`table`]: PidRouter::table
123    commodity_table: HashMap<(u32, Sparte), Box<str>>,
124    /// Tracks which module registered each PID for conflict detection.
125    ///
126    /// Populated by [`register_with_module`]; used to produce actionable
127    /// panic messages when two modules register the same PID to different workflows.
128    ///
129    /// [`register_with_module`]: PidRouter::register_with_module
130    registered_by: HashMap<u32, Box<str>>,
131}
132
133impl PidRouter {
134    /// Create an empty router.
135    #[must_use]
136    pub fn new() -> Self {
137        Self::default()
138    }
139
140    /// Register `pid` as routing to `workflow_name`.
141    ///
142    /// If `pid` was already registered, the previous mapping is silently
143    /// replaced. Call this only at build time (via [`EngineModule::register_pids`]);
144    /// the method is `&mut self` to prevent accidental runtime mutation once the
145    /// router is sealed inside [`EngineContext`].
146    ///
147    /// Accepts any string — `&'static str`, `String`, or `Box<str>`.
148    ///
149    /// For conflict-detected registration (preferred in multi-module builds),
150    /// use [`register_with_module`] instead.
151    ///
152    /// [`EngineModule::register_pids`]: crate::builder::EngineModule::register_pids
153    /// [`EngineContext`]: crate::builder::EngineContext
154    /// [`register_with_module`]: PidRouter::register_with_module
155    pub fn register(&mut self, pid: u32, workflow_name: impl Into<Box<str>>) {
156        let wf = workflow_name.into();
157        self.table.insert(pid, wf);
158    }
159
160    /// Register `pid` → `workflow_name` with module-attribution conflict detection.
161    ///
162    /// # Panics
163    ///
164    /// Panics at **build time** (before the engine starts) if `pid` is already
165    /// registered to a *different* workflow name by a *different* module. Two
166    /// modules registering the same PID to the **same** workflow are silently
167    /// accepted (idempotent).
168    ///
169    /// Use [`DeploymentRoles`] to prevent two modules from registering the same
170    /// PID when only one role is active:
171    ///
172    /// ```rust,ignore
173    /// // Both GPKE (NB role) and WiM (nMSB role) register 19001 → different workflows.
174    /// // Set explicit roles so only one module's conditional block fires:
175    /// use mako_engine::marktrolle::{DeploymentRoles, Marktrolle};
176    /// let roles = DeploymentRoles::from_roles([Marktrolle::Nb]);
177    /// // Now only GPKE registers 19001 → "gpke-konfiguration".
178    /// ```
179    ///
180    /// [`DeploymentRoles`]: crate::marktrolle::DeploymentRoles
181    pub fn register_with_module(
182        &mut self,
183        pid: u32,
184        workflow_name: impl Into<Box<str>>,
185        module: &str,
186    ) {
187        let wf = workflow_name.into();
188        if let Some(existing_wf) = self.table.get(&pid)
189            && *existing_wf != wf
190        {
191            let existing_mod = self
192                .registered_by
193                .get(&pid)
194                .map_or("<unknown>", Box::as_ref);
195            panic!(
196                "PID {pid} routing conflict:\n  \
197                     module '{module}' tried to register PID {pid} → '{wf}'\n  \
198                     but it was already registered → '{existing_wf}' by module '{existing_mod}'\n  \
199                     Hint: use DeploymentRoles to prevent conflicting modules from \
200                     both registering shared PIDs (e.g. 19001/19002 are claimed by \
201                     gpke-konfiguration for NB role and wim-geraeteubernahme for nMSB role).\n  \
202                     Set EngineBuilder::with_deployment_roles(DeploymentRoles::nb()) to keep \
203                     only the NB-role registration."
204            );
205        }
206        self.table.insert(pid, wf);
207        self.registered_by.insert(pid, module.into());
208    }
209
210    /// Register `pid` → `workflow_name` for a specific commodity ([`Sparte`]).
211    ///
212    /// Use this for PIDs that map to **different workflows depending on whether
213    /// the message concerns electricity (Strom) or gas (Gas)**. At runtime call
214    /// [`route_with_sparte`] to prefer the commodity-specific entry over the
215    /// unambiguous fallback registered via [`register`].
216    ///
217    /// ```rust,ignore
218    /// // In GpkeModule (Strom):
219    /// router.register_with_sparte(17115, Sparte::Strom, "gpke-sperrung");
220    ///
221    /// // In GeliGasModule (Gas):
222    /// router.register_with_sparte(17115, Sparte::Gas, "geli-gas-sperrung-lf");
223    /// ```
224    ///
225    /// [`route_with_sparte`]: PidRouter::route_with_sparte
226    /// [`register`]: PidRouter::register
227    pub fn register_with_sparte(
228        &mut self,
229        pid: u32,
230        sparte: Sparte,
231        workflow_name: impl Into<Box<str>>,
232    ) {
233        self.commodity_table
234            .insert((pid, sparte), workflow_name.into());
235    }
236
237    /// Look up the workflow name for `pid`, preferring the commodity-qualified
238    /// entry for `sparte` over the unambiguous fallback.
239    ///
240    /// Resolution order:
241    /// 1. `commodity_table[(pid, sparte)]` — registered via [`register_with_sparte`]
242    /// 2. `table[pid]` — registered via [`register`] (unambiguous fallback)
243    ///
244    /// Returns `None` when neither table has an entry for `pid`.
245    ///
246    /// # Example
247    ///
248    /// ```rust
249    /// use mako_engine::pid_router::PidRouter;
250    /// use mako_engine::types::Sparte;
251    ///
252    /// let mut r = PidRouter::new();
253    /// r.register(17115, "gpke-sperrung");                            // Strom fallback
254    /// r.register_with_sparte(17115, Sparte::Strom, "gpke-sperrung");
255    /// r.register_with_sparte(17115, Sparte::Gas,   "geli-gas-sperrung-lf");
256    ///
257    /// assert_eq!(r.route_with_sparte(17115, Sparte::Strom), Some("gpke-sperrung"));
258    /// assert_eq!(r.route_with_sparte(17115, Sparte::Gas),   Some("geli-gas-sperrung-lf"));
259    /// assert_eq!(r.route(17115),                             Some("gpke-sperrung"));
260    /// ```
261    ///
262    /// [`register_with_sparte`]: PidRouter::register_with_sparte
263    /// [`register`]: PidRouter::register
264    #[must_use]
265    pub fn route_with_sparte(&self, pid: u32, sparte: Sparte) -> Option<&str> {
266        self.commodity_table
267            .get(&(pid, sparte))
268            .or_else(|| self.table.get(&pid))
269            .map(Box::as_ref)
270    }
271
272    /// Look up the workflow name for `pid`.
273    ///
274    /// Returns `None` when `pid` has not been registered. The caller should
275    /// dead-letter the message and return an appropriate error to the sender
276    /// rather than panicking.
277    #[must_use]
278    pub fn route(&self, pid: u32) -> Option<&str> {
279        self.table.get(&pid).map(Box::as_ref)
280    }
281
282    /// Return an iterator over all registered PID values.
283    ///
284    /// Useful for validation (e.g. comparing against PIDs declared in
285    /// AHB profile JSON files to detect missing workflow implementations).
286    pub fn registered_pids(&self) -> impl Iterator<Item = u32> + '_ {
287        self.table.keys().copied()
288    }
289
290    /// Return an iterator over all commodity-qualified routing entries.
291    ///
292    /// Yields `(pid, sparte, workflow_name)` tuples.
293    ///
294    /// Used by [`EngineBuilder`] to copy sparte-qualified entries from a scratch
295    /// router into the real [`PidRouter`] without conflict detection (commodity
296    /// entries use distinct `(pid, sparte)` keys and never conflict across modules).
297    ///
298    /// [`EngineBuilder`]: crate::builder::EngineBuilder
299    pub fn registered_commodity_entries(&self) -> impl Iterator<Item = (u32, Sparte, &str)> + '_ {
300        self.commodity_table
301            .iter()
302            .map(|((pid, sparte), wf)| (*pid, *sparte, wf.as_ref()))
303    }
304
305    /// Return a sorted, deduplicated list of all workflow names registered in
306    /// this router (across both the unambiguous table and commodity-qualified
307    /// table).
308    ///
309    /// Used at startup by `validate_dispatch_completeness` to verify that every
310    /// workflow name reachable via PID routing has a matching arm in the
311    /// `EdifactIngestDispatcher`.
312    #[must_use]
313    pub fn workflow_names(&self) -> Vec<&str> {
314        let mut names: Vec<&str> = self
315            .table
316            .values()
317            .map(Box::as_ref)
318            .chain(self.commodity_table.values().map(Box::as_ref))
319            .collect();
320        names.sort_unstable();
321        names.dedup();
322        names
323    }
324
325    /// Return the number of registered PID mappings.
326    #[must_use]
327    pub fn len(&self) -> usize {
328        self.table.len()
329    }
330
331    /// Return `true` when no PIDs have been registered.
332    #[must_use]
333    pub fn is_empty(&self) -> bool {
334        self.table.is_empty()
335    }
336}
337
338#[cfg(test)]
339mod tests {
340    use super::*;
341
342    #[test]
343    fn route_registered_pid() {
344        let mut r = PidRouter::new();
345        r.register(55001, "GpkeSupplierChange");
346        assert_eq!(r.route(55001), Some("GpkeSupplierChange"));
347    }
348    #[test]
349    fn route_unregistered_pid_returns_none() {
350        let r = PidRouter::new();
351        assert_eq!(r.route(55001), None);
352    }
353
354    #[test]
355    fn register_overwrites_previous_mapping() {
356        let mut r = PidRouter::new();
357        r.register(55001, "OldWorkflow");
358        r.register(55001, "NewWorkflow");
359        assert_eq!(r.route(55001), Some("NewWorkflow"));
360        assert_eq!(r.len(), 1);
361    }
362
363    #[test]
364    fn registered_pids_covers_all_entries() {
365        let mut r = PidRouter::new();
366        r.register(55001, "A");
367        r.register(55002, "B");
368        r.register(11001, "C");
369
370        let mut pids: Vec<u32> = r.registered_pids().collect();
371        pids.sort_unstable();
372        assert_eq!(pids, [11001, 55001, 55002]);
373    }
374
375    #[test]
376    fn multiple_pids_same_workflow() {
377        let mut r = PidRouter::new();
378        r.register(55001, "GpkeSupplierChange");
379        r.register(55002, "GpkeSupplierChange");
380        r.register(55003, "GpkeSupplierChange");
381
382        assert_eq!(r.len(), 3);
383        for pid in [55001, 55002, 55003] {
384            assert_eq!(r.route(pid), Some("GpkeSupplierChange"));
385        }
386    }
387
388    #[test]
389    fn is_empty_and_len() {
390        let mut r = PidRouter::new();
391        assert!(r.is_empty());
392        r.register(55001, "W");
393        assert!(!r.is_empty());
394        assert_eq!(r.len(), 1);
395    }
396
397    // ── Commodity-aware routing ───────────────────────────────────────────────
398    //
399    // The mechanism is exercised on ORDERS 17115 (Sperrauftrag), which GPKE and
400    // GeLi Gas answer with genuinely different processes. A PID whose two
401    // Sparten differ only in Frist or Codeliste does not belong here: one
402    // workflow takes the Sparte as an argument instead.
403
404    #[test]
405    fn route_with_sparte_prefers_commodity_entry() {
406        use crate::types::Sparte;
407        let mut r = PidRouter::new();
408        r.register(17115, "gpke-sperrung"); // unambiguous fallback
409        r.register_with_sparte(17115, Sparte::Strom, "gpke-sperrung");
410        r.register_with_sparte(17115, Sparte::Gas, "geli-gas-sperrung-lf");
411
412        assert_eq!(
413            r.route_with_sparte(17115, Sparte::Strom),
414            Some("gpke-sperrung")
415        );
416        assert_eq!(
417            r.route_with_sparte(17115, Sparte::Gas),
418            Some("geli-gas-sperrung-lf")
419        );
420        // Unambiguous route() is unaffected by commodity table.
421        assert_eq!(r.route(17115), Some("gpke-sperrung"));
422    }
423
424    #[test]
425    fn route_with_sparte_falls_back_to_unambiguous() {
426        use crate::types::Sparte;
427        let mut r = PidRouter::new();
428        // No commodity entry — only unambiguous.
429        r.register(55001, "GpkeSupplierChange");
430
431        assert_eq!(
432            r.route_with_sparte(55001, Sparte::Strom),
433            Some("GpkeSupplierChange")
434        );
435        assert_eq!(
436            r.route_with_sparte(55001, Sparte::Gas),
437            Some("GpkeSupplierChange")
438        );
439    }
440
441    #[test]
442    fn route_with_sparte_returns_none_for_unregistered() {
443        use crate::types::Sparte;
444        let r = PidRouter::new();
445        assert_eq!(r.route_with_sparte(17115, Sparte::Strom), None);
446        assert_eq!(r.route_with_sparte(23001, Sparte::Gas), None);
447    }
448
449    #[test]
450    fn route_with_sparte_gas_only_deployment() {
451        use crate::types::Sparte;
452        // Gas-standalone: only Gas entries; no Strom fallback registered.
453        let mut r = PidRouter::new();
454        r.register(17115, "geli-gas-sperrung-lf"); // unambiguous (Gas-standalone)
455        r.register_with_sparte(17115, Sparte::Gas, "geli-gas-sperrung-lf");
456
457        assert_eq!(
458            r.route_with_sparte(17115, Sparte::Gas),
459            Some("geli-gas-sperrung-lf")
460        );
461        // Strom falls back to unambiguous (no Strom-specific entry exists).
462        assert_eq!(
463            r.route_with_sparte(17115, Sparte::Strom),
464            Some("geli-gas-sperrung-lf")
465        );
466    }
467
468    #[test]
469    fn route_with_sparte_combined_deployment_keeps_both_families() {
470        use crate::types::Sparte;
471        let mut r = PidRouter::new();
472        // GpkeModule registers the Sperrprozess ORDERS as Strom:
473        for pid in [17115_u32, 17116, 17117] {
474            r.register(pid, "gpke-sperrung");
475            r.register_with_sparte(pid, Sparte::Strom, "gpke-sperrung");
476        }
477        // GeliGasModule registers the same PIDs as Gas:
478        for pid in [17115_u32, 17116, 17117] {
479            r.register_with_sparte(pid, Sparte::Gas, "geli-gas-sperrung-lf");
480        }
481
482        for pid in [17115_u32, 17116, 17117] {
483            assert_eq!(
484                r.route_with_sparte(pid, Sparte::Strom),
485                Some("gpke-sperrung"),
486                "PID {pid} Strom"
487            );
488            assert_eq!(
489                r.route_with_sparte(pid, Sparte::Gas),
490                Some("geli-gas-sperrung-lf"),
491                "PID {pid} Gas"
492            );
493        }
494    }
495}