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, marktrolle::Marktrolle, 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// ── PartnerRecord ─────────────────────────────────────────────────────────────
151
152/// Full trading-partner master record as stored in the [`PartnerStore`].
153///
154/// Populated either from static `makod.toml` config (minimal — GLN + AS4 URL
155/// only) or from an inbound PARTIN EDIFACT message (complete). Records from
156/// different sources coexist: a bootstrapped config record is upgraded in-place
157/// when the same partner later sends a PARTIN.
158///
159/// ## Constructors
160///
161/// - [`PartnerRecord::minimal`] — for bootstrapping from `GLN=URL` config pairs
162/// - [`PartnerRecord::from_cli_pairs`] — parse `[as4] partners` list from config
163///
164/// ## Merging
165///
166/// Use [`PartnerRecord::merge_from_partin`] to update an existing record with
167/// fields from a newer inbound PARTIN (respects validity dates).
168#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
169#[serde(deny_unknown_fields)]
170pub struct PartnerRecord {
171    /// The partner's Marktpartner-ID — a BDEW-Codenummer, a DVGW-Codenummer,
172    /// a GS1 GLN or an EIC (Allgemeine Festlegungen §2.13).
173    pub mp_id: MarktpartnerCode,
174
175    /// Company name from the PARTIN `NAD` segment.
176    #[serde(default)]
177    pub display_name: Option<Box<str>>,
178
179    /// All communication channels from PARTIN `COM` segments.
180    ///
181    /// The AS4 endpoint is the entry with qualifier `"AK"` (PARTIN AHB 1.0f
182    /// DE 3155 convention).  Use [`as4_endpoint`] for direct access.
183    ///
184    /// [`as4_endpoint`]: PartnerRecord::as4_endpoint
185    #[serde(default)]
186    pub channels: Vec<CommunicationChannel>,
187
188    /// Market roles this partner has declared via PARTIN.
189    ///
190    /// Derived from the PARTIN Prüfidentifikator via
191    /// [`Marktrolle::from_partin_pid`]. Serialises as BDEW role codes
192    /// (`"LF"`, `"NB"`, `"MSB"`, …).
193    #[serde(default)]
194    pub roles: Vec<Marktrolle>,
195
196    /// Date from which this record version is valid (`DTM/137`).
197    ///
198    /// `None` when bootstrapped from static config (no validity date known).
199    #[serde(
200        default,
201        skip_serializing_if = "Option::is_none",
202        with = "time::serde::rfc3339::option"
203    )]
204    pub valid_from: Option<OffsetDateTime>,
205
206    /// Contact persons from the PARTIN *Ansprechpartner* group.
207    #[serde(default)]
208    pub contacts: Vec<ContactPerson>,
209
210    /// ISO 3166-1 alpha-2 country code from `NAD+MS+++...+DE` (usually `DE`).
211    #[serde(default)]
212    pub country_code: Option<Box<str>>,
213
214    /// Wall-clock time when this record was last written to the store.
215    ///
216    /// Server-owned. It defaults on deserialisation because a client has no
217    /// business asserting when *we* last wrote a record — and because
218    /// [`merge_from_partin`] carries it forward, a caller who could set it
219    /// would be writing into a field the merge reads.
220    ///
221    /// [`merge_from_partin`]: PartnerRecord::merge_from_partin
222    #[serde(default = "OffsetDateTime::now_utc", with = "time::serde::rfc3339")]
223    pub updated_at: OffsetDateTime,
224}
225
226impl PartnerRecord {
227    /// Create a minimal record from a GLN and an AS4 endpoint URL.
228    ///
229    /// Used when bootstrapping from `[as4] partners = ["GLN=URL", …]` in
230    /// `makod.toml`. The record has no PARTIN-derived metadata — only the
231    /// GLN and a single AS4 channel.
232    #[must_use]
233    pub fn minimal(mp_id: impl Into<MarktpartnerCode>, as4_url: impl Into<Box<str>>) -> Self {
234        Self {
235            mp_id: mp_id.into(),
236            display_name: None,
237            channels: vec![CommunicationChannel::as4(as4_url)],
238            roles: Vec::new(),
239            valid_from: None,
240            contacts: Vec::new(),
241            country_code: None,
242            updated_at: OffsetDateTime::now_utc(),
243        }
244    }
245
246    /// Parse `["GLN=HTTPS-URL", …]` configuration entries into minimal records.
247    ///
248    /// Returns an error on the first malformed or non-HTTPS entry.
249    ///
250    /// # Errors
251    ///
252    /// Returns [`EngineError::Partner`] when an entry lacks `=`, has an empty
253    /// GLN, or uses a non-HTTPS URL.
254    pub fn from_cli_pairs(pairs: &[impl AsRef<str>]) -> Result<Vec<Self>, EngineError> {
255        pairs
256            .iter()
257            .map(|entry| {
258                let pair = entry.as_ref();
259                let (mp_id, url) = pair.split_once('=').ok_or_else(|| {
260                    EngineError::partner(format!(
261                        "invalid partner entry {pair:?} — expected <GLN>=<HTTPS-URL>"
262                    ))
263                })?;
264                let mp_id = mp_id.trim();
265                let url = url.trim();
266                if mp_id.is_empty() {
267                    return Err(EngineError::partner(format!(
268                        "invalid partner entry {pair:?} — GLN must not be empty"
269                    )));
270                }
271                if !url.starts_with("https://") {
272                    return Err(EngineError::partner(format!(
273                        "invalid partner entry {pair:?} — endpoint URL must use HTTPS (got {url:?})"
274                    )));
275                }
276                Ok(Self::minimal(mp_id, url))
277            })
278            .collect()
279    }
280
281    /// Return the AS4 endpoint URL if one has been registered.
282    ///
283    /// Looks for a channel with qualifier `"AK"` (PARTIN AHB 1.0f
284    /// convention for the AS4 endpoint). Falls back to `"AS4"` for records
285    /// that were imported with a non-standard qualifier.
286    #[must_use]
287    pub fn as4_endpoint(&self) -> Option<&str> {
288        self.channels
289            .iter()
290            .find(|c| c.qualifier.as_ref() == "AK" || c.qualifier.as_ref() == "AS4")
291            .map(|c| c.address.as_ref())
292    }
293
294    /// Return the primary email address if one has been registered.
295    ///
296    /// Looks for a channel with qualifier `"EM"`.
297    #[must_use]
298    pub fn email(&self) -> Option<&str> {
299        self.channels
300            .iter()
301            .find(|c| c.qualifier.as_ref() == "EM")
302            .map(|c| c.address.as_ref())
303    }
304
305    /// Return the API-Webdienste Strom base URL if one has been registered.
306    ///
307    /// Looks for a channel with qualifier `"AW"`.  This URL is typically
308    /// populated by the Verzeichnisdienst discovery worker and is
309    /// used by `MaloIdentSender` to reach the LF's callback endpoint.
310    #[must_use]
311    pub fn api_webdienste_endpoint(&self) -> Option<&str> {
312        self.channels
313            .iter()
314            .find(|c| c.qualifier.as_ref() == "AW")
315            .map(|c| c.address.as_ref())
316    }
317
318    /// Merge fields from a newer PARTIN-derived record into `self`.
319    ///
320    /// Only updates `self` when `incoming.valid_from` is newer than
321    /// `self.valid_from` (or when `self.valid_from` is `None`). Config-
322    /// bootstrapped records (no `valid_from`) are always overwritten.
323    ///
324    /// The GLN must match — mismatches are silently ignored (the caller is
325    /// responsible for routing PARTIN messages to the correct record).
326    pub fn merge_from_partin(&mut self, incoming: PartnerRecord) {
327        if incoming.mp_id != self.mp_id {
328            return;
329        }
330        let should_update = match (self.valid_from, incoming.valid_from) {
331            (None, _) => true,
332            (Some(_), None) => false, // keep the dated record
333            (Some(a), Some(b)) => b >= a,
334        };
335        if !should_update {
336            return;
337        }
338        self.display_name = incoming.display_name.or(self.display_name.take());
339        self.channels = incoming.channels;
340        self.roles = incoming.roles;
341        self.valid_from = incoming.valid_from;
342        self.contacts = incoming.contacts;
343        self.country_code = incoming.country_code.or(self.country_code.take());
344        self.updated_at = incoming.updated_at;
345    }
346}
347
348// ── PartnerStore ──────────────────────────────────────────────────────────────
349
350/// Durable store for trading-partner master records.
351///
352/// Provides tenant-scoped access to [`PartnerRecord`]s. Records are upserted
353/// when a new PARTIN message arrives or when `makod` bootstraps from static
354/// config.
355///
356/// All three operations are idempotent — reinserting the same record is safe.
357///
358/// ## Blanket `Arc` implementation
359///
360/// `Arc<S>` implements `PartnerStore` whenever `S: PartnerStore`.
361#[allow(async_fn_in_trait)]
362pub trait PartnerStore: Send + Sync {
363    /// Insert or update the record for `(tenant_id, record.mp_id)`.
364    ///
365    /// If a record already exists for this GLN, it is **merged** via
366    /// [`PartnerRecord::merge_from_partin`] — i.e. the newer PARTIN-derived
367    /// record wins, but a config-only bootstrap is always overwritten.
368    ///
369    /// # Errors
370    ///
371    /// Returns [`EngineError::Partner`] on storage failure.
372    async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError>;
373
374    /// Return the record for `(tenant_id, gln)`, or `None` if not registered.
375    ///
376    /// # Errors
377    ///
378    /// Returns [`EngineError::Partner`] on storage failure.
379    async fn get(
380        &self,
381        tenant_id: TenantId,
382        mp_id: &MarktpartnerCode,
383    ) -> Result<Option<PartnerRecord>, EngineError>;
384
385    /// Remove the record for `(tenant_id, gln)`.
386    ///
387    /// No-op when the record does not exist.
388    ///
389    /// # Errors
390    ///
391    /// Returns [`EngineError::Partner`] on storage failure.
392    async fn remove(
393        &self,
394        tenant_id: TenantId,
395        mp_id: &MarktpartnerCode,
396    ) -> Result<(), EngineError>;
397
398    /// Return all records registered for `tenant_id`.
399    ///
400    /// # Errors
401    ///
402    /// Returns [`EngineError::Partner`] on storage failure.
403    async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError>;
404
405    /// Return the AS4 endpoint URL for `gln`, if known.
406    ///
407    /// Convenience wrapper over `get` + `as4_endpoint`.
408    ///
409    /// # Errors
410    ///
411    /// Returns [`EngineError::Partner`] on storage failure.
412    async fn as4_endpoint(
413        &self,
414        tenant_id: TenantId,
415        mp_id: &MarktpartnerCode,
416    ) -> Result<Option<Box<str>>, EngineError> {
417        Ok(self
418            .get(tenant_id, mp_id)
419            .await?
420            .and_then(|r| r.as4_endpoint().map(std::convert::Into::into)))
421    }
422
423    /// Return the API-Webdienste Strom base URL for `gln`, if known.
424    ///
425    /// Looks for a channel with qualifier `"AW"` (populated by the
426    /// Verzeichnisdienst discovery path.
427    ///
428    /// Convenience wrapper over `get` + `api_webdienste_endpoint`.
429    ///
430    /// # Errors
431    ///
432    /// Returns [`EngineError::Partner`] on storage failure.
433    async fn api_webdienste_endpoint(
434        &self,
435        tenant_id: TenantId,
436        mp_id: &MarktpartnerCode,
437    ) -> Result<Option<Box<str>>, EngineError> {
438        Ok(self
439            .get(tenant_id, mp_id)
440            .await?
441            .and_then(|r| r.api_webdienste_endpoint().map(std::convert::Into::into)))
442    }
443}
444
445// ── Arc<S> blanket impl ───────────────────────────────────────────────────────
446
447impl<S: PartnerStore> PartnerStore for Arc<S> {
448    async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
449        self.as_ref().upsert(tenant_id, record).await
450    }
451
452    async fn get(
453        &self,
454        tenant_id: TenantId,
455        mp_id: &MarktpartnerCode,
456    ) -> Result<Option<PartnerRecord>, EngineError> {
457        self.as_ref().get(tenant_id, mp_id).await
458    }
459
460    async fn remove(
461        &self,
462        tenant_id: TenantId,
463        mp_id: &MarktpartnerCode,
464    ) -> Result<(), EngineError> {
465        self.as_ref().remove(tenant_id, mp_id).await
466    }
467
468    async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
469        self.as_ref().list(tenant_id).await
470    }
471}
472
473// ── NoopPartnerStore ──────────────────────────────────────────────────────────
474
475/// A [`PartnerStore`] that never persists anything.
476///
477/// Every `get` returns `None`. Use as the default in deployments that rely
478/// exclusively on static config-based partner lookup (i.e. when
479/// `PartnerDirectory::from_cli_pairs` is sufficient).
480///
481/// ⚠️ **Data loss**: All upserts are silently discarded. PARTIN-derived
482/// updates received at runtime will not be retained across restarts.
483#[cfg_attr(
484    not(any(test, feature = "testing")),
485    deprecated = "NoopPartnerStore must not be instantiated in production builds; \
486                  PARTIN-derived partner updates will be silently discarded. \
487                  Use SlateDbPartnerStore or another durable PartnerStore instead."
488)]
489#[derive(Debug, Clone, Copy, Default)]
490pub struct NoopPartnerStore;
491
492// The `#[allow(deprecated)]` is required because the `deprecated` attribute on
493// `NoopPartnerStore` fires on the impl block inside the same file. This is a
494// known Rust quirk (implementing a deprecated type fires the lint even in the
495// defining module). The guard is still effective: *callers* that instantiate
496// `NoopPartnerStore` outside of test/feature-gated code will see the warning.
497#[cfg(any(test, feature = "testing"))]
498#[allow(deprecated)]
499impl PartnerStore for NoopPartnerStore {
500    async fn upsert(
501        &self,
502        _tenant_id: TenantId,
503        _record: &PartnerRecord,
504    ) -> Result<(), EngineError> {
505        Ok(())
506    }
507
508    async fn get(
509        &self,
510        _tenant_id: TenantId,
511        _mp_id: &MarktpartnerCode,
512    ) -> Result<Option<PartnerRecord>, EngineError> {
513        Ok(None)
514    }
515
516    async fn remove(
517        &self,
518        _tenant_id: TenantId,
519        _mp_id: &MarktpartnerCode,
520    ) -> Result<(), EngineError> {
521        Ok(())
522    }
523
524    async fn list(&self, _tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
525        Ok(vec![])
526    }
527}
528
529// ── InMemoryPartnerStore ──────────────────────────────────────────────────────
530
531/// An in-memory [`PartnerStore`] for tests and development.
532///
533/// Backed by a `HashMap<(TenantId, MarktpartnerCode), PartnerRecord>` protected by an
534/// `Arc<RwLock<…>>`. Clones share the underlying data — all clones see the
535/// same records. Upsert calls `merge_from_partin` for existing records.
536///
537/// Only available in `#[cfg(test)]` or with the `testing` feature enabled.
538#[cfg(any(test, feature = "testing"))]
539#[derive(Debug, Clone, Default)]
540pub struct InMemoryPartnerStore {
541    inner: Arc<RwLock<HashMap<(TenantId, MarktpartnerCode), PartnerRecord>>>,
542}
543
544#[cfg(any(test, feature = "testing"))]
545impl InMemoryPartnerStore {
546    /// Create a new empty store.
547    #[must_use]
548    pub fn new() -> Self {
549        Self::default()
550    }
551}
552
553#[cfg(any(test, feature = "testing"))]
554impl PartnerStore for InMemoryPartnerStore {
555    async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
556        let mut guard = self.inner.write().await;
557        let key = (tenant_id, record.mp_id.clone());
558        match guard.get_mut(&key) {
559            Some(existing) => existing.merge_from_partin(record.clone()),
560            None => {
561                guard.insert(key, record.clone());
562            }
563        }
564        Ok(())
565    }
566
567    async fn get(
568        &self,
569        tenant_id: TenantId,
570        mp_id: &MarktpartnerCode,
571    ) -> Result<Option<PartnerRecord>, EngineError> {
572        Ok(self
573            .inner
574            .read()
575            .await
576            .get(&(tenant_id, mp_id.clone()))
577            .cloned())
578    }
579
580    async fn remove(
581        &self,
582        tenant_id: TenantId,
583        mp_id: &MarktpartnerCode,
584    ) -> Result<(), EngineError> {
585        self.inner.write().await.remove(&(tenant_id, mp_id.clone()));
586        Ok(())
587    }
588
589    async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
590        Ok(self
591            .inner
592            .read()
593            .await
594            .iter()
595            .filter(|((tid, _), _)| *tid == tenant_id)
596            .map(|(_, record)| record.clone())
597            .collect())
598    }
599}
600
601// ── Tests ─────────────────────────────────────────────────────────────────────
602
603#[cfg(test)]
604mod tests {
605    use super::*;
606
607    fn mp_id(s: &str) -> MarktpartnerCode {
608        MarktpartnerCode::new(s)
609    }
610    fn tid() -> TenantId {
611        TenantId::new()
612    }
613
614    fn minimal_record(gln_str: &str, url: &str) -> PartnerRecord {
615        PartnerRecord::minimal(mp_id(gln_str), url)
616    }
617
618    // ── from_cli_pairs ────────────────────────────────────────────────────────
619
620    #[test]
621    fn from_cli_pairs_parses_valid_entries() {
622        let pairs = vec![
623            "9900000000002=https://partner-a.example/as4/inbox",
624            "9900000000003=https://partner-b.example/as4/inbox",
625        ];
626        let records = PartnerRecord::from_cli_pairs(&pairs).unwrap();
627        assert_eq!(records.len(), 2);
628        assert_eq!(records[0].mp_id.as_str(), "9900000000002");
629        assert_eq!(
630            records[0].as4_endpoint(),
631            Some("https://partner-a.example/as4/inbox")
632        );
633        assert_eq!(records[1].mp_id.as_str(), "9900000000003");
634    }
635
636    #[test]
637    fn from_cli_pairs_rejects_missing_equals() {
638        let pairs = vec!["9900000000002https://no-equals.example"];
639        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
640    }
641
642    #[test]
643    fn from_cli_pairs_rejects_http_url() {
644        let pairs = vec!["9900000000002=http://insecure.example/as4"];
645        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
646    }
647
648    #[test]
649    fn from_cli_pairs_rejects_empty_gln() {
650        let pairs = vec!["=https://no-mp_id.example/as4"];
651        assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
652    }
653
654    // ── as4_endpoint ──────────────────────────────────────────────────────────
655
656    #[test]
657    fn as4_endpoint_returns_ak_channel() {
658        let r = minimal_record("9900000000002", "https://a.example/as4");
659        assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
660    }
661
662    #[test]
663    fn as4_endpoint_returns_none_when_absent() {
664        let r = PartnerRecord {
665            mp_id: mp_id("9900000000002"),
666            display_name: None,
667            channels: vec![CommunicationChannel::email("info@example.de")],
668            roles: vec![],
669            valid_from: None,
670            contacts: vec![],
671            country_code: None,
672            updated_at: OffsetDateTime::now_utc(),
673        };
674        assert!(r.as4_endpoint().is_none());
675    }
676
677    // ── merge_from_partin ─────────────────────────────────────────────────────
678
679    #[test]
680    fn merge_overwrites_config_record_with_partin_data() {
681        let mut base = minimal_record("9900000000002", "https://old.example/as4");
682        let newer = PartnerRecord {
683            mp_id: mp_id("9900000000002"),
684            display_name: Some("Stadtwerke AG".into()),
685            channels: vec![
686                CommunicationChannel::as4("https://new.example/as4"),
687                CommunicationChannel::email("edifact@sw.example"),
688            ],
689            roles: vec![Marktrolle::Nb],
690            valid_from: Some(OffsetDateTime::now_utc()),
691            contacts: vec![],
692            country_code: Some("DE".into()),
693            updated_at: OffsetDateTime::now_utc(),
694        };
695        base.merge_from_partin(newer.clone());
696        assert_eq!(base.as4_endpoint(), Some("https://new.example/as4"));
697        assert_eq!(base.display_name.as_deref(), Some("Stadtwerke AG"));
698        assert_eq!(base.roles, vec![Marktrolle::Nb]);
699    }
700
701    #[test]
702    fn merge_ignores_older_partin() {
703        use time::Duration;
704        let old_ts = OffsetDateTime::now_utc() - Duration::days(30);
705        let new_ts = OffsetDateTime::now_utc();
706
707        let mut current = PartnerRecord {
708            mp_id: mp_id("9900000000002"),
709            display_name: Some("Current Name".into()),
710            channels: vec![CommunicationChannel::as4("https://current.example/as4")],
711            roles: vec![Marktrolle::Nb],
712            valid_from: Some(new_ts),
713            contacts: vec![],
714            country_code: Some("DE".into()),
715            updated_at: OffsetDateTime::now_utc(),
716        };
717
718        let stale = PartnerRecord {
719            mp_id: mp_id("9900000000002"),
720            display_name: Some("Stale Name".into()),
721            channels: vec![CommunicationChannel::as4("https://stale.example/as4")],
722            roles: vec![],
723            valid_from: Some(old_ts),
724            contacts: vec![],
725            country_code: None,
726            updated_at: OffsetDateTime::now_utc(),
727        };
728
729        current.merge_from_partin(stale);
730        // Should not be overwritten
731        assert_eq!(current.display_name.as_deref(), Some("Current Name"));
732        assert_eq!(current.as4_endpoint(), Some("https://current.example/as4"));
733    }
734
735    #[test]
736    fn merge_ignores_wrong_gln() {
737        let mut r = minimal_record("9900000000002", "https://a.example/as4");
738        let other = minimal_record("9900000000003", "https://b.example/as4");
739        r.merge_from_partin(other);
740        assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
741    }
742
743    // ── roles serde (BDEW codes) ──────────────────────────────────────────────
744
745    #[test]
746    fn roles_serialize_as_bdew_codes() {
747        let mut r = minimal_record("9900000000002", "https://a.example/as4");
748        r.roles = vec![Marktrolle::Nb, Marktrolle::Msb];
749        let json = serde_json::to_value(&r).unwrap();
750        assert_eq!(json["roles"], serde_json::json!(["NB", "MSB"]));
751        let back: PartnerRecord = serde_json::from_value(json).unwrap();
752        assert_eq!(back.roles, vec![Marktrolle::Nb, Marktrolle::Msb]);
753    }
754
755    // ── InMemoryPartnerStore ──────────────────────────────────────────────────
756
757    #[tokio::test]
758    async fn in_memory_upsert_and_get() {
759        let store = InMemoryPartnerStore::new();
760        let tenant = tid();
761        let record = minimal_record("9900000000001", "https://a.example/as4");
762
763        store.upsert(tenant, &record).await.unwrap();
764        let found = store
765            .get(tenant, &mp_id("9900000000001"))
766            .await
767            .unwrap()
768            .unwrap();
769        assert_eq!(found.as4_endpoint(), Some("https://a.example/as4"));
770    }
771
772    #[tokio::test]
773    async fn in_memory_get_returns_none_for_unknown() {
774        let store = InMemoryPartnerStore::new();
775        assert!(
776            store
777                .get(tid(), &mp_id("9900000000099"))
778                .await
779                .unwrap()
780                .is_none()
781        );
782    }
783
784    #[tokio::test]
785    async fn in_memory_upsert_merges_into_existing() {
786        let store = InMemoryPartnerStore::new();
787        let tenant = tid();
788        let base = minimal_record("9900000000001", "https://old.example/as4");
789        store.upsert(tenant, &base).await.unwrap();
790
791        let newer = PartnerRecord {
792            mp_id: mp_id("9900000000001"),
793            display_name: Some("Partner AG".into()),
794            channels: vec![CommunicationChannel::as4("https://new.example/as4")],
795            roles: vec![Marktrolle::Lf],
796            valid_from: Some(OffsetDateTime::now_utc()),
797            contacts: vec![],
798            country_code: Some("DE".into()),
799            updated_at: OffsetDateTime::now_utc(),
800        };
801        store.upsert(tenant, &newer).await.unwrap();
802
803        let found = store
804            .get(tenant, &mp_id("9900000000001"))
805            .await
806            .unwrap()
807            .unwrap();
808        assert_eq!(found.as4_endpoint(), Some("https://new.example/as4"));
809        assert_eq!(found.display_name.as_deref(), Some("Partner AG"));
810    }
811
812    #[tokio::test]
813    async fn in_memory_remove_clears_record() {
814        let store = InMemoryPartnerStore::new();
815        let tenant = tid();
816        let record = minimal_record("9900000000001", "https://a.example/as4");
817
818        store.upsert(tenant, &record).await.unwrap();
819        store.remove(tenant, &mp_id("9900000000001")).await.unwrap();
820        assert!(
821            store
822                .get(tenant, &mp_id("9900000000001"))
823                .await
824                .unwrap()
825                .is_none()
826        );
827    }
828
829    #[tokio::test]
830    async fn in_memory_list_is_tenant_scoped() {
831        let store = InMemoryPartnerStore::new();
832        let t1 = tid();
833        let t2 = tid();
834
835        store
836            .upsert(
837                t1,
838                &minimal_record("9900000000001", "https://a.example/as4"),
839            )
840            .await
841            .unwrap();
842        store
843            .upsert(
844                t2,
845                &minimal_record("9900000000002", "https://b.example/as4"),
846            )
847            .await
848            .unwrap();
849
850        let t1_list = store.list(t1).await.unwrap();
851        assert_eq!(t1_list.len(), 1);
852        assert_eq!(t1_list[0].mp_id.as_str(), "9900000000001");
853
854        let t2_list = store.list(t2).await.unwrap();
855        assert_eq!(t2_list.len(), 1);
856        assert_eq!(t2_list[0].mp_id.as_str(), "9900000000002");
857    }
858
859    #[tokio::test]
860    async fn as4_endpoint_convenience_method() {
861        let store = InMemoryPartnerStore::new();
862        let tenant = tid();
863        let record = minimal_record("9900000000001", "https://a.example/as4");
864
865        store.upsert(tenant, &record).await.unwrap();
866        let url = store
867            .as4_endpoint(tenant, &mp_id("9900000000001"))
868            .await
869            .unwrap();
870        assert_eq!(url.as_deref(), Some("https://a.example/as4"));
871
872        let none = store
873            .as4_endpoint(tenant, &mp_id("9900000000099"))
874            .await
875            .unwrap();
876        assert!(none.is_none());
877    }
878}