Skip to main content

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#[cfg_attr(
539    not(any(test, feature = "testing")),
540    deprecated = "NoopPartnerStore must not be instantiated in production builds; \
541                  PARTIN-derived partner updates will be silently discarded. \
542                  Use SlateDbPartnerStore or another durable PartnerStore instead."
543)]
544#[derive(Debug, Clone, Copy, Default)]
545pub struct NoopPartnerStore;
546
547// The `#[allow(deprecated)]` is required because the `deprecated` attribute on
548// `NoopPartnerStore` fires on the impl block inside the same file. This is a
549// known Rust quirk (implementing a deprecated type fires the lint even in the
550// defining module). The guard is still effective: *callers* that instantiate
551// `NoopPartnerStore` outside of test/feature-gated code will see the warning.
552#[cfg(any(test, feature = "testing"))]
553#[allow(deprecated)]
554impl PartnerStore for NoopPartnerStore {
555    async fn upsert(
556        &self,
557        _tenant_id: TenantId,
558        _record: &PartnerRecord,
559    ) -> Result<(), EngineError> {
560        Ok(())
561    }
562
563    async fn get(
564        &self,
565        _tenant_id: TenantId,
566        _mp_id: &MarktpartnerCode,
567    ) -> Result<Option<PartnerRecord>, EngineError> {
568        Ok(None)
569    }
570
571    async fn remove(
572        &self,
573        _tenant_id: TenantId,
574        _mp_id: &MarktpartnerCode,
575    ) -> Result<(), EngineError> {
576        Ok(())
577    }
578
579    async fn list(&self, _tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
580        Ok(vec![])
581    }
582}
583
584// ── InMemoryPartnerStore ──────────────────────────────────────────────────────
585
586/// An in-memory [`PartnerStore`] for tests and development.
587///
588/// Backed by a `HashMap<(TenantId, MarktpartnerCode), PartnerRecord>` protected by an
589/// `Arc<RwLock<…>>`. Clones share the underlying data — all clones see the
590/// same records. Upsert calls `merge_from_partin` for existing records.
591///
592/// Only available in `#[cfg(test)]` or with the `testing` feature enabled.
593#[cfg(any(test, feature = "testing"))]
594#[derive(Debug, Clone, Default)]
595pub struct InMemoryPartnerStore {
596    inner: Arc<RwLock<HashMap<(TenantId, MarktpartnerCode), PartnerRecord>>>,
597}
598
599#[cfg(any(test, feature = "testing"))]
600impl InMemoryPartnerStore {
601    /// Create a new empty store.
602    #[must_use]
603    pub fn new() -> Self {
604        Self::default()
605    }
606}
607
608#[cfg(any(test, feature = "testing"))]
609impl PartnerStore for InMemoryPartnerStore {
610    async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
611        let mut guard = self.inner.write().await;
612        let key = (tenant_id, record.mp_id.clone());
613        match guard.get_mut(&key) {
614            Some(existing) => existing.merge_from_partin(record.clone()),
615            None => {
616                guard.insert(key, record.clone());
617            }
618        }
619        Ok(())
620    }
621
622    async fn get(
623        &self,
624        tenant_id: TenantId,
625        mp_id: &MarktpartnerCode,
626    ) -> Result<Option<PartnerRecord>, EngineError> {
627        Ok(self
628            .inner
629            .read()
630            .await
631            .get(&(tenant_id, mp_id.clone()))
632            .cloned())
633    }
634
635    async fn remove(
636        &self,
637        tenant_id: TenantId,
638        mp_id: &MarktpartnerCode,
639    ) -> Result<(), EngineError> {
640        self.inner.write().await.remove(&(tenant_id, mp_id.clone()));
641        Ok(())
642    }
643
644    async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
645        Ok(self
646            .inner
647            .read()
648            .await
649            .iter()
650            .filter(|((tid, _), _)| *tid == tenant_id)
651            .map(|(_, record)| record.clone())
652            .collect())
653    }
654}
655
656// ── Tests ─────────────────────────────────────────────────────────────────────
657
658#[cfg(test)]
659mod tests {
660    use super::*;
661
662    fn mp_id(s: &str) -> MarktpartnerCode {
663        MarktpartnerCode::new(s)
664    }
665    fn tid() -> TenantId {
666        TenantId::new()
667    }
668
669    fn minimal_record(gln_str: &str, url: &str) -> PartnerRecord {
670        PartnerRecord::minimal(mp_id(gln_str), url)
671    }
672
673    // ── from_cli_pairs ────────────────────────────────────────────────────────
674
675    #[test]
676    fn from_cli_pairs_parses_valid_entries() {
677        let pairs = vec![
678            "9900000000002=https://partner-a.example/as4/inbox",
679            "9900000000003=https://partner-b.example/as4/inbox",
680        ];
681        let records = PartnerRecord::from_cli_pairs(&pairs).unwrap();
682        assert_eq!(records.len(), 2);
683        assert_eq!(records[0].mp_id.as_str(), "9900000000002");
684        assert_eq!(
685            records[0].as4_endpoint(),
686            Some("https://partner-a.example/as4/inbox")
687        );
688        assert_eq!(records[1].mp_id.as_str(), "9900000000003");
689    }
690
691    #[test]
692    fn from_cli_pairs_rejects_missing_equals() {
693        let pairs = vec!["9900000000002https://no-equals.example"];
694        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
695    }
696
697    #[test]
698    fn from_cli_pairs_rejects_http_url() {
699        let pairs = vec!["9900000000002=http://insecure.example/as4"];
700        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
701    }
702
703    #[test]
704    fn from_cli_pairs_rejects_empty_gln() {
705        let pairs = vec!["=https://no-mp_id.example/as4"];
706        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
707    }
708
709    // ── as4_endpoint ──────────────────────────────────────────────────────────
710
711    #[test]
712    fn as4_endpoint_returns_ak_channel() {
713        let r = minimal_record("9900000000002", "https://a.example/as4");
714        assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
715    }
716
717    #[test]
718    fn as4_endpoint_returns_none_when_absent() {
719        let r = PartnerRecord {
720            mp_id: mp_id("9900000000002"),
721            display_name: None,
722            channels: vec![CommunicationChannel::email("info@example.de")],
723            roles: vec![],
724            valid_from: None,
725            contacts: vec![],
726            country_code: None,
727            updated_at: OffsetDateTime::now_utc(),
728        };
729        assert!(r.as4_endpoint().is_none());
730    }
731
732    // ── merge_from_partin ─────────────────────────────────────────────────────
733
734    #[test]
735    fn merge_overwrites_config_record_with_partin_data() {
736        let mut base = minimal_record("9900000000002", "https://old.example/as4");
737        let newer = PartnerRecord {
738            mp_id: mp_id("9900000000002"),
739            display_name: Some("Stadtwerke AG".into()),
740            channels: vec![
741                CommunicationChannel::as4("https://new.example/as4"),
742                CommunicationChannel::email("edifact@sw.example"),
743            ],
744            roles: vec![MarketRole::NbStrom],
745            valid_from: Some(OffsetDateTime::now_utc()),
746            contacts: vec![],
747            country_code: Some("DE".into()),
748            updated_at: OffsetDateTime::now_utc(),
749        };
750        base.merge_from_partin(newer.clone());
751        assert_eq!(base.as4_endpoint(), Some("https://new.example/as4"));
752        assert_eq!(base.display_name.as_deref(), Some("Stadtwerke AG"));
753        assert_eq!(base.roles, vec![MarketRole::NbStrom]);
754    }
755
756    #[test]
757    fn merge_ignores_older_partin() {
758        use time::Duration;
759        let old_ts = OffsetDateTime::now_utc() - Duration::days(30);
760        let new_ts = OffsetDateTime::now_utc();
761
762        let mut current = PartnerRecord {
763            mp_id: mp_id("9900000000002"),
764            display_name: Some("Current Name".into()),
765            channels: vec![CommunicationChannel::as4("https://current.example/as4")],
766            roles: vec![MarketRole::NbStrom],
767            valid_from: Some(new_ts),
768            contacts: vec![],
769            country_code: Some("DE".into()),
770            updated_at: OffsetDateTime::now_utc(),
771        };
772
773        let stale = PartnerRecord {
774            mp_id: mp_id("9900000000002"),
775            display_name: Some("Stale Name".into()),
776            channels: vec![CommunicationChannel::as4("https://stale.example/as4")],
777            roles: vec![],
778            valid_from: Some(old_ts),
779            contacts: vec![],
780            country_code: None,
781            updated_at: OffsetDateTime::now_utc(),
782        };
783
784        current.merge_from_partin(stale);
785        // Should not be overwritten
786        assert_eq!(current.display_name.as_deref(), Some("Current Name"));
787        assert_eq!(current.as4_endpoint(), Some("https://current.example/as4"));
788    }
789
790    #[test]
791    fn merge_ignores_wrong_gln() {
792        let mut r = minimal_record("9900000000002", "https://a.example/as4");
793        let other = minimal_record("9900000000003", "https://b.example/as4");
794        r.merge_from_partin(other);
795        assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
796    }
797
798    // ── MarketRole::from_pid ──────────────────────────────────────────────────
799
800    #[test]
801    fn market_role_from_pid_covers_all_partin_pids() {
802        for pid in [
803            37000u32, 37001, 37002, 37003, 37004, 37005, 37006, 37008, 37009, 37010, 37011, 37012,
804            37013, 37014,
805        ] {
806            assert!(
807                MarketRole::from_pid(pid).is_some(),
808                "MarketRole::from_pid({pid}) should return Some"
809            );
810        }
811        // PID 37007 is not in the AHB (gap)
812        assert!(MarketRole::from_pid(37007).is_none());
813        assert!(MarketRole::from_pid(0).is_none());
814    }
815
816    // ── InMemoryPartnerStore ──────────────────────────────────────────────────
817
818    #[tokio::test]
819    async fn in_memory_upsert_and_get() {
820        let store = InMemoryPartnerStore::new();
821        let tenant = tid();
822        let record = minimal_record("9900000000001", "https://a.example/as4");
823
824        store.upsert(tenant, &record).await.unwrap();
825        let found = store
826            .get(tenant, &mp_id("9900000000001"))
827            .await
828            .unwrap()
829            .unwrap();
830        assert_eq!(found.as4_endpoint(), Some("https://a.example/as4"));
831    }
832
833    #[tokio::test]
834    async fn in_memory_get_returns_none_for_unknown() {
835        let store = InMemoryPartnerStore::new();
836        assert!(
837            store
838                .get(tid(), &mp_id("9900000000099"))
839                .await
840                .unwrap()
841                .is_none()
842        );
843    }
844
845    #[tokio::test]
846    async fn in_memory_upsert_merges_into_existing() {
847        let store = InMemoryPartnerStore::new();
848        let tenant = tid();
849        let base = minimal_record("9900000000001", "https://old.example/as4");
850        store.upsert(tenant, &base).await.unwrap();
851
852        let newer = PartnerRecord {
853            mp_id: mp_id("9900000000001"),
854            display_name: Some("Partner AG".into()),
855            channels: vec![CommunicationChannel::as4("https://new.example/as4")],
856            roles: vec![MarketRole::LfStrom],
857            valid_from: Some(OffsetDateTime::now_utc()),
858            contacts: vec![],
859            country_code: Some("DE".into()),
860            updated_at: OffsetDateTime::now_utc(),
861        };
862        store.upsert(tenant, &newer).await.unwrap();
863
864        let found = store
865            .get(tenant, &mp_id("9900000000001"))
866            .await
867            .unwrap()
868            .unwrap();
869        assert_eq!(found.as4_endpoint(), Some("https://new.example/as4"));
870        assert_eq!(found.display_name.as_deref(), Some("Partner AG"));
871    }
872
873    #[tokio::test]
874    async fn in_memory_remove_clears_record() {
875        let store = InMemoryPartnerStore::new();
876        let tenant = tid();
877        let record = minimal_record("9900000000001", "https://a.example/as4");
878
879        store.upsert(tenant, &record).await.unwrap();
880        store.remove(tenant, &mp_id("9900000000001")).await.unwrap();
881        assert!(
882            store
883                .get(tenant, &mp_id("9900000000001"))
884                .await
885                .unwrap()
886                .is_none()
887        );
888    }
889
890    #[tokio::test]
891    async fn in_memory_list_is_tenant_scoped() {
892        let store = InMemoryPartnerStore::new();
893        let t1 = tid();
894        let t2 = tid();
895
896        store
897            .upsert(
898                t1,
899                &minimal_record("9900000000001", "https://a.example/as4"),
900            )
901            .await
902            .unwrap();
903        store
904            .upsert(
905                t2,
906                &minimal_record("9900000000002", "https://b.example/as4"),
907            )
908            .await
909            .unwrap();
910
911        let t1_list = store.list(t1).await.unwrap();
912        assert_eq!(t1_list.len(), 1);
913        assert_eq!(t1_list[0].mp_id.as_str(), "9900000000001");
914
915        let t2_list = store.list(t2).await.unwrap();
916        assert_eq!(t2_list.len(), 1);
917        assert_eq!(t2_list[0].mp_id.as_str(), "9900000000002");
918    }
919
920    #[tokio::test]
921    async fn as4_endpoint_convenience_method() {
922        let store = InMemoryPartnerStore::new();
923        let tenant = tid();
924        let record = minimal_record("9900000000001", "https://a.example/as4");
925
926        store.upsert(tenant, &record).await.unwrap();
927        let url = store
928            .as4_endpoint(tenant, &mp_id("9900000000001"))
929            .await
930            .unwrap();
931        assert_eq!(url.as_deref(), Some("https://a.example/as4"));
932
933        let none = store
934            .as4_endpoint(tenant, &mp_id("9900000000099"))
935            .await
936            .unwrap();
937        assert!(none.is_none());
938    }
939}