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)]
169#[serde(deny_unknown_fields)]
170pub struct PartnerRecord {
171 pub mp_id: MarktpartnerCode,
174
175 #[serde(default)]
177 pub display_name: Option<Box<str>>,
178
179 #[serde(default)]
186 pub channels: Vec<CommunicationChannel>,
187
188 #[serde(default)]
194 pub roles: Vec<Marktrolle>,
195
196 #[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 #[serde(default)]
208 pub contacts: Vec<ContactPerson>,
209
210 #[serde(default)]
212 pub country_code: Option<Box<str>>,
213
214 #[serde(default = "OffsetDateTime::now_utc", with = "time::serde::rfc3339")]
223 pub updated_at: OffsetDateTime,
224}
225
226impl PartnerRecord {
227 #[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 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 #[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 #[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 #[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 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, (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#[allow(async_fn_in_trait)]
362pub trait PartnerStore: Send + Sync {
363 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError>;
373
374 async fn get(
380 &self,
381 tenant_id: TenantId,
382 mp_id: &MarktpartnerCode,
383 ) -> Result<Option<PartnerRecord>, EngineError>;
384
385 async fn remove(
393 &self,
394 tenant_id: TenantId,
395 mp_id: &MarktpartnerCode,
396 ) -> Result<(), EngineError>;
397
398 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError>;
404
405 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 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
445impl<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#[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#[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#[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 #[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#[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 #[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 #[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 #[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 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 #[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 #[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}