mako_engine/partner.rs
1//! Trading-partner master data — the [`PartnerStore`] trait and supporting
2//! types.
3//!
4//! # Why not just a `HashMap<GLN, URL>` in config?
5//!
6//! The `partners = ["GLN=URL", …]` field in `makod.toml` works for
7//! development but falls short in production:
8//!
9//! | Requirement | Config-only | `PartnerStore` |
10//! |---|---|---|
11//! | Survives restarts without re-deployment | ❌ | ✅ |
12//! | Carries PARTIN-derived metadata (validity, contacts, bank) | ❌ | ✅ |
13//! | Updatable from inbound PARTIN messages at runtime | ❌ | ✅ |
14//! | Tenant-scoped isolation | ❌ | ✅ |
15//! | Multiple communication channels per partner | ❌ | ✅ |
16//! | Validity windows (Gültig Ab) for future-dated updates | ❌ | ✅ |
17//!
18//! # PARTIN data model
19//!
20//! The German energy market uses EDIFACT **PARTIN** messages (PIDs 37000–37014)
21//! to distribute market-participant master data. Each PARTIN carries:
22//!
23//! - `NAD` → GLN, company name, country code
24//! - `COM` → communication channels: AS4 endpoint URL, email, fax (up to 5)
25//! - `CCI/CAV` → availability windows (*Erreichbarkeit*)
26//! - `FII` → bank account (IBAN, BIC)
27//! - `RFF` → tax number, VAT ID
28//! - `CTA/NAD` → contact persons (*Ansprechpartner*)
29//! - `DTM` → valid-from date (*Gültig Ab*)
30//! - `CCI` → associated Bilanzkreis
31//!
32//! [`PartnerRecord`] captures all of these fields in a form that is both
33//! serializable to SlateDB and constructible from static config.
34//!
35//! # Bootstrap pattern
36//!
37//! ```rust,ignore
38//! // At startup — seed from makod.toml `[as4] partners` list:
39//! for record in PartnerRecord::from_cli_pairs(&config.as4.partners)? {
40//! store.upsert(tenant_id, &record).await?;
41//! }
42//!
43//! // Later — update from inbound PARTIN message:
44//! let record = parse_partin_37001(&edifact_interchange)?;
45//! store.upsert(tenant_id, &record).await?;
46//!
47//! // Outbound AS4 dispatch:
48//! let partner = store.get(tenant_id, &gln).await?
49//! .ok_or(EngineError::partner(format!("no endpoint for {mp_id}")))?;
50//! let endpoint = partner.as4_endpoint
51//! .ok_or(EngineError::partner(format!("{mp_id} has no AS4 endpoint")))?;
52//! ```
53//!
54//! # Key schema (SlateDB)
55//!
56//! `pt/{tenant_id}/{mp_id}` → `JSON(PartnerRecord)`
57//!
58//! Both `TenantId` and GLN are fixed-width strings, giving a
59//! `pt/{36-chars}/{13-chars}` prefix that bounds efficient per-tenant scans.
60
61use std::sync::Arc;
62
63#[cfg(any(test, feature = "testing"))]
64use std::collections::HashMap;
65#[cfg(any(test, feature = "testing"))]
66use tokio::sync::RwLock;
67
68use serde::{Deserialize, Serialize};
69use time::OffsetDateTime;
70
71use crate::{error::EngineError, ids::TenantId, types::MarktpartnerCode};
72
73// ── CommunicationChannel ──────────────────────────────────────────────────────
74
75/// A single communication channel extracted from a PARTIN `COM` segment.
76///
77/// PARTIN allows up to 5 `COM` segments per party. The `qualifier` uses the
78/// UN/EDIFACT DE 3155 code list:
79///
80/// | Qualifier | Meaning |
81/// |---|---|
82/// | `EM` | Electronic mail (primary) |
83/// | `AK` | Electronic mail (alternative) |
84/// | `TE` | Telephone |
85/// | `FX` | Fax |
86/// | `AS4` | BDEW AS4 endpoint URL (non-standard extension) |
87/// | `AW` | BDEW API-Webdienste Strom endpoint URL (Verzeichnisdienst-discovered) |
88///
89/// > **Note**: BDEW uses qualifier `AK` for the AS4 endpoint URL in PARTIN
90/// > AHB 1.0f. The `AS4` literal is used here as an explicit semantic label
91/// > for channels that have already been identified as AS4 endpoints.
92/// >
93/// > `AW` is a project-internal qualifier used to store the API-Webdienste
94/// > Strom base URL discovered from the BDEW Verzeichnisdienst.
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub struct CommunicationChannel {
97 /// DE 3155 communication qualifier (`EM`, `TE`, `FX`, `AK`, …).
98 pub qualifier: Box<str>,
99 /// The communication address (URL, email address, phone number).
100 pub address: Box<str>,
101}
102
103impl CommunicationChannel {
104 /// Construct a new channel.
105 #[must_use]
106 pub fn new(qualifier: impl Into<Box<str>>, address: impl Into<Box<str>>) -> Self {
107 Self {
108 qualifier: qualifier.into(),
109 address: address.into(),
110 }
111 }
112
113 /// Convenience: construct an AS4 endpoint channel.
114 ///
115 /// Uses qualifier `"AK"` per PARTIN AHB 1.0f DE 3155 convention.
116 #[must_use]
117 pub fn as4(endpoint_url: impl Into<Box<str>>) -> Self {
118 Self::new("AK", endpoint_url)
119 }
120
121 /// Convenience: construct an email channel.
122 #[must_use]
123 pub fn email(address: impl Into<Box<str>>) -> Self {
124 Self::new("EM", address)
125 }
126
127 /// Convenience: construct an API-Webdienste Strom endpoint channel.
128 ///
129 /// Uses qualifier `"AW"` (project-internal) to store the base URL
130 /// discovered from the BDEW Verzeichnisdienst for a given partner.
131 #[must_use]
132 pub fn api_webdienste(base_url: impl Into<Box<str>>) -> Self {
133 Self::new("AW", base_url)
134 }
135}
136
137// ── ContactPerson ─────────────────────────────────────────────────────────────
138
139/// A contact person extracted from a PARTIN `CTA`/`NAD`/`COM` group.
140///
141/// Corresponds to the *Ansprechpartner* group in PARTIN AHB 1.0f.
142#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
143pub struct ContactPerson {
144 /// Full name or department name.
145 pub name: Box<str>,
146 /// Contact channels (phone, email, …).
147 pub channels: Vec<CommunicationChannel>,
148}
149
150// ── MarketRole ────────────────────────────────────────────────────────────────
151
152/// The market role of a trading partner as declared in their PARTIN message.
153///
154/// Matches BDEW PARTIN Prüfidentifikator prefixes:
155///
156/// | PID | Role |
157/// |---|---|
158/// | 37000 | Lieferant Strom (`LfStrom`) |
159/// | 37001 | Netzbetreiber Strom (`NbStrom`) |
160/// | 37002 | Messstellenbetreiber Strom (`MsbStrom`) |
161/// | 37003 | Bilanzkreisverantwortlicher Strom (`Bkv`) |
162/// | 37004 | Bilanzkoordinator Strom (`Biko`) |
163/// | 37005 | Übertragungsnetzbetreiber Strom (`Uenb`) |
164/// | 37006 | Energiedienstleister/Serviceanbieter Strom (`Esa`) |
165/// | 37008 | Lieferant Gas (`LfGas`) |
166/// | 37009 | Netzbetreiber Gas (`NbGas`) |
167/// | 37010 | Messstellenbetreiber Gas (`MsbGas`) |
168/// | 37011 | Marktgebietsverantwortlicher Gas (`Mgv`) |
169/// | 37012–37014 | Cross-commodity roles |
170#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
171#[non_exhaustive]
172pub enum MarketRole {
173 /// Lieferant Strom (PID 37000)
174 LfStrom,
175 /// Netzbetreiber Strom (PID 37001)
176 NbStrom,
177 /// Messstellenbetreiber Strom (PID 37002)
178 MsbStrom,
179 /// Bilanzkreisverantwortlicher Strom (PID 37003)
180 Bkv,
181 /// Bilanzkoordinator Strom (PID 37004)
182 Biko,
183 /// Übertragungsnetzbetreiber Strom (PID 37005)
184 Uenb,
185 /// Energiedienstleister / Serviceanbieter Strom (PID 37006)
186 Esa,
187 /// Lieferant Gas (PID 37008)
188 LfGas,
189 /// Netzbetreiber Gas (PID 37009)
190 NbGas,
191 /// Messstellenbetreiber Gas (PID 37010)
192 MsbGas,
193 /// Marktgebietsverantwortlicher Gas (PID 37011)
194 Mgv,
195 /// Cross-commodity (PIDs 37012–37014)
196 CrossCommodity,
197}
198
199impl MarketRole {
200 /// Map a PARTIN Prüfidentifikator code to the corresponding `MarketRole`.
201 ///
202 /// Returns `None` for unrecognised codes.
203 #[must_use]
204 pub fn from_pid(pid: u32) -> Option<Self> {
205 match pid {
206 37000 => Some(Self::LfStrom),
207 37001 => Some(Self::NbStrom),
208 37002 => Some(Self::MsbStrom),
209 37003 => Some(Self::Bkv),
210 37004 => Some(Self::Biko),
211 37005 => Some(Self::Uenb),
212 37006 => Some(Self::Esa),
213 37008 => Some(Self::LfGas),
214 37009 => Some(Self::NbGas),
215 37010 => Some(Self::MsbGas),
216 37011 => Some(Self::Mgv),
217 37012..=37014 => Some(Self::CrossCommodity),
218 _ => None,
219 }
220 }
221}
222
223// ── PartnerRecord ─────────────────────────────────────────────────────────────
224
225/// Full trading-partner master record as stored in the [`PartnerStore`].
226///
227/// Populated either from static `makod.toml` config (minimal — GLN + AS4 URL
228/// only) or from an inbound PARTIN EDIFACT message (complete). Records from
229/// different sources coexist: a bootstrapped config record is upgraded in-place
230/// when the same partner later sends a PARTIN.
231///
232/// ## Constructors
233///
234/// - [`PartnerRecord::minimal`] — for bootstrapping from `GLN=URL` config pairs
235/// - [`PartnerRecord::from_cli_pairs`] — parse `[as4] partners` list from config
236///
237/// ## Merging
238///
239/// Use [`PartnerRecord::merge_from_partin`] to update an existing record with
240/// fields from a newer inbound PARTIN (respects validity dates).
241#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
242pub struct PartnerRecord {
243 /// The partner's 13-digit Global Location Number.
244 pub mp_id: MarktpartnerCode,
245
246 /// Company name from the PARTIN `NAD` segment.
247 pub display_name: Option<Box<str>>,
248
249 /// All communication channels from PARTIN `COM` segments.
250 ///
251 /// The AS4 endpoint is the entry with qualifier `"AK"` (PARTIN AHB 1.0f
252 /// DE 3155 convention). Use [`as4_endpoint`] for direct access.
253 ///
254 /// [`as4_endpoint`]: PartnerRecord::as4_endpoint
255 pub channels: Vec<CommunicationChannel>,
256
257 /// Market roles this partner has declared via PARTIN.
258 pub roles: Vec<MarketRole>,
259
260 /// Date from which this record version is valid (`DTM/137`).
261 ///
262 /// `None` when bootstrapped from static config (no validity date known).
263 #[serde(
264 default,
265 skip_serializing_if = "Option::is_none",
266 with = "time::serde::rfc3339::option"
267 )]
268 pub valid_from: Option<OffsetDateTime>,
269
270 /// Contact persons from the PARTIN *Ansprechpartner* group.
271 pub contacts: Vec<ContactPerson>,
272
273 /// ISO 3166-1 alpha-2 country code from `NAD+MS+++...+DE` (usually `DE`).
274 pub country_code: Option<Box<str>>,
275
276 /// Wall-clock time when this record was last written to the store.
277 #[serde(with = "time::serde::rfc3339")]
278 pub updated_at: OffsetDateTime,
279}
280
281impl PartnerRecord {
282 /// Create a minimal record from a GLN and an AS4 endpoint URL.
283 ///
284 /// Used when bootstrapping from `[as4] partners = ["GLN=URL", …]` in
285 /// `makod.toml`. The record has no PARTIN-derived metadata — only the
286 /// GLN and a single AS4 channel.
287 #[must_use]
288 pub fn minimal(mp_id: impl Into<MarktpartnerCode>, as4_url: impl Into<Box<str>>) -> Self {
289 Self {
290 mp_id: mp_id.into(),
291 display_name: None,
292 channels: vec![CommunicationChannel::as4(as4_url)],
293 roles: Vec::new(),
294 valid_from: None,
295 contacts: Vec::new(),
296 country_code: None,
297 updated_at: OffsetDateTime::now_utc(),
298 }
299 }
300
301 /// Parse `["GLN=HTTPS-URL", …]` configuration entries into minimal records.
302 ///
303 /// Returns an error on the first malformed or non-HTTPS entry.
304 ///
305 /// # Errors
306 ///
307 /// Returns [`EngineError::Partner`] when an entry lacks `=`, has an empty
308 /// GLN, or uses a non-HTTPS URL.
309 pub fn from_cli_pairs(pairs: &[impl AsRef<str>]) -> Result<Vec<Self>, EngineError> {
310 pairs
311 .iter()
312 .map(|entry| {
313 let pair = entry.as_ref();
314 let (mp_id, url) = pair.split_once('=').ok_or_else(|| {
315 EngineError::partner(format!(
316 "invalid partner entry {pair:?} — expected <GLN>=<HTTPS-URL>"
317 ))
318 })?;
319 let mp_id = mp_id.trim();
320 let url = url.trim();
321 if mp_id.is_empty() {
322 return Err(EngineError::partner(format!(
323 "invalid partner entry {pair:?} — GLN must not be empty"
324 )));
325 }
326 if !url.starts_with("https://") {
327 return Err(EngineError::partner(format!(
328 "invalid partner entry {pair:?} — endpoint URL must use HTTPS (got {url:?})"
329 )));
330 }
331 Ok(Self::minimal(mp_id, url))
332 })
333 .collect()
334 }
335
336 /// Return the AS4 endpoint URL if one has been registered.
337 ///
338 /// Looks for a channel with qualifier `"AK"` (PARTIN AHB 1.0f
339 /// convention for the AS4 endpoint). Falls back to `"AS4"` for records
340 /// that were imported with a non-standard qualifier.
341 #[must_use]
342 pub fn as4_endpoint(&self) -> Option<&str> {
343 self.channels
344 .iter()
345 .find(|c| c.qualifier.as_ref() == "AK" || c.qualifier.as_ref() == "AS4")
346 .map(|c| c.address.as_ref())
347 }
348
349 /// Return the primary email address if one has been registered.
350 ///
351 /// Looks for a channel with qualifier `"EM"`.
352 #[must_use]
353 pub fn email(&self) -> Option<&str> {
354 self.channels
355 .iter()
356 .find(|c| c.qualifier.as_ref() == "EM")
357 .map(|c| c.address.as_ref())
358 }
359
360 /// Return the API-Webdienste Strom base URL if one has been registered.
361 ///
362 /// Looks for a channel with qualifier `"AW"`. This URL is typically
363 /// populated by the Verzeichnisdienst discovery worker and is
364 /// used by `MaloIdentSender` to reach the LF's callback endpoint.
365 #[must_use]
366 pub fn api_webdienste_endpoint(&self) -> Option<&str> {
367 self.channels
368 .iter()
369 .find(|c| c.qualifier.as_ref() == "AW")
370 .map(|c| c.address.as_ref())
371 }
372
373 /// Merge fields from a newer PARTIN-derived record into `self`.
374 ///
375 /// Only updates `self` when `incoming.valid_from` is newer than
376 /// `self.valid_from` (or when `self.valid_from` is `None`). Config-
377 /// bootstrapped records (no `valid_from`) are always overwritten.
378 ///
379 /// The GLN must match — mismatches are silently ignored (the caller is
380 /// responsible for routing PARTIN messages to the correct record).
381 pub fn merge_from_partin(&mut self, incoming: PartnerRecord) {
382 if incoming.mp_id != self.mp_id {
383 return;
384 }
385 let should_update = match (self.valid_from, incoming.valid_from) {
386 (None, _) => true,
387 (Some(_), None) => false, // keep the dated record
388 (Some(a), Some(b)) => b >= a,
389 };
390 if !should_update {
391 return;
392 }
393 self.display_name = incoming.display_name.or(self.display_name.take());
394 self.channels = incoming.channels;
395 self.roles = incoming.roles;
396 self.valid_from = incoming.valid_from;
397 self.contacts = incoming.contacts;
398 self.country_code = incoming.country_code.or(self.country_code.take());
399 self.updated_at = incoming.updated_at;
400 }
401}
402
403// ── PartnerStore ──────────────────────────────────────────────────────────────
404
405/// Durable store for trading-partner master records.
406///
407/// Provides tenant-scoped access to [`PartnerRecord`]s. Records are upserted
408/// when a new PARTIN message arrives or when `makod` bootstraps from static
409/// config.
410///
411/// All three operations are idempotent — reinserting the same record is safe.
412///
413/// ## Blanket `Arc` implementation
414///
415/// `Arc<S>` implements `PartnerStore` whenever `S: PartnerStore`.
416#[allow(async_fn_in_trait)]
417pub trait PartnerStore: Send + Sync {
418 /// Insert or update the record for `(tenant_id, record.mp_id)`.
419 ///
420 /// If a record already exists for this GLN, it is **merged** via
421 /// [`PartnerRecord::merge_from_partin`] — i.e. the newer PARTIN-derived
422 /// record wins, but a config-only bootstrap is always overwritten.
423 ///
424 /// # Errors
425 ///
426 /// Returns [`EngineError::Partner`] on storage failure.
427 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError>;
428
429 /// Return the record for `(tenant_id, gln)`, or `None` if not registered.
430 ///
431 /// # Errors
432 ///
433 /// Returns [`EngineError::Partner`] on storage failure.
434 async fn get(
435 &self,
436 tenant_id: TenantId,
437 mp_id: &MarktpartnerCode,
438 ) -> Result<Option<PartnerRecord>, EngineError>;
439
440 /// Remove the record for `(tenant_id, gln)`.
441 ///
442 /// No-op when the record does not exist.
443 ///
444 /// # Errors
445 ///
446 /// Returns [`EngineError::Partner`] on storage failure.
447 async fn remove(
448 &self,
449 tenant_id: TenantId,
450 mp_id: &MarktpartnerCode,
451 ) -> Result<(), EngineError>;
452
453 /// Return all records registered for `tenant_id`.
454 ///
455 /// # Errors
456 ///
457 /// Returns [`EngineError::Partner`] on storage failure.
458 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError>;
459
460 /// Return the AS4 endpoint URL for `gln`, if known.
461 ///
462 /// Convenience wrapper over `get` + `as4_endpoint`.
463 ///
464 /// # Errors
465 ///
466 /// Returns [`EngineError::Partner`] on storage failure.
467 async fn as4_endpoint(
468 &self,
469 tenant_id: TenantId,
470 mp_id: &MarktpartnerCode,
471 ) -> Result<Option<Box<str>>, EngineError> {
472 Ok(self
473 .get(tenant_id, mp_id)
474 .await?
475 .and_then(|r| r.as4_endpoint().map(std::convert::Into::into)))
476 }
477
478 /// Return the API-Webdienste Strom base URL for `gln`, if known.
479 ///
480 /// Looks for a channel with qualifier `"AW"` (populated by the
481 /// Verzeichnisdienst discovery path.
482 ///
483 /// Convenience wrapper over `get` + `api_webdienste_endpoint`.
484 ///
485 /// # Errors
486 ///
487 /// Returns [`EngineError::Partner`] on storage failure.
488 async fn api_webdienste_endpoint(
489 &self,
490 tenant_id: TenantId,
491 mp_id: &MarktpartnerCode,
492 ) -> Result<Option<Box<str>>, EngineError> {
493 Ok(self
494 .get(tenant_id, mp_id)
495 .await?
496 .and_then(|r| r.api_webdienste_endpoint().map(std::convert::Into::into)))
497 }
498}
499
500// ── Arc<S> blanket impl ───────────────────────────────────────────────────────
501
502impl<S: PartnerStore> PartnerStore for Arc<S> {
503 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
504 self.as_ref().upsert(tenant_id, record).await
505 }
506
507 async fn get(
508 &self,
509 tenant_id: TenantId,
510 mp_id: &MarktpartnerCode,
511 ) -> Result<Option<PartnerRecord>, EngineError> {
512 self.as_ref().get(tenant_id, mp_id).await
513 }
514
515 async fn remove(
516 &self,
517 tenant_id: TenantId,
518 mp_id: &MarktpartnerCode,
519 ) -> Result<(), EngineError> {
520 self.as_ref().remove(tenant_id, mp_id).await
521 }
522
523 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
524 self.as_ref().list(tenant_id).await
525 }
526}
527
528// ── NoopPartnerStore ──────────────────────────────────────────────────────────
529
530/// A [`PartnerStore`] that never persists anything.
531///
532/// Every `get` returns `None`. Use as the default in deployments that rely
533/// exclusively on static config-based partner lookup (i.e. when
534/// `PartnerDirectory::from_cli_pairs` is sufficient).
535///
536/// ⚠️ **Data loss**: All upserts are silently discarded. PARTIN-derived
537/// updates received at runtime will not be retained across restarts.
538#[derive(Debug, Clone, Copy, Default)]
539pub struct NoopPartnerStore;
540
541impl PartnerStore for NoopPartnerStore {
542 async fn upsert(
543 &self,
544 _tenant_id: TenantId,
545 _record: &PartnerRecord,
546 ) -> Result<(), EngineError> {
547 Ok(())
548 }
549
550 async fn get(
551 &self,
552 _tenant_id: TenantId,
553 _mp_id: &MarktpartnerCode,
554 ) -> Result<Option<PartnerRecord>, EngineError> {
555 Ok(None)
556 }
557
558 async fn remove(
559 &self,
560 _tenant_id: TenantId,
561 _mp_id: &MarktpartnerCode,
562 ) -> Result<(), EngineError> {
563 Ok(())
564 }
565
566 async fn list(&self, _tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
567 Ok(vec![])
568 }
569}
570
571// ── InMemoryPartnerStore ──────────────────────────────────────────────────────
572
573/// An in-memory [`PartnerStore`] for tests and development.
574///
575/// Backed by a `HashMap<(TenantId, MarktpartnerCode), PartnerRecord>` protected by an
576/// `Arc<RwLock<…>>`. Clones share the underlying data — all clones see the
577/// same records. Upsert calls `merge_from_partin` for existing records.
578///
579/// Only available in `#[cfg(test)]` or with the `testing` feature enabled.
580#[cfg(any(test, feature = "testing"))]
581#[derive(Debug, Clone, Default)]
582pub struct InMemoryPartnerStore {
583 inner: Arc<RwLock<HashMap<(TenantId, MarktpartnerCode), PartnerRecord>>>,
584}
585
586#[cfg(any(test, feature = "testing"))]
587impl InMemoryPartnerStore {
588 /// Create a new empty store.
589 #[must_use]
590 pub fn new() -> Self {
591 Self::default()
592 }
593}
594
595#[cfg(any(test, feature = "testing"))]
596impl PartnerStore for InMemoryPartnerStore {
597 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
598 let mut guard = self.inner.write().await;
599 let key = (tenant_id, record.mp_id.clone());
600 match guard.get_mut(&key) {
601 Some(existing) => existing.merge_from_partin(record.clone()),
602 None => {
603 guard.insert(key, record.clone());
604 }
605 }
606 Ok(())
607 }
608
609 async fn get(
610 &self,
611 tenant_id: TenantId,
612 mp_id: &MarktpartnerCode,
613 ) -> Result<Option<PartnerRecord>, EngineError> {
614 Ok(self
615 .inner
616 .read()
617 .await
618 .get(&(tenant_id, mp_id.clone()))
619 .cloned())
620 }
621
622 async fn remove(
623 &self,
624 tenant_id: TenantId,
625 mp_id: &MarktpartnerCode,
626 ) -> Result<(), EngineError> {
627 self.inner.write().await.remove(&(tenant_id, mp_id.clone()));
628 Ok(())
629 }
630
631 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
632 Ok(self
633 .inner
634 .read()
635 .await
636 .iter()
637 .filter(|((tid, _), _)| *tid == tenant_id)
638 .map(|(_, record)| record.clone())
639 .collect())
640 }
641}
642
643// ── Tests ─────────────────────────────────────────────────────────────────────
644
645#[cfg(test)]
646mod tests {
647 use super::*;
648
649 fn mp_id(s: &str) -> MarktpartnerCode {
650 MarktpartnerCode::new(s)
651 }
652 fn tid() -> TenantId {
653 TenantId::new()
654 }
655
656 fn minimal_record(gln_str: &str, url: &str) -> PartnerRecord {
657 PartnerRecord::minimal(mp_id(gln_str), url)
658 }
659
660 // ── from_cli_pairs ────────────────────────────────────────────────────────
661
662 #[test]
663 fn from_cli_pairs_parses_valid_entries() {
664 let pairs = vec![
665 "9900000000002=https://partner-a.example/as4/inbox",
666 "9900000000003=https://partner-b.example/as4/inbox",
667 ];
668 let records = PartnerRecord::from_cli_pairs(&pairs).unwrap();
669 assert_eq!(records.len(), 2);
670 assert_eq!(records[0].mp_id.as_str(), "9900000000002");
671 assert_eq!(
672 records[0].as4_endpoint(),
673 Some("https://partner-a.example/as4/inbox")
674 );
675 assert_eq!(records[1].mp_id.as_str(), "9900000000003");
676 }
677
678 #[test]
679 fn from_cli_pairs_rejects_missing_equals() {
680 let pairs = vec!["9900000000002https://no-equals.example"];
681 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
682 }
683
684 #[test]
685 fn from_cli_pairs_rejects_http_url() {
686 let pairs = vec!["9900000000002=http://insecure.example/as4"];
687 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
688 }
689
690 #[test]
691 fn from_cli_pairs_rejects_empty_gln() {
692 let pairs = vec!["=https://no-mp_id.example/as4"];
693 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
694 }
695
696 // ── as4_endpoint ──────────────────────────────────────────────────────────
697
698 #[test]
699 fn as4_endpoint_returns_ak_channel() {
700 let r = minimal_record("9900000000002", "https://a.example/as4");
701 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
702 }
703
704 #[test]
705 fn as4_endpoint_returns_none_when_absent() {
706 let r = PartnerRecord {
707 mp_id: mp_id("9900000000002"),
708 display_name: None,
709 channels: vec![CommunicationChannel::email("info@example.de")],
710 roles: vec![],
711 valid_from: None,
712 contacts: vec![],
713 country_code: None,
714 updated_at: OffsetDateTime::now_utc(),
715 };
716 assert!(r.as4_endpoint().is_none());
717 }
718
719 // ── merge_from_partin ─────────────────────────────────────────────────────
720
721 #[test]
722 fn merge_overwrites_config_record_with_partin_data() {
723 let mut base = minimal_record("9900000000002", "https://old.example/as4");
724 let newer = PartnerRecord {
725 mp_id: mp_id("9900000000002"),
726 display_name: Some("Stadtwerke AG".into()),
727 channels: vec![
728 CommunicationChannel::as4("https://new.example/as4"),
729 CommunicationChannel::email("edifact@sw.example"),
730 ],
731 roles: vec![MarketRole::NbStrom],
732 valid_from: Some(OffsetDateTime::now_utc()),
733 contacts: vec![],
734 country_code: Some("DE".into()),
735 updated_at: OffsetDateTime::now_utc(),
736 };
737 base.merge_from_partin(newer.clone());
738 assert_eq!(base.as4_endpoint(), Some("https://new.example/as4"));
739 assert_eq!(base.display_name.as_deref(), Some("Stadtwerke AG"));
740 assert_eq!(base.roles, vec![MarketRole::NbStrom]);
741 }
742
743 #[test]
744 fn merge_ignores_older_partin() {
745 use time::Duration;
746 let old_ts = OffsetDateTime::now_utc() - Duration::days(30);
747 let new_ts = OffsetDateTime::now_utc();
748
749 let mut current = PartnerRecord {
750 mp_id: mp_id("9900000000002"),
751 display_name: Some("Current Name".into()),
752 channels: vec![CommunicationChannel::as4("https://current.example/as4")],
753 roles: vec![MarketRole::NbStrom],
754 valid_from: Some(new_ts),
755 contacts: vec![],
756 country_code: Some("DE".into()),
757 updated_at: OffsetDateTime::now_utc(),
758 };
759
760 let stale = PartnerRecord {
761 mp_id: mp_id("9900000000002"),
762 display_name: Some("Stale Name".into()),
763 channels: vec![CommunicationChannel::as4("https://stale.example/as4")],
764 roles: vec![],
765 valid_from: Some(old_ts),
766 contacts: vec![],
767 country_code: None,
768 updated_at: OffsetDateTime::now_utc(),
769 };
770
771 current.merge_from_partin(stale);
772 // Should not be overwritten
773 assert_eq!(current.display_name.as_deref(), Some("Current Name"));
774 assert_eq!(current.as4_endpoint(), Some("https://current.example/as4"));
775 }
776
777 #[test]
778 fn merge_ignores_wrong_gln() {
779 let mut r = minimal_record("9900000000002", "https://a.example/as4");
780 let other = minimal_record("9900000000003", "https://b.example/as4");
781 r.merge_from_partin(other);
782 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
783 }
784
785 // ── MarketRole::from_pid ──────────────────────────────────────────────────
786
787 #[test]
788 fn market_role_from_pid_covers_all_partin_pids() {
789 for pid in [
790 37000u32, 37001, 37002, 37003, 37004, 37005, 37006, 37008, 37009, 37010, 37011, 37012,
791 37013, 37014,
792 ] {
793 assert!(
794 MarketRole::from_pid(pid).is_some(),
795 "MarketRole::from_pid({pid}) should return Some"
796 );
797 }
798 // PID 37007 is not in the AHB (gap)
799 assert!(MarketRole::from_pid(37007).is_none());
800 assert!(MarketRole::from_pid(0).is_none());
801 }
802
803 // ── InMemoryPartnerStore ──────────────────────────────────────────────────
804
805 #[tokio::test]
806 async fn in_memory_upsert_and_get() {
807 let store = InMemoryPartnerStore::new();
808 let tenant = tid();
809 let record = minimal_record("9900000000001", "https://a.example/as4");
810
811 store.upsert(tenant, &record).await.unwrap();
812 let found = store
813 .get(tenant, &mp_id("9900000000001"))
814 .await
815 .unwrap()
816 .unwrap();
817 assert_eq!(found.as4_endpoint(), Some("https://a.example/as4"));
818 }
819
820 #[tokio::test]
821 async fn in_memory_get_returns_none_for_unknown() {
822 let store = InMemoryPartnerStore::new();
823 assert!(
824 store
825 .get(tid(), &mp_id("9900000000099"))
826 .await
827 .unwrap()
828 .is_none()
829 );
830 }
831
832 #[tokio::test]
833 async fn in_memory_upsert_merges_into_existing() {
834 let store = InMemoryPartnerStore::new();
835 let tenant = tid();
836 let base = minimal_record("9900000000001", "https://old.example/as4");
837 store.upsert(tenant, &base).await.unwrap();
838
839 let newer = PartnerRecord {
840 mp_id: mp_id("9900000000001"),
841 display_name: Some("Partner AG".into()),
842 channels: vec![CommunicationChannel::as4("https://new.example/as4")],
843 roles: vec![MarketRole::LfStrom],
844 valid_from: Some(OffsetDateTime::now_utc()),
845 contacts: vec![],
846 country_code: Some("DE".into()),
847 updated_at: OffsetDateTime::now_utc(),
848 };
849 store.upsert(tenant, &newer).await.unwrap();
850
851 let found = store
852 .get(tenant, &mp_id("9900000000001"))
853 .await
854 .unwrap()
855 .unwrap();
856 assert_eq!(found.as4_endpoint(), Some("https://new.example/as4"));
857 assert_eq!(found.display_name.as_deref(), Some("Partner AG"));
858 }
859
860 #[tokio::test]
861 async fn in_memory_remove_clears_record() {
862 let store = InMemoryPartnerStore::new();
863 let tenant = tid();
864 let record = minimal_record("9900000000001", "https://a.example/as4");
865
866 store.upsert(tenant, &record).await.unwrap();
867 store.remove(tenant, &mp_id("9900000000001")).await.unwrap();
868 assert!(
869 store
870 .get(tenant, &mp_id("9900000000001"))
871 .await
872 .unwrap()
873 .is_none()
874 );
875 }
876
877 #[tokio::test]
878 async fn in_memory_list_is_tenant_scoped() {
879 let store = InMemoryPartnerStore::new();
880 let t1 = tid();
881 let t2 = tid();
882
883 store
884 .upsert(
885 t1,
886 &minimal_record("9900000000001", "https://a.example/as4"),
887 )
888 .await
889 .unwrap();
890 store
891 .upsert(
892 t2,
893 &minimal_record("9900000000002", "https://b.example/as4"),
894 )
895 .await
896 .unwrap();
897
898 let t1_list = store.list(t1).await.unwrap();
899 assert_eq!(t1_list.len(), 1);
900 assert_eq!(t1_list[0].mp_id.as_str(), "9900000000001");
901
902 let t2_list = store.list(t2).await.unwrap();
903 assert_eq!(t2_list.len(), 1);
904 assert_eq!(t2_list[0].mp_id.as_str(), "9900000000002");
905 }
906
907 #[tokio::test]
908 async fn as4_endpoint_convenience_method() {
909 let store = InMemoryPartnerStore::new();
910 let tenant = tid();
911 let record = minimal_record("9900000000001", "https://a.example/as4");
912
913 store.upsert(tenant, &record).await.unwrap();
914 let url = store
915 .as4_endpoint(tenant, &mp_id("9900000000001"))
916 .await
917 .unwrap();
918 assert_eq!(url.as_deref(), Some("https://a.example/as4"));
919
920 let none = store
921 .as4_endpoint(tenant, &mp_id("9900000000099"))
922 .await
923 .unwrap();
924 assert!(none.is_none());
925 }
926}