Skip to main content

mako_engine/
erp.rs

1//! ERP integration traits and reference implementations.
2//!
3//! ## Role
4//!
5//! `mako-engine` is a protocol processor — it handles EDIFACT parsing, BDEW
6//! process rules, AS4 delivery, and regulatory deadlines. All contract data,
7//! billing logic, and master data live in the operator's ERP.
8//!
9//! This module defines the **stable integration contract** between `mako-engine`
10//! and external ERP or backend systems.  The payload contract is **BO4E**, not
11//! raw EDIFACT.  ERP adapters never see EDIFACT segment codes or format-version
12//! identifiers — those are absorbed inside `mako-engine`.
13//!
14//! ## Outbound: mako → ERP
15//!
16//! Implement [`ErpAdapter`] and register it at startup.  Every domain event
17//! that requires ERP action is delivered as an [`ErpEvent`].  The production
18//! `WebhookErpAdapter` (in `makod`) serialises events as
19//! **[CloudEvents 1.0](https://cloudevents.io) structured-mode JSON** and POSTs
20//! them to the configured ERP endpoint.
21//!
22//! ```text
23//! POST <erp_webhook_url>
24//! Content-Type: application/cloudevents+json
25//! X-Idempotency-Key: <event.idempotency_key>
26//! X-Mako-Signature: <hmac-sha256-hex>   ← only when secret is configured
27//!
28//! {
29//!   "specversion": "1.0",
30//!   "id": "<idempotency_key>",
31//!   "source": "urn:mako:tenant:<tenant_id>",
32//!   "type": "de.mako.aperak.accepted",
33//!   "time": "2026-10-01T10:15:00+02:00",
34//!   "subject": "<process_id>",
35//!   "dataschema": "https://.../Marktlokation.json",
36//!   "datacontenttype": "application/json",
37//!   "makoconvid": "<conversation_id>",
38//!   "makocausationid": "<causation_id>",
39//!   "makopid": 55001,
40//!   "data": { "_typ": "MARKTLOKATION", ... }
41//! }
42//! ```
43//!
44//! See [`ErpEventType::cloud_event_type`] for the full type → CE type mapping.
45//! The BO4E payload is always in the `data` field; the `payload_schema` URL
46//! maps to the CloudEvents `dataschema` attribute.
47//!
48//! ## Inbound: ERP → mako (event-driven)
49//!
50//! For ERP systems with a message bus, implement [`ErpCommandSource`] to feed
51//! BO4E business objects into the engine without a synchronous REST round-trip.
52//!
53//! ```rust,ignore
54//! struct MyKafkaSource { consumer: KafkaConsumer }
55//!
56//! impl ErpCommandSource for MyKafkaSource {
57//!     async fn next(&self) -> Result<Option<InboundErpCommand>, ErpAdapterError> {
58//!         let msg = self.consumer.poll(Duration::from_millis(100)).await;
59//!         Ok(msg.map(|m| InboundErpCommand {
60//!             idempotency_key: m.offset().to_string(),
61//!             tenant_id: TenantId::new(),
62//!             payload_schema: "…/Marktlokation.json".into(),
63//!             payload: serde_json::from_slice(m.payload()).unwrap(),
64//!         }))
65//!     }
66//!
67//!     async fn ack(&self, id: &str) -> Result<(), ErpAdapterError> {
68//!         self.consumer.commit_offset(id.parse().unwrap()).await
69//!             .map_err(ErpAdapterError::transport)
70//!     }
71//!
72//!     async fn nack(&self, _id: &str, _reason: &str) -> Result<(), ErpAdapterError> {
73//!         Ok(()) // Kafka auto-redelivers on next poll
74//!     }
75//! }
76//! ```
77//!
78//! ## Reference implementations
79//!
80//! | Type | Feature | Use case |
81//! |------|---------|---------|
82//! | `NoopErpAdapter` | `testing` | Unit tests, CI |
83//! | [`LogErpAdapter`] | — | Structured log output; starting point for new integrations |
84//! | `NoopErpCommandSource` | `testing` | No-op inbound source for tests |
85//!
86//! For the production `WebhookErpAdapter` and `POST /api/v1/commands` endpoint,
87//! see `makod/src/erp_adapter.rs`.
88
89use std::sync::Arc;
90
91use serde::{Deserialize, Serialize};
92use time::OffsetDateTime;
93
94use crate::erc::ErcCode;
95use crate::ids::{ConversationId, EventId, ProcessId, TenantId};
96
97// ── ErpAdapterError ───────────────────────────────────────────────────────────
98
99/// Errors produced by [`ErpAdapter`] and [`ErpCommandSource`] implementations.
100#[derive(Debug, thiserror::Error)]
101pub enum ErpAdapterError {
102    /// The ERP response payload could not be deserialised or is semantically
103    /// invalid.
104    #[error("ERP payload error: {0}")]
105    Payload(String),
106
107    /// A transient transport error (network timeout, HTTP 5xx, broker
108    /// disconnect).  The delivery worker will retry with exponential backoff.
109    #[error("ERP transport error: {0}")]
110    Transport(String),
111
112    /// A permanent, non-retryable error (e.g. invalid configuration,
113    /// authentication failure).  The delivery worker will dead-letter the
114    /// message.
115    #[error("ERP permanent error: {0}")]
116    Permanent(String),
117}
118
119impl ErpAdapterError {
120    /// Construct a [`Payload`](ErpAdapterError::Payload) variant.
121    pub fn payload(e: impl std::fmt::Display) -> Self {
122        Self::Payload(e.to_string())
123    }
124
125    /// Construct a [`Transport`](ErpAdapterError::Transport) variant.
126    pub fn transport(e: impl std::fmt::Display) -> Self {
127        Self::Transport(e.to_string())
128    }
129
130    /// Construct a [`Permanent`](ErpAdapterError::Permanent) variant.
131    pub fn permanent(e: impl std::fmt::Display) -> Self {
132        Self::Permanent(e.to_string())
133    }
134
135    /// Returns `true` for transient errors that warrant a retry.
136    #[must_use]
137    pub fn is_retryable(&self) -> bool {
138        matches!(self, Self::Transport(_))
139    }
140}
141
142// ── ErpEventType ─────────────────────────────────────────────────────────────
143
144/// Semantic classification of an outbound ERP process event.
145///
146/// The ERP uses this to decide which action to take — update an order status,
147/// trigger a billing run, open a complaint ticket, etc.
148#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
149#[serde(rename_all = "snake_case")]
150pub enum ErpEventType {
151    /// A new MaKo process was spawned (e.g. inbound UTILMD received).
152    ProcessInitiated,
153    /// The counterparty sent an APERAK accepting our UTILMD.
154    AperakAccepted,
155    /// The counterparty sent an APERAK rejecting our UTILMD.
156    ///
157    /// `erc_code` is `Some` when the APERAK carried a structured ERC segment
158    /// (BDEW APERAK AHB 1.0 §2.2).  It is `None` for legacy outbox messages
159    /// that predate the typed ERC code field.
160    AperakRejected {
161        /// Structured BDEW ERC error code from the APERAK ERC segment.
162        ///
163        /// Use [`crate::erc::recommended_action`] to derive the
164        /// recommended automated ERP response.
165        #[serde(skip_serializing_if = "Option::is_none")]
166        erc_code: Option<ErcCode>,
167    },
168    /// No APERAK received within the regulatory SLA window (deadline expired).
169    AperakTimeout,
170    /// A CONTRL syntax acknowledgement was received.
171    ContrlReceived,
172    /// The process reached its terminal success state
173    /// (e.g. Lieferbeginn/Lieferende confirmed).
174    ProcessCompleted,
175    /// A MaLo identification request was successfully resolved: the MaLo was
176    /// found and the positive callback was delivered to the requesting LF.
177    ///
178    /// The `payload` field of the associated [`ErpEvent`] carries a BO4E
179    /// `Marktlokation` JSON object with the resolved MaLo data.
180    MaloIdentified,
181    /// The process failed permanently (regulatory timeout, data error, …).
182    ProcessFailed {
183        /// Human-readable failure description.
184        reason: Box<str>,
185    },
186    /// A WiM Steuerungsauftrag (PID 55168) dispatch was positively confirmed by
187    /// the MSB (`EndantwortPositiv`). Triggers downstream VPP settlement billing
188    /// in `billingd` via `POST /api/v1/webhooks/vpp-dispatch`.
189    ///
190    /// Only emitted for `Konfiguration` (load-reduction) commands — not for
191    /// `InitialZustand` (reset) commands, which restore normal operation.
192    ///
193    /// The CE `data` payload carries:
194    /// `tx_id`, `location_id`, `location_type`, `execution_time_from`,
195    /// `execution_time_until`, `max_power_kw`, `command_type`, `sender_mp_id`,
196    /// `produkt_code`.
197    VppDispatchConfirmed,
198}
199
200impl ErpEventType {
201    /// Short label for structured logging and metrics.
202    #[must_use]
203    pub fn label(&self) -> &'static str {
204        match self {
205            Self::ProcessInitiated => "process_initiated",
206            Self::AperakAccepted => "aperak_accepted",
207            Self::AperakRejected { .. } => "aperak_rejected",
208            Self::AperakTimeout => "aperak_timeout",
209            Self::ContrlReceived => "contrl_received",
210            Self::ProcessCompleted => "process_completed",
211            Self::MaloIdentified => "malo_identified",
212            Self::ProcessFailed { .. } => "process_failed",
213            Self::VppDispatchConfirmed => "vpp_dispatch_confirmed",
214        }
215    }
216
217    /// CloudEvents 1.0 `type` attribute for this event.
218    ///
219    /// Follows the reverse-DNS prefix convention (`de.mako.<domain>.<action>`).
220    /// Used by the `WebhookErpAdapter` to populate the `type` field of the
221    /// CloudEvents envelope.
222    #[must_use]
223    pub fn cloud_event_type(&self) -> &'static str {
224        match self {
225            Self::ProcessInitiated => "de.mako.process.initiated",
226            Self::AperakAccepted => "de.mako.aperak.accepted",
227            Self::AperakRejected { .. } => "de.mako.aperak.rejected",
228            Self::AperakTimeout => "de.mako.aperak.timeout",
229            Self::ContrlReceived => "de.mako.contrl.received",
230            Self::ProcessCompleted => "de.mako.process.completed",
231            Self::MaloIdentified => "de.mako.malo.identified",
232            Self::ProcessFailed { .. } => "de.mako.process.failed",
233            Self::VppDispatchConfirmed => "de.vpp.dispatch.confirmed",
234        }
235    }
236}
237
238// ── ErpEvent ──────────────────────────────────────────────────────────────────
239
240/// A structured process event delivered from `mako-engine` to the ERP.
241///
242/// The payload is always a **BO4E-typed JSON object** — the ERP adapter never
243/// receives raw EDIFACT bytes or EDIFACT format-version identifiers.
244///
245/// On the wire (via `WebhookErpAdapter`) this struct is serialised as a
246/// **[CloudEvents 1.0](https://cloudevents.io) structured-mode JSON** envelope
247/// with `Content-Type: application/cloudevents+json`.  The BO4E payload lives
248/// in the CloudEvents `data` field; `payload_schema` maps to `dataschema`;
249/// `event_type` maps to the `type` attribute via [`ErpEventType::cloud_event_type`].
250///
251/// ## Idempotency
252///
253/// `idempotency_key` maps to the CloudEvents `id` attribute and is also sent
254/// as `X-Idempotency-Key` for ERP middleware that keys on headers.  The ERP
255/// **must** persist this key and return `HTTP 200 OK` for duplicate deliveries.
256#[derive(Debug, Clone, Serialize, Deserialize)]
257pub struct ErpEvent {
258    /// Stable dedup key — store in the ERP to reject duplicate deliveries.
259    ///
260    /// Derived from the outbox `message_id`; stable across retries.
261    pub idempotency_key: String,
262
263    /// Semantic classification of this event.
264    pub event_type: ErpEventType,
265
266    /// The mako process that generated this event.
267    pub process_id: ProcessId,
268
269    /// Tenant (operator GLN) that owns this process.
270    pub tenant_id: TenantId,
271
272    /// BDEW business conversation identifier.
273    pub conversation_id: ConversationId,
274
275    /// The mako domain event that directly caused this ERP notification.
276    pub causation_id: EventId,
277
278    /// Prüfidentifikator of the process.
279    pub pid: u32,
280
281    /// BO4E JSON Schema URL that validates [`payload`](ErpEvent::payload).
282    ///
283    /// Examples:
284    /// - `"https://raw.githubusercontent.com/BO4E/BO4E-Schemas/v202607.0.0/src/bo4e_schemas/bo/Marktlokation.json"`
285    /// - `"https://raw.githubusercontent.com/BO4E/BO4E-Schemas/v202607.0.0/src/bo4e_schemas/bo/Messlokation.json"`
286    ///
287    /// `None` for events where no primary BO4E object is applicable
288    /// (e.g. `ContrlReceived`).
289    #[serde(skip_serializing_if = "Option::is_none")]
290    pub payload_schema: Option<String>,
291
292    /// BO4E-typed payload.
293    ///
294    /// Deserialise using the ERP's own BO4E library.  Raw EDIFACT structures
295    /// are never exposed here.  `null` when no payload is applicable.
296    pub payload: serde_json::Value,
297
298    /// Wall-clock time when the domain event was persisted.
299    pub occurred_at: OffsetDateTime,
300
301    /// W3C `traceparent` propagated from the request that caused this event.
302    ///
303    /// Copied from `OutboxMessage::trace_context`; injected into the webhook
304    /// delivery as the `traceparent` HTTP header and the CloudEvents
305    /// `traceparent` extension attribute (CloudEvents distributed-tracing
306    /// extension), so the ERP joins the same trace as the inbound transport.
307    #[serde(default, skip_serializing_if = "Option::is_none")]
308    pub trace_context: Option<Box<str>>,
309
310    /// Workflow family name that produced this event (e.g. `"gpke-sperrung"`).
311    ///
312    /// Carried through from `OutboxMessage::workflow_name`.  Emitted as the
313    /// `makoworkflow` CloudEvents extension attribute by `WebhookErpAdapter`.
314    /// `marktd` maps this to `marktrole` for role-scoped ERP subscriber fan-out.
315    ///
316    /// Empty string for events produced by legacy outbox messages that
317    /// predate this field.
318    pub workflow_name: Box<str>,
319}
320
321// ── ErpAdapter trait ──────────────────────────────────────────────────────────
322
323/// Outbound notification sink — `mako-engine` calls this when a process event
324/// should be reported to the ERP.
325///
326/// The payload is always a BO4E-typed JSON object; the adapter never receives
327/// raw EDIFACT bytes or format-version identifiers.
328///
329/// ## Contract
330///
331/// - Must be **idempotent** on `event.idempotency_key`.  Called twice with the
332///   same key must succeed without double-posting.
333/// - Return [`ErpAdapterError::Transport`] for transient failures — the caller
334///   will retry with exponential backoff.
335/// - Return [`ErpAdapterError::Permanent`] for non-retryable failures — the
336///   caller will dead-letter the event.
337#[allow(async_fn_in_trait)]
338pub trait ErpAdapter: Send + Sync + 'static {
339    /// Deliver `event` to the ERP.
340    async fn notify(&self, event: ErpEvent) -> Result<(), ErpAdapterError>;
341}
342
343/// Blanket `Arc` implementation so `ErpAdapter` can be shared across tasks.
344impl<T: ErpAdapter> ErpAdapter for Arc<T> {
345    async fn notify(&self, event: ErpEvent) -> Result<(), ErpAdapterError> {
346        (**self).notify(event).await
347    }
348}
349
350// ── InboundErpCommand ─────────────────────────────────────────────────────────
351
352/// A BO4E business object received from the ERP, intended to trigger a mako
353/// process.
354///
355/// `mako-engine` maps the BO4E payload to an internal `Command` via the
356/// domain crate's command mapper.
357#[derive(Debug, Clone, Serialize, Deserialize)]
358pub struct InboundErpCommand {
359    /// Stable dedup key — forwarded to [`InboxStore::accept`].
360    ///
361    /// The ERP must supply a stable, unique identifier per command so that
362    /// retransmissions do not double-execute the workflow.
363    ///
364    /// [`InboxStore::accept`]: crate::inbox::InboxStore::accept
365    pub idempotency_key: String,
366
367    /// Tenant (operator GLN) that owns the target process.
368    pub tenant_id: TenantId,
369
370    /// BO4E JSON Schema URL — identifies the object type without inspecting
371    /// `payload`.
372    ///
373    /// Example:
374    /// `"https://raw.githubusercontent.com/BO4E/BO4E-Schemas/v202607.0.0/src/bo4e_schemas/bo/Vertrag.json"`
375    pub payload_schema: String,
376
377    /// BO4E-typed JSON payload.  `mako-engine` maps this to an internal
378    /// `Command` via the registered domain command mapper.
379    pub payload: serde_json::Value,
380}
381
382// ── ErpCommandSource trait ────────────────────────────────────────────────────
383
384/// Inbound command source — `mako-engine` polls this for new BO4E objects
385/// from the ERP.
386///
387/// Implement this for broker-based inbound flows (Kafka consumer, SFTP poll,
388/// database change feed, …) to make the entire integration fully event-driven
389/// — no synchronous REST round-trip required.
390///
391/// ## Contract
392///
393/// - [`next`](ErpCommandSource::next) must be **non-blocking** when idle —
394///   return `Ok(None)` immediately when no command is available.
395/// - [`ack`](ErpCommandSource::ack) must suppress re-delivery of `id` after
396///   a successful ack (idempotent).
397/// - [`nack`](ErpCommandSource::nack) should allow re-delivery of `id` after
398///   an appropriate backoff.
399#[allow(async_fn_in_trait)]
400pub trait ErpCommandSource: Send + Sync + 'static {
401    /// Return the next pending BO4E command, or `None` when the source is idle.
402    async fn next(&self) -> Result<Option<InboundErpCommand>, ErpAdapterError>;
403
404    /// Acknowledge successful processing of `id`.
405    ///
406    /// After a successful ack the source must not re-deliver `id`.
407    async fn ack(&self, id: &str) -> Result<(), ErpAdapterError>;
408
409    /// Negative-acknowledge — allow re-delivery of `id` after backoff.
410    async fn nack(&self, id: &str, reason: &str) -> Result<(), ErpAdapterError>;
411}
412
413/// Blanket `Arc` implementation so `ErpCommandSource` can be shared across tasks.
414impl<S: ErpCommandSource> ErpCommandSource for Arc<S> {
415    async fn next(&self) -> Result<Option<InboundErpCommand>, ErpAdapterError> {
416        (**self).next().await
417    }
418    async fn ack(&self, id: &str) -> Result<(), ErpAdapterError> {
419        (**self).ack(id).await
420    }
421    async fn nack(&self, id: &str, reason: &str) -> Result<(), ErpAdapterError> {
422        (**self).nack(id, reason).await
423    }
424}
425
426// ── NoopErpAdapter ────────────────────────────────────────────────────────────
427
428/// An [`ErpAdapter`] that succeeds immediately without notifying anything.
429///
430/// Use in unit tests and CI where no real ERP endpoint is available.
431#[cfg(feature = "testing")]
432#[derive(Debug, Clone, Default)]
433pub struct NoopErpAdapter;
434
435#[cfg(feature = "testing")]
436impl ErpAdapter for NoopErpAdapter {
437    async fn notify(&self, _event: ErpEvent) -> Result<(), ErpAdapterError> {
438        Ok(())
439    }
440}
441
442// ── LogErpAdapter ─────────────────────────────────────────────────────────────
443
444/// An [`ErpAdapter`] that logs every event at `info` level without delivering
445/// it.
446///
447/// Useful as a development starting point — replace it with your concrete ERP
448/// adapter in production.
449#[derive(Debug, Clone, Default)]
450pub struct LogErpAdapter;
451
452impl ErpAdapter for LogErpAdapter {
453    async fn notify(&self, event: ErpEvent) -> Result<(), ErpAdapterError> {
454        tracing::info!(
455            idempotency_key = %event.idempotency_key,
456            event_type      = event.event_type.label(),
457            process_id      = %event.process_id,
458            tenant_id       = %event.tenant_id,
459            pid             = event.pid,
460            "ErpAdapter: event logged (no delivery configured)",
461        );
462        Ok(())
463    }
464}
465
466// ── NoopErpCommandSource ──────────────────────────────────────────────────────
467
468/// An [`ErpCommandSource`] that is always idle (returns `Ok(None)`).
469///
470/// Use in tests where no inbound ERP command flow is needed.
471#[cfg(feature = "testing")]
472#[derive(Debug, Clone, Default)]
473pub struct NoopErpCommandSource;
474
475#[cfg(feature = "testing")]
476impl ErpCommandSource for NoopErpCommandSource {
477    async fn next(&self) -> Result<Option<InboundErpCommand>, ErpAdapterError> {
478        Ok(None)
479    }
480    async fn ack(&self, _id: &str) -> Result<(), ErpAdapterError> {
481        Ok(())
482    }
483    async fn nack(&self, _id: &str, _reason: &str) -> Result<(), ErpAdapterError> {
484        Ok(())
485    }
486}
487
488// ── ErpAdapterTestHarness ─────────────────────────────────────────────────────
489
490/// A recording [`ErpAdapter`] for use in tests.
491///
492/// Records every [`ErpEvent`] delivered via [`notify`](ErpAdapter::notify) so
493/// tests can assert on event types, ordering, and BO4E payload shapes.
494///
495/// ```rust,ignore
496/// let harness = ErpAdapterTestHarness::new();
497/// my_workflow.run_with_adapter(harness.adapter()).await?;
498///
499/// let events = harness.events();
500/// assert_eq!(events[0].event_type, ErpEventType::ProcessInitiated);
501/// assert_eq!(events[1].event_type, ErpEventType::AperakAccepted);
502/// ```
503#[cfg(feature = "testing")]
504#[derive(Debug, Clone, Default)]
505pub struct ErpAdapterTestHarness {
506    events: Arc<tokio::sync::Mutex<Vec<ErpEvent>>>,
507}
508
509#[cfg(feature = "testing")]
510impl ErpAdapterTestHarness {
511    /// Create a new empty harness.
512    #[must_use]
513    pub fn new() -> Self {
514        Self::default()
515    }
516
517    /// Return a snapshot of all recorded events in delivery order.
518    pub async fn events(&self) -> Vec<ErpEvent> {
519        self.events.lock().await.clone()
520    }
521
522    /// Drain all recorded events, resetting the harness.
523    pub async fn drain(&self) -> Vec<ErpEvent> {
524        std::mem::take(&mut *self.events.lock().await)
525    }
526}
527
528#[cfg(feature = "testing")]
529impl ErpAdapter for ErpAdapterTestHarness {
530    async fn notify(&self, event: ErpEvent) -> Result<(), ErpAdapterError> {
531        self.events.lock().await.push(event);
532        Ok(())
533    }
534}
535
536// ── ErpCommandSourceTestHarness ───────────────────────────────────────────────
537
538/// A controllable [`ErpCommandSource`] for use in tests.
539///
540/// Inject canned [`InboundErpCommand`] payloads and verify that the engine
541/// processes them correctly.
542///
543/// ```text
544/// let source = ErpCommandSourceTestHarness::new();
545/// source.inject(InboundErpCommand {
546///     idempotency_key: "order-42".into(),
547///     tenant_id: TenantId::new(),
548///     payload_schema: ".../Vertrag.json".into(),
549///     payload: serde_json::json!({ "_typ": "VERTRAG", ... }),
550/// }).await;
551///
552/// // The engine picks up the command on the next poll.
553/// ```
554#[cfg(feature = "testing")]
555#[derive(Debug, Clone, Default)]
556pub struct ErpCommandSourceTestHarness {
557    queue: Arc<tokio::sync::Mutex<std::collections::VecDeque<InboundErpCommand>>>,
558    acked: Arc<tokio::sync::Mutex<Vec<String>>>,
559    nacked: Arc<tokio::sync::Mutex<Vec<(String, String)>>>,
560}
561
562#[cfg(feature = "testing")]
563impl ErpCommandSourceTestHarness {
564    /// Create a new empty harness.
565    #[must_use]
566    pub fn new() -> Self {
567        Self::default()
568    }
569
570    /// Enqueue a command to be returned by the next [`next`](ErpCommandSource::next) call.
571    pub async fn inject(&self, cmd: InboundErpCommand) {
572        self.queue.lock().await.push_back(cmd);
573    }
574
575    /// Return all acked command IDs.
576    pub async fn acked(&self) -> Vec<String> {
577        self.acked.lock().await.clone()
578    }
579
580    /// Return all nacked `(id, reason)` pairs.
581    pub async fn nacked(&self) -> Vec<(String, String)> {
582        self.nacked.lock().await.clone()
583    }
584}
585
586#[cfg(feature = "testing")]
587impl ErpCommandSource for ErpCommandSourceTestHarness {
588    async fn next(&self) -> Result<Option<InboundErpCommand>, ErpAdapterError> {
589        Ok(self.queue.lock().await.pop_front())
590    }
591
592    async fn ack(&self, id: &str) -> Result<(), ErpAdapterError> {
593        self.acked.lock().await.push(id.to_owned());
594        Ok(())
595    }
596
597    async fn nack(&self, id: &str, reason: &str) -> Result<(), ErpAdapterError> {
598        self.nacked
599            .lock()
600            .await
601            .push((id.to_owned(), reason.to_owned()));
602        Ok(())
603    }
604}
605
606// ── BO4E schema URL constants ─────────────────────────────────────────────────
607
608/// BO4E schema URL base for v202607.0.0.
609///
610/// Use `bo4e_schema_url!(Marktlokation)` to construct typed schema URLs at
611/// compile time.
612pub const BO4E_V202607_BASE: &str =
613    "https://raw.githubusercontent.com/BO4E/BO4E-Schemas/v202607.0.0/src/bo4e_schemas";
614
615/// Construct a BO4E v202607.0.0 JSON Schema URL for a Business Object.
616///
617/// ```rust
618/// use mako_engine::bo4e_schema_url;
619/// assert!(bo4e_schema_url!("bo", "Marktlokation").contains("Marktlokation"));
620/// ```
621#[macro_export]
622macro_rules! bo4e_schema_url {
623    ($category:literal, $name:literal) => {
624        concat!(
625            "https://raw.githubusercontent.com/BO4E/BO4E-Schemas/v202607.0.0/src/bo4e_schemas/",
626            $category,
627            "/",
628            $name,
629            ".json",
630        )
631    };
632}
633
634/// BO4E JSON Schema URL for `Marktlokation`.
635pub const BO4E_SCHEMA_MARKTLOKATION: &str = bo4e_schema_url!("bo", "Marktlokation");
636
637/// BO4E JSON Schema URL for `Messlokation`.
638pub const BO4E_SCHEMA_MESSLOKATION: &str = bo4e_schema_url!("bo", "Messlokation");
639
640/// BO4E JSON Schema URL for `Vertrag`.
641pub const BO4E_SCHEMA_VERTRAG: &str = bo4e_schema_url!("bo", "Vertrag");
642
643/// BO4E JSON Schema URL for `Energiemenge`.
644pub const BO4E_SCHEMA_ENERGIEMENGE: &str = bo4e_schema_url!("bo", "Energiemenge");
645
646/// BO4E JSON Schema URL for `Rechnung`.
647pub const BO4E_SCHEMA_RECHNUNG: &str = bo4e_schema_url!("bo", "Rechnung");
648
649/// BO4E JSON Schema URL for `Zaehler`.
650pub const BO4E_SCHEMA_ZAEHLER: &str = bo4e_schema_url!("bo", "Zaehler");
651
652/// BO4E JSON Schema URL for `Geschaeftspartner`.
653pub const BO4E_SCHEMA_GESCHAEFTSPARTNER: &str = bo4e_schema_url!("bo", "Geschaeftspartner");