1use 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub struct CommunicationChannel {
97 pub qualifier: Box<str>,
99 pub address: Box<str>,
101}
102
103impl CommunicationChannel {
104 #[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 #[must_use]
117 pub fn as4(endpoint_url: impl Into<Box<str>>) -> Self {
118 Self::new("AK", endpoint_url)
119 }
120
121 #[must_use]
123 pub fn email(address: impl Into<Box<str>>) -> Self {
124 Self::new("EM", address)
125 }
126
127 #[must_use]
132 pub fn api_webdienste(base_url: impl Into<Box<str>>) -> Self {
133 Self::new("AW", base_url)
134 }
135}
136
137#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
143pub struct ContactPerson {
144 pub name: Box<str>,
146 pub channels: Vec<CommunicationChannel>,
148}
149
150#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
169pub struct PartnerRecord {
170 pub mp_id: MarktpartnerCode,
173
174 #[serde(default)]
176 pub display_name: Option<Box<str>>,
177
178 #[serde(default)]
185 pub channels: Vec<CommunicationChannel>,
186
187 #[serde(default)]
193 pub roles: Vec<Marktrolle>,
194
195 #[serde(
199 default,
200 skip_serializing_if = "Option::is_none",
201 with = "time::serde::rfc3339::option"
202 )]
203 pub valid_from: Option<OffsetDateTime>,
204
205 #[serde(default)]
207 pub contacts: Vec<ContactPerson>,
208
209 #[serde(default)]
211 pub country_code: Option<Box<str>>,
212
213 #[serde(default = "OffsetDateTime::now_utc", with = "time::serde::rfc3339")]
222 pub updated_at: OffsetDateTime,
223}
224
225impl PartnerRecord {
226 #[must_use]
232 pub fn minimal(mp_id: impl Into<MarktpartnerCode>, as4_url: impl Into<Box<str>>) -> Self {
233 Self {
234 mp_id: mp_id.into(),
235 display_name: None,
236 channels: vec![CommunicationChannel::as4(as4_url)],
237 roles: Vec::new(),
238 valid_from: None,
239 contacts: Vec::new(),
240 country_code: None,
241 updated_at: OffsetDateTime::now_utc(),
242 }
243 }
244
245 pub fn from_cli_pairs(pairs: &[impl AsRef<str>]) -> Result<Vec<Self>, EngineError> {
254 pairs
255 .iter()
256 .map(|entry| {
257 let pair = entry.as_ref();
258 let (mp_id, url) = pair.split_once('=').ok_or_else(|| {
259 EngineError::partner(format!(
260 "invalid partner entry {pair:?} — expected <GLN>=<HTTPS-URL>"
261 ))
262 })?;
263 let mp_id = mp_id.trim();
264 let url = url.trim();
265 if mp_id.is_empty() {
266 return Err(EngineError::partner(format!(
267 "invalid partner entry {pair:?} — GLN must not be empty"
268 )));
269 }
270 if !url.starts_with("https://") {
271 return Err(EngineError::partner(format!(
272 "invalid partner entry {pair:?} — endpoint URL must use HTTPS (got {url:?})"
273 )));
274 }
275 Ok(Self::minimal(mp_id, url))
276 })
277 .collect()
278 }
279
280 #[must_use]
286 pub fn as4_endpoint(&self) -> Option<&str> {
287 self.channels
288 .iter()
289 .find(|c| c.qualifier.as_ref() == "AK" || c.qualifier.as_ref() == "AS4")
290 .map(|c| c.address.as_ref())
291 }
292
293 #[must_use]
297 pub fn email(&self) -> Option<&str> {
298 self.channels
299 .iter()
300 .find(|c| c.qualifier.as_ref() == "EM")
301 .map(|c| c.address.as_ref())
302 }
303
304 #[must_use]
310 pub fn api_webdienste_endpoint(&self) -> Option<&str> {
311 self.channels
312 .iter()
313 .find(|c| c.qualifier.as_ref() == "AW")
314 .map(|c| c.address.as_ref())
315 }
316
317 pub fn merge_from_partin(&mut self, incoming: PartnerRecord) {
326 if incoming.mp_id != self.mp_id {
327 return;
328 }
329 let should_update = match (self.valid_from, incoming.valid_from) {
330 (None, _) => true,
331 (Some(_), None) => false, (Some(a), Some(b)) => b >= a,
333 };
334 if !should_update {
335 return;
336 }
337 self.display_name = incoming.display_name.or(self.display_name.take());
338 self.channels = incoming.channels;
339 self.roles = incoming.roles;
340 self.valid_from = incoming.valid_from;
341 self.contacts = incoming.contacts;
342 self.country_code = incoming.country_code.or(self.country_code.take());
343 self.updated_at = incoming.updated_at;
344 }
345}
346
347#[allow(async_fn_in_trait)]
361pub trait PartnerStore: Send + Sync {
362 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError>;
372
373 async fn get(
379 &self,
380 tenant_id: TenantId,
381 mp_id: &MarktpartnerCode,
382 ) -> Result<Option<PartnerRecord>, EngineError>;
383
384 async fn remove(
392 &self,
393 tenant_id: TenantId,
394 mp_id: &MarktpartnerCode,
395 ) -> Result<(), EngineError>;
396
397 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError>;
403
404 async fn as4_endpoint(
412 &self,
413 tenant_id: TenantId,
414 mp_id: &MarktpartnerCode,
415 ) -> Result<Option<Box<str>>, EngineError> {
416 Ok(self
417 .get(tenant_id, mp_id)
418 .await?
419 .and_then(|r| r.as4_endpoint().map(std::convert::Into::into)))
420 }
421
422 async fn api_webdienste_endpoint(
433 &self,
434 tenant_id: TenantId,
435 mp_id: &MarktpartnerCode,
436 ) -> Result<Option<Box<str>>, EngineError> {
437 Ok(self
438 .get(tenant_id, mp_id)
439 .await?
440 .and_then(|r| r.api_webdienste_endpoint().map(std::convert::Into::into)))
441 }
442}
443
444impl<S: PartnerStore> PartnerStore for Arc<S> {
447 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
448 self.as_ref().upsert(tenant_id, record).await
449 }
450
451 async fn get(
452 &self,
453 tenant_id: TenantId,
454 mp_id: &MarktpartnerCode,
455 ) -> Result<Option<PartnerRecord>, EngineError> {
456 self.as_ref().get(tenant_id, mp_id).await
457 }
458
459 async fn remove(
460 &self,
461 tenant_id: TenantId,
462 mp_id: &MarktpartnerCode,
463 ) -> Result<(), EngineError> {
464 self.as_ref().remove(tenant_id, mp_id).await
465 }
466
467 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
468 self.as_ref().list(tenant_id).await
469 }
470}
471
472#[cfg_attr(
483 not(any(test, feature = "testing")),
484 deprecated = "NoopPartnerStore must not be instantiated in production builds; \
485 PARTIN-derived partner updates will be silently discarded. \
486 Use SlateDbPartnerStore or another durable PartnerStore instead."
487)]
488#[derive(Debug, Clone, Copy, Default)]
489pub struct NoopPartnerStore;
490
491#[cfg(any(test, feature = "testing"))]
497#[allow(deprecated)]
498impl PartnerStore for NoopPartnerStore {
499 async fn upsert(
500 &self,
501 _tenant_id: TenantId,
502 _record: &PartnerRecord,
503 ) -> Result<(), EngineError> {
504 Ok(())
505 }
506
507 async fn get(
508 &self,
509 _tenant_id: TenantId,
510 _mp_id: &MarktpartnerCode,
511 ) -> Result<Option<PartnerRecord>, EngineError> {
512 Ok(None)
513 }
514
515 async fn remove(
516 &self,
517 _tenant_id: TenantId,
518 _mp_id: &MarktpartnerCode,
519 ) -> Result<(), EngineError> {
520 Ok(())
521 }
522
523 async fn list(&self, _tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
524 Ok(vec![])
525 }
526}
527
528#[cfg(any(test, feature = "testing"))]
538#[derive(Debug, Clone, Default)]
539pub struct InMemoryPartnerStore {
540 inner: Arc<RwLock<HashMap<(TenantId, MarktpartnerCode), PartnerRecord>>>,
541}
542
543#[cfg(any(test, feature = "testing"))]
544impl InMemoryPartnerStore {
545 #[must_use]
547 pub fn new() -> Self {
548 Self::default()
549 }
550}
551
552#[cfg(any(test, feature = "testing"))]
553impl PartnerStore for InMemoryPartnerStore {
554 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
555 let mut guard = self.inner.write().await;
556 let key = (tenant_id, record.mp_id.clone());
557 match guard.get_mut(&key) {
558 Some(existing) => existing.merge_from_partin(record.clone()),
559 None => {
560 guard.insert(key, record.clone());
561 }
562 }
563 Ok(())
564 }
565
566 async fn get(
567 &self,
568 tenant_id: TenantId,
569 mp_id: &MarktpartnerCode,
570 ) -> Result<Option<PartnerRecord>, EngineError> {
571 Ok(self
572 .inner
573 .read()
574 .await
575 .get(&(tenant_id, mp_id.clone()))
576 .cloned())
577 }
578
579 async fn remove(
580 &self,
581 tenant_id: TenantId,
582 mp_id: &MarktpartnerCode,
583 ) -> Result<(), EngineError> {
584 self.inner.write().await.remove(&(tenant_id, mp_id.clone()));
585 Ok(())
586 }
587
588 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
589 Ok(self
590 .inner
591 .read()
592 .await
593 .iter()
594 .filter(|((tid, _), _)| *tid == tenant_id)
595 .map(|(_, record)| record.clone())
596 .collect())
597 }
598}
599
600#[cfg(test)]
603mod tests {
604 use super::*;
605
606 fn mp_id(s: &str) -> MarktpartnerCode {
607 MarktpartnerCode::new(s)
608 }
609 fn tid() -> TenantId {
610 TenantId::new()
611 }
612
613 fn minimal_record(gln_str: &str, url: &str) -> PartnerRecord {
614 PartnerRecord::minimal(mp_id(gln_str), url)
615 }
616
617 #[test]
620 fn from_cli_pairs_parses_valid_entries() {
621 let pairs = vec![
622 "9900000000002=https://partner-a.example/as4/inbox",
623 "9900000000003=https://partner-b.example/as4/inbox",
624 ];
625 let records = PartnerRecord::from_cli_pairs(&pairs).unwrap();
626 assert_eq!(records.len(), 2);
627 assert_eq!(records[0].mp_id.as_str(), "9900000000002");
628 assert_eq!(
629 records[0].as4_endpoint(),
630 Some("https://partner-a.example/as4/inbox")
631 );
632 assert_eq!(records[1].mp_id.as_str(), "9900000000003");
633 }
634
635 #[test]
636 fn from_cli_pairs_rejects_missing_equals() {
637 let pairs = vec!["9900000000002https://no-equals.example"];
638 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
639 }
640
641 #[test]
642 fn from_cli_pairs_rejects_http_url() {
643 let pairs = vec!["9900000000002=http://insecure.example/as4"];
644 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
645 }
646
647 #[test]
648 fn from_cli_pairs_rejects_empty_gln() {
649 let pairs = vec!["=https://no-mp_id.example/as4"];
650 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
651 }
652
653 #[test]
656 fn as4_endpoint_returns_ak_channel() {
657 let r = minimal_record("9900000000002", "https://a.example/as4");
658 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
659 }
660
661 #[test]
662 fn as4_endpoint_returns_none_when_absent() {
663 let r = PartnerRecord {
664 mp_id: mp_id("9900000000002"),
665 display_name: None,
666 channels: vec![CommunicationChannel::email("info@example.de")],
667 roles: vec![],
668 valid_from: None,
669 contacts: vec![],
670 country_code: None,
671 updated_at: OffsetDateTime::now_utc(),
672 };
673 assert!(r.as4_endpoint().is_none());
674 }
675
676 #[test]
679 fn merge_overwrites_config_record_with_partin_data() {
680 let mut base = minimal_record("9900000000002", "https://old.example/as4");
681 let newer = PartnerRecord {
682 mp_id: mp_id("9900000000002"),
683 display_name: Some("Stadtwerke AG".into()),
684 channels: vec![
685 CommunicationChannel::as4("https://new.example/as4"),
686 CommunicationChannel::email("edifact@sw.example"),
687 ],
688 roles: vec![Marktrolle::Nb],
689 valid_from: Some(OffsetDateTime::now_utc()),
690 contacts: vec![],
691 country_code: Some("DE".into()),
692 updated_at: OffsetDateTime::now_utc(),
693 };
694 base.merge_from_partin(newer.clone());
695 assert_eq!(base.as4_endpoint(), Some("https://new.example/as4"));
696 assert_eq!(base.display_name.as_deref(), Some("Stadtwerke AG"));
697 assert_eq!(base.roles, vec![Marktrolle::Nb]);
698 }
699
700 #[test]
701 fn merge_ignores_older_partin() {
702 use time::Duration;
703 let old_ts = OffsetDateTime::now_utc() - Duration::days(30);
704 let new_ts = OffsetDateTime::now_utc();
705
706 let mut current = PartnerRecord {
707 mp_id: mp_id("9900000000002"),
708 display_name: Some("Current Name".into()),
709 channels: vec![CommunicationChannel::as4("https://current.example/as4")],
710 roles: vec![Marktrolle::Nb],
711 valid_from: Some(new_ts),
712 contacts: vec![],
713 country_code: Some("DE".into()),
714 updated_at: OffsetDateTime::now_utc(),
715 };
716
717 let stale = PartnerRecord {
718 mp_id: mp_id("9900000000002"),
719 display_name: Some("Stale Name".into()),
720 channels: vec![CommunicationChannel::as4("https://stale.example/as4")],
721 roles: vec![],
722 valid_from: Some(old_ts),
723 contacts: vec![],
724 country_code: None,
725 updated_at: OffsetDateTime::now_utc(),
726 };
727
728 current.merge_from_partin(stale);
729 assert_eq!(current.display_name.as_deref(), Some("Current Name"));
731 assert_eq!(current.as4_endpoint(), Some("https://current.example/as4"));
732 }
733
734 #[test]
735 fn merge_ignores_wrong_gln() {
736 let mut r = minimal_record("9900000000002", "https://a.example/as4");
737 let other = minimal_record("9900000000003", "https://b.example/as4");
738 r.merge_from_partin(other);
739 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
740 }
741
742 #[test]
745 fn roles_serialize_as_bdew_codes() {
746 let mut r = minimal_record("9900000000002", "https://a.example/as4");
747 r.roles = vec![Marktrolle::Nb, Marktrolle::Msb];
748 let json = serde_json::to_value(&r).unwrap();
749 assert_eq!(json["roles"], serde_json::json!(["NB", "MSB"]));
750 let back: PartnerRecord = serde_json::from_value(json).unwrap();
751 assert_eq!(back.roles, vec![Marktrolle::Nb, Marktrolle::Msb]);
752 }
753
754 #[tokio::test]
757 async fn in_memory_upsert_and_get() {
758 let store = InMemoryPartnerStore::new();
759 let tenant = tid();
760 let record = minimal_record("9900000000001", "https://a.example/as4");
761
762 store.upsert(tenant, &record).await.unwrap();
763 let found = store
764 .get(tenant, &mp_id("9900000000001"))
765 .await
766 .unwrap()
767 .unwrap();
768 assert_eq!(found.as4_endpoint(), Some("https://a.example/as4"));
769 }
770
771 #[tokio::test]
772 async fn in_memory_get_returns_none_for_unknown() {
773 let store = InMemoryPartnerStore::new();
774 assert!(
775 store
776 .get(tid(), &mp_id("9900000000099"))
777 .await
778 .unwrap()
779 .is_none()
780 );
781 }
782
783 #[tokio::test]
784 async fn in_memory_upsert_merges_into_existing() {
785 let store = InMemoryPartnerStore::new();
786 let tenant = tid();
787 let base = minimal_record("9900000000001", "https://old.example/as4");
788 store.upsert(tenant, &base).await.unwrap();
789
790 let newer = PartnerRecord {
791 mp_id: mp_id("9900000000001"),
792 display_name: Some("Partner AG".into()),
793 channels: vec![CommunicationChannel::as4("https://new.example/as4")],
794 roles: vec![Marktrolle::Lf],
795 valid_from: Some(OffsetDateTime::now_utc()),
796 contacts: vec![],
797 country_code: Some("DE".into()),
798 updated_at: OffsetDateTime::now_utc(),
799 };
800 store.upsert(tenant, &newer).await.unwrap();
801
802 let found = store
803 .get(tenant, &mp_id("9900000000001"))
804 .await
805 .unwrap()
806 .unwrap();
807 assert_eq!(found.as4_endpoint(), Some("https://new.example/as4"));
808 assert_eq!(found.display_name.as_deref(), Some("Partner AG"));
809 }
810
811 #[tokio::test]
812 async fn in_memory_remove_clears_record() {
813 let store = InMemoryPartnerStore::new();
814 let tenant = tid();
815 let record = minimal_record("9900000000001", "https://a.example/as4");
816
817 store.upsert(tenant, &record).await.unwrap();
818 store.remove(tenant, &mp_id("9900000000001")).await.unwrap();
819 assert!(
820 store
821 .get(tenant, &mp_id("9900000000001"))
822 .await
823 .unwrap()
824 .is_none()
825 );
826 }
827
828 #[tokio::test]
829 async fn in_memory_list_is_tenant_scoped() {
830 let store = InMemoryPartnerStore::new();
831 let t1 = tid();
832 let t2 = tid();
833
834 store
835 .upsert(
836 t1,
837 &minimal_record("9900000000001", "https://a.example/as4"),
838 )
839 .await
840 .unwrap();
841 store
842 .upsert(
843 t2,
844 &minimal_record("9900000000002", "https://b.example/as4"),
845 )
846 .await
847 .unwrap();
848
849 let t1_list = store.list(t1).await.unwrap();
850 assert_eq!(t1_list.len(), 1);
851 assert_eq!(t1_list[0].mp_id.as_str(), "9900000000001");
852
853 let t2_list = store.list(t2).await.unwrap();
854 assert_eq!(t2_list.len(), 1);
855 assert_eq!(t2_list[0].mp_id.as_str(), "9900000000002");
856 }
857
858 #[tokio::test]
859 async fn as4_endpoint_convenience_method() {
860 let store = InMemoryPartnerStore::new();
861 let tenant = tid();
862 let record = minimal_record("9900000000001", "https://a.example/as4");
863
864 store.upsert(tenant, &record).await.unwrap();
865 let url = store
866 .as4_endpoint(tenant, &mp_id("9900000000001"))
867 .await
868 .unwrap();
869 assert_eq!(url.as_deref(), Some("https://a.example/as4"));
870
871 let none = store
872 .as4_endpoint(tenant, &mp_id("9900000000099"))
873 .await
874 .unwrap();
875 assert!(none.is_none());
876 }
877}