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");