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,
172
173 pub display_name: Option<Box<str>>,
175
176 pub channels: Vec<CommunicationChannel>,
183
184 pub roles: Vec<Marktrolle>,
190
191 #[serde(
195 default,
196 skip_serializing_if = "Option::is_none",
197 with = "time::serde::rfc3339::option"
198 )]
199 pub valid_from: Option<OffsetDateTime>,
200
201 pub contacts: Vec<ContactPerson>,
203
204 pub country_code: Option<Box<str>>,
206
207 #[serde(with = "time::serde::rfc3339")]
209 pub updated_at: OffsetDateTime,
210}
211
212impl PartnerRecord {
213 #[must_use]
219 pub fn minimal(mp_id: impl Into<MarktpartnerCode>, as4_url: impl Into<Box<str>>) -> Self {
220 Self {
221 mp_id: mp_id.into(),
222 display_name: None,
223 channels: vec![CommunicationChannel::as4(as4_url)],
224 roles: Vec::new(),
225 valid_from: None,
226 contacts: Vec::new(),
227 country_code: None,
228 updated_at: OffsetDateTime::now_utc(),
229 }
230 }
231
232 pub fn from_cli_pairs(pairs: &[impl AsRef<str>]) -> Result<Vec<Self>, EngineError> {
241 pairs
242 .iter()
243 .map(|entry| {
244 let pair = entry.as_ref();
245 let (mp_id, url) = pair.split_once('=').ok_or_else(|| {
246 EngineError::partner(format!(
247 "invalid partner entry {pair:?} — expected <GLN>=<HTTPS-URL>"
248 ))
249 })?;
250 let mp_id = mp_id.trim();
251 let url = url.trim();
252 if mp_id.is_empty() {
253 return Err(EngineError::partner(format!(
254 "invalid partner entry {pair:?} — GLN must not be empty"
255 )));
256 }
257 if !url.starts_with("https://") {
258 return Err(EngineError::partner(format!(
259 "invalid partner entry {pair:?} — endpoint URL must use HTTPS (got {url:?})"
260 )));
261 }
262 Ok(Self::minimal(mp_id, url))
263 })
264 .collect()
265 }
266
267 #[must_use]
273 pub fn as4_endpoint(&self) -> Option<&str> {
274 self.channels
275 .iter()
276 .find(|c| c.qualifier.as_ref() == "AK" || c.qualifier.as_ref() == "AS4")
277 .map(|c| c.address.as_ref())
278 }
279
280 #[must_use]
284 pub fn email(&self) -> Option<&str> {
285 self.channels
286 .iter()
287 .find(|c| c.qualifier.as_ref() == "EM")
288 .map(|c| c.address.as_ref())
289 }
290
291 #[must_use]
297 pub fn api_webdienste_endpoint(&self) -> Option<&str> {
298 self.channels
299 .iter()
300 .find(|c| c.qualifier.as_ref() == "AW")
301 .map(|c| c.address.as_ref())
302 }
303
304 pub fn merge_from_partin(&mut self, incoming: PartnerRecord) {
313 if incoming.mp_id != self.mp_id {
314 return;
315 }
316 let should_update = match (self.valid_from, incoming.valid_from) {
317 (None, _) => true,
318 (Some(_), None) => false, (Some(a), Some(b)) => b >= a,
320 };
321 if !should_update {
322 return;
323 }
324 self.display_name = incoming.display_name.or(self.display_name.take());
325 self.channels = incoming.channels;
326 self.roles = incoming.roles;
327 self.valid_from = incoming.valid_from;
328 self.contacts = incoming.contacts;
329 self.country_code = incoming.country_code.or(self.country_code.take());
330 self.updated_at = incoming.updated_at;
331 }
332}
333
334#[allow(async_fn_in_trait)]
348pub trait PartnerStore: Send + Sync {
349 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError>;
359
360 async fn get(
366 &self,
367 tenant_id: TenantId,
368 mp_id: &MarktpartnerCode,
369 ) -> Result<Option<PartnerRecord>, EngineError>;
370
371 async fn remove(
379 &self,
380 tenant_id: TenantId,
381 mp_id: &MarktpartnerCode,
382 ) -> Result<(), EngineError>;
383
384 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError>;
390
391 async fn as4_endpoint(
399 &self,
400 tenant_id: TenantId,
401 mp_id: &MarktpartnerCode,
402 ) -> Result<Option<Box<str>>, EngineError> {
403 Ok(self
404 .get(tenant_id, mp_id)
405 .await?
406 .and_then(|r| r.as4_endpoint().map(std::convert::Into::into)))
407 }
408
409 async fn api_webdienste_endpoint(
420 &self,
421 tenant_id: TenantId,
422 mp_id: &MarktpartnerCode,
423 ) -> Result<Option<Box<str>>, EngineError> {
424 Ok(self
425 .get(tenant_id, mp_id)
426 .await?
427 .and_then(|r| r.api_webdienste_endpoint().map(std::convert::Into::into)))
428 }
429}
430
431impl<S: PartnerStore> PartnerStore for Arc<S> {
434 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
435 self.as_ref().upsert(tenant_id, record).await
436 }
437
438 async fn get(
439 &self,
440 tenant_id: TenantId,
441 mp_id: &MarktpartnerCode,
442 ) -> Result<Option<PartnerRecord>, EngineError> {
443 self.as_ref().get(tenant_id, mp_id).await
444 }
445
446 async fn remove(
447 &self,
448 tenant_id: TenantId,
449 mp_id: &MarktpartnerCode,
450 ) -> Result<(), EngineError> {
451 self.as_ref().remove(tenant_id, mp_id).await
452 }
453
454 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
455 self.as_ref().list(tenant_id).await
456 }
457}
458
459#[cfg_attr(
470 not(any(test, feature = "testing")),
471 deprecated = "NoopPartnerStore must not be instantiated in production builds; \
472 PARTIN-derived partner updates will be silently discarded. \
473 Use SlateDbPartnerStore or another durable PartnerStore instead."
474)]
475#[derive(Debug, Clone, Copy, Default)]
476pub struct NoopPartnerStore;
477
478#[cfg(any(test, feature = "testing"))]
484#[allow(deprecated)]
485impl PartnerStore for NoopPartnerStore {
486 async fn upsert(
487 &self,
488 _tenant_id: TenantId,
489 _record: &PartnerRecord,
490 ) -> Result<(), EngineError> {
491 Ok(())
492 }
493
494 async fn get(
495 &self,
496 _tenant_id: TenantId,
497 _mp_id: &MarktpartnerCode,
498 ) -> Result<Option<PartnerRecord>, EngineError> {
499 Ok(None)
500 }
501
502 async fn remove(
503 &self,
504 _tenant_id: TenantId,
505 _mp_id: &MarktpartnerCode,
506 ) -> Result<(), EngineError> {
507 Ok(())
508 }
509
510 async fn list(&self, _tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
511 Ok(vec![])
512 }
513}
514
515#[cfg(any(test, feature = "testing"))]
525#[derive(Debug, Clone, Default)]
526pub struct InMemoryPartnerStore {
527 inner: Arc<RwLock<HashMap<(TenantId, MarktpartnerCode), PartnerRecord>>>,
528}
529
530#[cfg(any(test, feature = "testing"))]
531impl InMemoryPartnerStore {
532 #[must_use]
534 pub fn new() -> Self {
535 Self::default()
536 }
537}
538
539#[cfg(any(test, feature = "testing"))]
540impl PartnerStore for InMemoryPartnerStore {
541 async fn upsert(&self, tenant_id: TenantId, record: &PartnerRecord) -> Result<(), EngineError> {
542 let mut guard = self.inner.write().await;
543 let key = (tenant_id, record.mp_id.clone());
544 match guard.get_mut(&key) {
545 Some(existing) => existing.merge_from_partin(record.clone()),
546 None => {
547 guard.insert(key, record.clone());
548 }
549 }
550 Ok(())
551 }
552
553 async fn get(
554 &self,
555 tenant_id: TenantId,
556 mp_id: &MarktpartnerCode,
557 ) -> Result<Option<PartnerRecord>, EngineError> {
558 Ok(self
559 .inner
560 .read()
561 .await
562 .get(&(tenant_id, mp_id.clone()))
563 .cloned())
564 }
565
566 async fn remove(
567 &self,
568 tenant_id: TenantId,
569 mp_id: &MarktpartnerCode,
570 ) -> Result<(), EngineError> {
571 self.inner.write().await.remove(&(tenant_id, mp_id.clone()));
572 Ok(())
573 }
574
575 async fn list(&self, tenant_id: TenantId) -> Result<Vec<PartnerRecord>, EngineError> {
576 Ok(self
577 .inner
578 .read()
579 .await
580 .iter()
581 .filter(|((tid, _), _)| *tid == tenant_id)
582 .map(|(_, record)| record.clone())
583 .collect())
584 }
585}
586
587#[cfg(test)]
590mod tests {
591 use super::*;
592
593 fn mp_id(s: &str) -> MarktpartnerCode {
594 MarktpartnerCode::new(s)
595 }
596 fn tid() -> TenantId {
597 TenantId::new()
598 }
599
600 fn minimal_record(gln_str: &str, url: &str) -> PartnerRecord {
601 PartnerRecord::minimal(mp_id(gln_str), url)
602 }
603
604 #[test]
607 fn from_cli_pairs_parses_valid_entries() {
608 let pairs = vec![
609 "9900000000002=https://partner-a.example/as4/inbox",
610 "9900000000003=https://partner-b.example/as4/inbox",
611 ];
612 let records = PartnerRecord::from_cli_pairs(&pairs).unwrap();
613 assert_eq!(records.len(), 2);
614 assert_eq!(records[0].mp_id.as_str(), "9900000000002");
615 assert_eq!(
616 records[0].as4_endpoint(),
617 Some("https://partner-a.example/as4/inbox")
618 );
619 assert_eq!(records[1].mp_id.as_str(), "9900000000003");
620 }
621
622 #[test]
623 fn from_cli_pairs_rejects_missing_equals() {
624 let pairs = vec!["9900000000002https://no-equals.example"];
625 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
626 }
627
628 #[test]
629 fn from_cli_pairs_rejects_http_url() {
630 let pairs = vec!["9900000000002=http://insecure.example/as4"];
631 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
632 }
633
634 #[test]
635 fn from_cli_pairs_rejects_empty_gln() {
636 let pairs = vec!["=https://no-mp_id.example/as4"];
637 assert!(PartnerRecord::from_cli_pairs(&pairs).is_err());
638 }
639
640 #[test]
643 fn as4_endpoint_returns_ak_channel() {
644 let r = minimal_record("9900000000002", "https://a.example/as4");
645 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
646 }
647
648 #[test]
649 fn as4_endpoint_returns_none_when_absent() {
650 let r = PartnerRecord {
651 mp_id: mp_id("9900000000002"),
652 display_name: None,
653 channels: vec![CommunicationChannel::email("info@example.de")],
654 roles: vec![],
655 valid_from: None,
656 contacts: vec![],
657 country_code: None,
658 updated_at: OffsetDateTime::now_utc(),
659 };
660 assert!(r.as4_endpoint().is_none());
661 }
662
663 #[test]
666 fn merge_overwrites_config_record_with_partin_data() {
667 let mut base = minimal_record("9900000000002", "https://old.example/as4");
668 let newer = PartnerRecord {
669 mp_id: mp_id("9900000000002"),
670 display_name: Some("Stadtwerke AG".into()),
671 channels: vec![
672 CommunicationChannel::as4("https://new.example/as4"),
673 CommunicationChannel::email("edifact@sw.example"),
674 ],
675 roles: vec![Marktrolle::Nb],
676 valid_from: Some(OffsetDateTime::now_utc()),
677 contacts: vec![],
678 country_code: Some("DE".into()),
679 updated_at: OffsetDateTime::now_utc(),
680 };
681 base.merge_from_partin(newer.clone());
682 assert_eq!(base.as4_endpoint(), Some("https://new.example/as4"));
683 assert_eq!(base.display_name.as_deref(), Some("Stadtwerke AG"));
684 assert_eq!(base.roles, vec![Marktrolle::Nb]);
685 }
686
687 #[test]
688 fn merge_ignores_older_partin() {
689 use time::Duration;
690 let old_ts = OffsetDateTime::now_utc() - Duration::days(30);
691 let new_ts = OffsetDateTime::now_utc();
692
693 let mut current = PartnerRecord {
694 mp_id: mp_id("9900000000002"),
695 display_name: Some("Current Name".into()),
696 channels: vec![CommunicationChannel::as4("https://current.example/as4")],
697 roles: vec![Marktrolle::Nb],
698 valid_from: Some(new_ts),
699 contacts: vec![],
700 country_code: Some("DE".into()),
701 updated_at: OffsetDateTime::now_utc(),
702 };
703
704 let stale = PartnerRecord {
705 mp_id: mp_id("9900000000002"),
706 display_name: Some("Stale Name".into()),
707 channels: vec![CommunicationChannel::as4("https://stale.example/as4")],
708 roles: vec![],
709 valid_from: Some(old_ts),
710 contacts: vec![],
711 country_code: None,
712 updated_at: OffsetDateTime::now_utc(),
713 };
714
715 current.merge_from_partin(stale);
716 assert_eq!(current.display_name.as_deref(), Some("Current Name"));
718 assert_eq!(current.as4_endpoint(), Some("https://current.example/as4"));
719 }
720
721 #[test]
722 fn merge_ignores_wrong_gln() {
723 let mut r = minimal_record("9900000000002", "https://a.example/as4");
724 let other = minimal_record("9900000000003", "https://b.example/as4");
725 r.merge_from_partin(other);
726 assert_eq!(r.as4_endpoint(), Some("https://a.example/as4"));
727 }
728
729 #[test]
732 fn roles_serialize_as_bdew_codes() {
733 let mut r = minimal_record("9900000000002", "https://a.example/as4");
734 r.roles = vec![Marktrolle::Nb, Marktrolle::Msb];
735 let json = serde_json::to_value(&r).unwrap();
736 assert_eq!(json["roles"], serde_json::json!(["NB", "MSB"]));
737 let back: PartnerRecord = serde_json::from_value(json).unwrap();
738 assert_eq!(back.roles, vec![Marktrolle::Nb, Marktrolle::Msb]);
739 }
740
741 #[tokio::test]
744 async fn in_memory_upsert_and_get() {
745 let store = InMemoryPartnerStore::new();
746 let tenant = tid();
747 let record = minimal_record("9900000000001", "https://a.example/as4");
748
749 store.upsert(tenant, &record).await.unwrap();
750 let found = store
751 .get(tenant, &mp_id("9900000000001"))
752 .await
753 .unwrap()
754 .unwrap();
755 assert_eq!(found.as4_endpoint(), Some("https://a.example/as4"));
756 }
757
758 #[tokio::test]
759 async fn in_memory_get_returns_none_for_unknown() {
760 let store = InMemoryPartnerStore::new();
761 assert!(
762 store
763 .get(tid(), &mp_id("9900000000099"))
764 .await
765 .unwrap()
766 .is_none()
767 );
768 }
769
770 #[tokio::test]
771 async fn in_memory_upsert_merges_into_existing() {
772 let store = InMemoryPartnerStore::new();
773 let tenant = tid();
774 let base = minimal_record("9900000000001", "https://old.example/as4");
775 store.upsert(tenant, &base).await.unwrap();
776
777 let newer = PartnerRecord {
778 mp_id: mp_id("9900000000001"),
779 display_name: Some("Partner AG".into()),
780 channels: vec![CommunicationChannel::as4("https://new.example/as4")],
781 roles: vec![Marktrolle::Lf],
782 valid_from: Some(OffsetDateTime::now_utc()),
783 contacts: vec![],
784 country_code: Some("DE".into()),
785 updated_at: OffsetDateTime::now_utc(),
786 };
787 store.upsert(tenant, &newer).await.unwrap();
788
789 let found = store
790 .get(tenant, &mp_id("9900000000001"))
791 .await
792 .unwrap()
793 .unwrap();
794 assert_eq!(found.as4_endpoint(), Some("https://new.example/as4"));
795 assert_eq!(found.display_name.as_deref(), Some("Partner AG"));
796 }
797
798 #[tokio::test]
799 async fn in_memory_remove_clears_record() {
800 let store = InMemoryPartnerStore::new();
801 let tenant = tid();
802 let record = minimal_record("9900000000001", "https://a.example/as4");
803
804 store.upsert(tenant, &record).await.unwrap();
805 store.remove(tenant, &mp_id("9900000000001")).await.unwrap();
806 assert!(
807 store
808 .get(tenant, &mp_id("9900000000001"))
809 .await
810 .unwrap()
811 .is_none()
812 );
813 }
814
815 #[tokio::test]
816 async fn in_memory_list_is_tenant_scoped() {
817 let store = InMemoryPartnerStore::new();
818 let t1 = tid();
819 let t2 = tid();
820
821 store
822 .upsert(
823 t1,
824 &minimal_record("9900000000001", "https://a.example/as4"),
825 )
826 .await
827 .unwrap();
828 store
829 .upsert(
830 t2,
831 &minimal_record("9900000000002", "https://b.example/as4"),
832 )
833 .await
834 .unwrap();
835
836 let t1_list = store.list(t1).await.unwrap();
837 assert_eq!(t1_list.len(), 1);
838 assert_eq!(t1_list[0].mp_id.as_str(), "9900000000001");
839
840 let t2_list = store.list(t2).await.unwrap();
841 assert_eq!(t2_list.len(), 1);
842 assert_eq!(t2_list[0].mp_id.as_str(), "9900000000002");
843 }
844
845 #[tokio::test]
846 async fn as4_endpoint_convenience_method() {
847 let store = InMemoryPartnerStore::new();
848 let tenant = tid();
849 let record = minimal_record("9900000000001", "https://a.example/as4");
850
851 store.upsert(tenant, &record).await.unwrap();
852 let url = store
853 .as4_endpoint(tenant, &mp_id("9900000000001"))
854 .await
855 .unwrap();
856 assert_eq!(url.as_deref(), Some("https://a.example/as4"));
857
858 let none = store
859 .as4_endpoint(tenant, &mp_id("9900000000099"))
860 .await
861 .unwrap();
862 assert!(none.is_none());
863 }
864}