#![forbid(unsafe_code)]
use rust_decimal::RoundingStrategy;
use std::collections::HashMap;
use std::marker::PhantomData;
use mako_engine::types::Pruefidentifikator;
use mako_engine::{
envelope::EventEnvelope,
error::WorkflowError,
ids::DeadlineId,
outbox::PendingOutbox,
projection::Projection,
types::{MarktpartnerCode, MessageRef},
workflow::{CommandPayload, EventPayload, Workflow, WorkflowOutput},
};
use rubo4e::current::Rechnung;
pub const REMADV_PIDS: &[u32] = &[33001, 33002, 33003, 33004];
pub const REMADV_CONFIRMATION_PID: u32 = 33001;
pub const ZAHLUNGSAVIS_PID: u32 = REMADV_CONFIRMATION_PID;
pub const ABWEISUNG_PID: u32 = 33002;
pub const COMDIS_ABLEHNUNG_PID: Pruefidentifikator = Pruefidentifikator::const_new(29001);
#[must_use]
pub fn remadv_confirms(pid: Pruefidentifikator) -> bool {
pid.as_u32() == REMADV_CONFIRMATION_PID
}
pub trait InvoicFamily: Send + Sync + 'static {
const WORKFLOW_NAME: &'static str;
const DEADLINE_LABEL: &'static str;
const INVOIC_PIDS: &'static [u32];
const SENDS_INVOIC: bool;
const ANSWERS_COMDIS: bool;
#[must_use]
fn pid_hint() -> String {
Self::INVOIC_PIDS
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("/")
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct InvoicData {
pub pruefidentifikator: Pruefidentifikator,
pub sender: MarktpartnerCode,
pub recipient: MarktpartnerCode,
pub document_date: String,
pub invoice_ref: MessageRef,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rechnung: Option<Box<Rechnung>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bestellung_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rechnungstyp: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RemadvBefund {
pub code: String,
pub ebene: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub positionsnummer: Option<u16>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub detail: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RemadvAntwort {
pub ebd: String,
pub befunde: Vec<RemadvBefund>,
pub remadv_pid: u32,
}
impl RemadvAntwort {
#[must_use]
pub fn erster_code(&self) -> Option<&str> {
self.befunde.first().map(|b| b.code.as_str())
}
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(tag = "status", content = "data")]
pub enum InvoicState {
#[default]
New,
InvoicReceived(InvoicData),
ValidationPassed(InvoicData),
Settled(InvoicData),
Disputed {
data: InvoicData,
reason: String,
},
Rejected {
reason: String,
},
InvoicSent(InvoicData),
PaymentConfirmed(InvoicData),
PaymentDisputed {
data: InvoicData,
remadv_pid: Pruefidentifikator,
},
ComdisRejected(InvoicData),
}
impl InvoicState {
#[must_use]
pub fn label(&self) -> &'static str {
match self {
Self::New => "New",
Self::InvoicReceived(_) => "InvoicReceived",
Self::ValidationPassed(_) => "ValidationPassed",
Self::Settled(_) => "Settled",
Self::Disputed { .. } => "Disputed",
Self::Rejected { .. } => "Rejected",
Self::InvoicSent(_) => "InvoicSent",
Self::PaymentConfirmed(_) => "PaymentConfirmed",
Self::PaymentDisputed { .. } => "PaymentDisputed",
Self::ComdisRejected(_) => "ComdisRejected",
}
}
#[must_use]
pub fn data(&self) -> Option<&InvoicData> {
match self {
Self::InvoicReceived(d)
| Self::ValidationPassed(d)
| Self::Settled(d)
| Self::InvoicSent(d)
| Self::PaymentConfirmed(d)
| Self::ComdisRejected(d) => Some(d),
Self::Disputed { data, .. } | Self::PaymentDisputed { data, .. } => Some(data),
Self::New | Self::Rejected { .. } => None,
}
}
#[must_use]
pub const fn is_terminal(&self) -> bool {
matches!(
self,
Self::Settled(_)
| Self::Disputed { .. }
| Self::Rejected { .. }
| Self::PaymentConfirmed(_)
| Self::PaymentDisputed { .. }
| Self::ComdisRejected(_)
)
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", content = "data")]
pub enum InvoicEvent {
InvoicReceived {
invoice_ref: MessageRef,
sender: MarktpartnerCode,
recipient: MarktpartnerCode,
document_date: String,
pruefidentifikator: Pruefidentifikator,
#[serde(default, skip_serializing_if = "Option::is_none")]
rechnung: Option<Box<Rechnung>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
bestellung_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
rechnungstyp: Option<String>,
},
ValidationPassed {
invoice_ref: MessageRef,
},
InvoiceSettled,
InvoiceDisputed {
reason: String,
},
Rejected {
reason: String,
},
DeadlineExpired {
deadline_id: DeadlineId,
label: Box<str>,
},
InvoicSent {
pruefidentifikator: Pruefidentifikator,
sender: MarktpartnerCode,
recipient: MarktpartnerCode,
document_date: String,
invoice_ref: MessageRef,
},
RemadvReceived {
pid: Pruefidentifikator,
remadv_ref: MessageRef,
sender: MarktpartnerCode,
is_confirmed: bool,
},
ComdisAbLehnungReceived {
comdis_ref: MessageRef,
},
}
impl EventPayload for InvoicEvent {
fn event_type(&self) -> &'static str {
match self {
Self::InvoicReceived { .. } => "InvoicReceived",
Self::ValidationPassed { .. } => "InvoicValidationPassed",
Self::InvoiceSettled => "InvoiceSettled",
Self::InvoiceDisputed { .. } => "InvoiceDisputed",
Self::Rejected { .. } => "InvoicRejected",
Self::DeadlineExpired { .. } => "InvoicDeadlineExpired",
Self::InvoicSent { .. } => "InvoicSent",
Self::RemadvReceived { .. } => "RemadvReceived",
Self::ComdisAbLehnungReceived { .. } => "ComdisAblehnungReceived",
}
}
}
#[derive(Clone)]
pub enum InvoicCommand {
ReceiveInvoic {
pid: Pruefidentifikator,
sender: MarktpartnerCode,
recipient: MarktpartnerCode,
invoice_ref: MessageRef,
document_date: String,
validation_passed: bool,
validation_errors: Vec<String>,
rechnung: Option<Box<Rechnung>>,
bestellung_ref: Option<String>,
rechnungstyp: Option<String>,
},
SendInvoic {
pid: Pruefidentifikator,
sender: MarktpartnerCode,
recipient: MarktpartnerCode,
document_date: String,
invoice_ref: MessageRef,
},
ReceiveRemadv {
pid: Pruefidentifikator,
remadv_ref: MessageRef,
sender: MarktpartnerCode,
},
ReceiveComdis {
comdis_ref: MessageRef,
},
SettleInvoice {
message_ref: MessageRef,
},
DisputeInvoice {
message_ref: MessageRef,
reason: String,
#[allow(clippy::struct_field_names)]
antwort: Option<RemadvAntwort>,
},
TimeoutExpired {
deadline_id: DeadlineId,
label: Box<str>,
},
}
impl CommandPayload for InvoicCommand {}
pub struct InvoicWorkflow<F: InvoicFamily>(PhantomData<fn() -> F>);
impl<F: InvoicFamily> Workflow for InvoicWorkflow<F> {
type State = InvoicState;
type Event = InvoicEvent;
type Command = InvoicCommand;
fn on_deadline(
deadline: &mako_engine::deadline::Deadline,
state: &Self::State,
) -> Option<Self::Command> {
match (deadline.label(), state) {
(label, InvoicState::InvoicReceived(_) | InvoicState::ValidationPassed(_))
if label == F::DEADLINE_LABEL =>
{
Some(InvoicCommand::TimeoutExpired {
deadline_id: deadline.deadline_id(),
label: deadline.label().into(),
})
}
_ => None,
}
}
fn apply(state: Self::State, event: &Self::Event) -> Self::State {
match event {
InvoicEvent::InvoicReceived {
invoice_ref,
sender,
recipient,
document_date,
pruefidentifikator,
rechnung,
bestellung_ref,
rechnungstyp,
} => InvoicState::InvoicReceived(InvoicData {
pruefidentifikator: *pruefidentifikator,
sender: sender.clone(),
recipient: recipient.clone(),
document_date: document_date.clone(),
invoice_ref: invoice_ref.clone(),
rechnung: rechnung.clone(),
bestellung_ref: bestellung_ref.clone(),
rechnungstyp: rechnungstyp.clone(),
}),
InvoicEvent::ValidationPassed { .. } => match state {
InvoicState::InvoicReceived(data) => InvoicState::ValidationPassed(data),
other => other,
},
InvoicEvent::InvoiceSettled => match state {
InvoicState::ComdisRejected(data) => InvoicState::Settled(data),
InvoicState::ValidationPassed(data) => InvoicState::Settled(data),
other => other,
},
InvoicEvent::InvoiceDisputed { reason } => match state {
InvoicState::ComdisRejected(data) => InvoicState::Disputed {
data,
reason: reason.clone(),
},
InvoicState::ValidationPassed(data) => InvoicState::Disputed {
data,
reason: reason.clone(),
},
other => other,
},
InvoicEvent::Rejected { reason } => InvoicState::Rejected {
reason: reason.clone(),
},
InvoicEvent::DeadlineExpired { label, .. } => {
if state.is_terminal() {
state
} else {
InvoicState::Rejected {
reason: format!("settlement deadline expired: {label}"),
}
}
}
InvoicEvent::InvoicSent {
pruefidentifikator,
sender,
recipient,
document_date,
invoice_ref,
} => InvoicState::InvoicSent(InvoicData {
pruefidentifikator: *pruefidentifikator,
sender: sender.clone(),
recipient: recipient.clone(),
document_date: document_date.clone(),
invoice_ref: invoice_ref.clone(),
rechnung: None,
bestellung_ref: None,
rechnungstyp: None,
}),
InvoicEvent::RemadvReceived {
pid, is_confirmed, ..
} => match state {
InvoicState::InvoicSent(data) => {
if *is_confirmed {
InvoicState::PaymentConfirmed(data)
} else {
InvoicState::PaymentDisputed {
remadv_pid: *pid,
data,
}
}
}
other => other,
},
InvoicEvent::ComdisAbLehnungReceived { .. } => match state {
InvoicState::ValidationPassed(data)
| InvoicState::Settled(data)
| InvoicState::Disputed { data, .. } => InvoicState::ComdisRejected(data),
other => other,
},
}
}
fn handle(
state: &Self::State,
command: Self::Command,
) -> Result<WorkflowOutput<Self::Event>, WorkflowError> {
match command {
InvoicCommand::ReceiveInvoic {
pid,
sender,
recipient,
invoice_ref,
document_date,
validation_passed,
validation_errors,
rechnung,
bestellung_ref,
rechnungstyp,
} => {
if !matches!(state, InvoicState::New) {
return Err(WorkflowError::invalid_state("New", state.label()));
}
if !F::INVOIC_PIDS.contains(&pid.as_u32()) {
return Err(WorkflowError::rejected(format!(
"expected an INVOIC PID for {} ({}), got {pid}",
F::WORKFLOW_NAME,
F::pid_hint(),
)));
}
let mut events = vec![InvoicEvent::InvoicReceived {
invoice_ref: invoice_ref.clone(),
sender: sender.clone(),
recipient: recipient.clone(),
document_date,
pruefidentifikator: pid,
rechnung: rechnung.clone(),
bestellung_ref: bestellung_ref.clone(),
rechnungstyp: rechnungstyp.clone(),
}];
let mut outbox: Vec<PendingOutbox> = Vec::new();
if validation_passed {
events.push(InvoicEvent::ValidationPassed {
invoice_ref: invoice_ref.clone(),
});
outbox.push(
PendingOutbox::new(
"ProcessInitiated",
recipient.as_str(),
serde_json::json!({
"pid": pid.as_u32(),
"invoice_ref": invoice_ref.as_str(),
"sender_mp_id": sender.as_str(),
"workflow": F::WORKFLOW_NAME,
"rechnung": serde_json::to_value(rechnung.as_deref())
.unwrap_or(serde_json::Value::Null),
"bestellung_ref": bestellung_ref,
"rechnungstyp": rechnungstyp,
}),
)
.caused_by(1),
);
} else {
events.push(InvoicEvent::Rejected {
reason: validation_errors.join("; "),
});
}
Ok(WorkflowOutput::with_outbox(events, outbox))
}
InvoicCommand::SettleInvoice { message_ref } => {
if !answerable(state) {
return Err(WorkflowError::invalid_state(
"ValidationPassed|ComdisRejected",
state.label(),
));
}
Ok(WorkflowOutput::with_outbox(
vec![InvoicEvent::InvoiceSettled],
vec![
remadv_outbox(state, ZAHLUNGSAVIS_PID, &message_ref, None, None),
completion_outbox::<F>(state, "settled", None),
],
))
}
InvoicCommand::DisputeInvoice {
message_ref,
reason,
antwort,
} => {
if !answerable(state) {
return Err(WorkflowError::invalid_state(
"ValidationPassed|ComdisRejected",
state.label(),
));
}
let pid = antwort.as_ref().map_or(ABWEISUNG_PID, |a| a.remadv_pid);
let outbox = vec![
remadv_outbox(state, pid, &message_ref, Some(&reason), antwort.as_ref()),
completion_outbox::<F>(state, "disputed", Some(&reason)),
];
Ok(WorkflowOutput::with_outbox(
vec![InvoicEvent::InvoiceDisputed { reason }],
outbox,
))
}
InvoicCommand::TimeoutExpired { deadline_id, label } => {
if state.is_terminal() {
return Ok(WorkflowOutput::events(vec![]));
}
Ok(vec![InvoicEvent::DeadlineExpired { deadline_id, label }].into())
}
InvoicCommand::SendInvoic {
pid,
sender,
recipient,
document_date,
invoice_ref,
} => {
if !F::SENDS_INVOIC {
return Err(WorkflowError::rejected(format!(
"{} does not play the issuer role — it receives invoices only",
F::WORKFLOW_NAME,
)));
}
if !matches!(state, InvoicState::New) {
return Err(WorkflowError::invalid_state("New", state.label()));
}
if !F::INVOIC_PIDS.contains(&pid.as_u32()) {
return Err(WorkflowError::rejected(format!(
"expected an INVOIC PID for {} ({}), got {pid}",
F::WORKFLOW_NAME,
F::pid_hint(),
)));
}
Ok(vec![InvoicEvent::InvoicSent {
pruefidentifikator: pid,
sender,
recipient,
document_date,
invoice_ref,
}]
.into())
}
InvoicCommand::ReceiveRemadv {
pid,
remadv_ref,
sender,
} => {
if !F::SENDS_INVOIC {
return Err(WorkflowError::rejected(format!(
"{} never issues an invoice, so no REMADV can answer one",
F::WORKFLOW_NAME,
)));
}
if !matches!(state, InvoicState::InvoicSent(_)) {
return Err(WorkflowError::invalid_state("InvoicSent", state.label()));
}
if !REMADV_PIDS.contains(&pid.as_u32()) {
return Err(WorkflowError::rejected(format!(
"expected a REMADV PID (33001–33004), got {pid}",
)));
}
let is_confirmed = remadv_confirms(pid);
Ok(vec![InvoicEvent::RemadvReceived {
pid,
remadv_ref,
sender,
is_confirmed,
}]
.into())
}
InvoicCommand::ReceiveComdis { comdis_ref } => {
if !F::ANSWERS_COMDIS {
return Err(WorkflowError::rejected(format!(
"{} does not exchange COMDIS 29001",
F::WORKFLOW_NAME,
)));
}
if !matches!(
state,
InvoicState::ValidationPassed(_)
| InvoicState::Settled(_)
| InvoicState::Disputed { .. }
) {
return Err(WorkflowError::invalid_state(
"ValidationPassed|Settled|Disputed",
state.label(),
));
}
Ok(vec![InvoicEvent::ComdisAbLehnungReceived { comdis_ref }].into())
}
}
}
}
const fn answerable(state: &InvoicState) -> bool {
matches!(
state,
InvoicState::ValidationPassed(_) | InvoicState::ComdisRejected(_)
)
}
fn remadv_outbox(
state: &InvoicState,
pid: u32,
message_ref: &MessageRef,
reason: Option<&str>,
antwort: Option<&RemadvAntwort>,
) -> PendingOutbox {
let data = state.data();
let issuer = data.map(|d| d.sender.as_str()).unwrap_or_default();
let mut payload = serde_json::json!({
"pid": pid,
"sender": data.map(|d| d.recipient.as_str()).unwrap_or_default(),
"receiver": issuer,
"message_ref": message_ref.as_str(),
"document_code": if pid == ZAHLUNGSAVIS_PID { "481" } else { "239" },
"invoice_ref": data.map(|d| d.invoice_ref.to_string()).unwrap_or_default(),
"document_date": data.map(|d| d.document_date.clone()).unwrap_or_default(),
});
let Some(obj) = payload.as_object_mut() else {
return PendingOutbox::new("REMADV", issuer, payload);
};
if let Some(r) = data.and_then(|d| d.rechnung.as_deref()) {
let faellig = r
.zu_zahlen
.as_ref()
.or(r.gesamtbrutto.as_ref())
.and_then(|b| b.wert)
.unwrap_or_default();
let gutschrift = r.ist_storno == Some(true);
let ueberweisung = if pid == ZAHLUNGSAVIS_PID {
if gutschrift { -faellig } else { faellig }
} else {
rust_decimal::Decimal::ZERO
};
obj.insert(
"rechnungsbezug".to_owned(),
serde_json::json!({
"dokumentenart": match (gutschrift, r.ist_original == Some(false)) {
(true, true) => "Z25",
(true, false) => "457",
(false, true) => "389",
(false, false) => "380",
},
"rechnungsnummer": r.rechnungsnummer.clone().unwrap_or_default(),
"faelliger_betrag": faellig.round_dp_with_strategy(2, RoundingStrategy::MidpointAwayFromZero).to_string(),
"ueberweisungsbetrag": ueberweisung.round_dp_with_strategy(2, RoundingStrategy::MidpointAwayFromZero).to_string(),
"rechnungsdatum": r
.rechnungsdatum
.map(|d| d.date().to_string())
.unwrap_or_default(),
}),
);
}
if let Some(reason) = reason {
obj.insert("ablehnungsgrund".to_owned(), serde_json::json!(reason));
}
if let Some(a) = antwort {
obj.insert(
"antwort_code".to_owned(),
serde_json::json!(a.erster_code()),
);
obj.insert("antwort_codeliste".to_owned(), serde_json::json!(a.ebd));
obj.insert("antwort_befunde".to_owned(), serde_json::json!(a.befunde));
}
PendingOutbox::new("REMADV", issuer, payload)
}
fn completion_outbox<F: InvoicFamily>(
state: &InvoicState,
outcome: &str,
reason: Option<&str>,
) -> PendingOutbox {
let data = state.data();
let mut payload = serde_json::json!({
"pid": data.map_or(0, |d| d.pruefidentifikator.as_u32()),
"invoice_ref": data.map(|d| d.invoice_ref.to_string()).unwrap_or_default(),
"workflow": F::WORKFLOW_NAME,
"outcome": outcome,
});
if let Some(reason) = reason
&& let Some(obj) = payload.as_object_mut()
{
obj.insert("reason".to_owned(), serde_json::json!(reason));
}
PendingOutbox::new("ProcessCompleted", "", payload)
}
#[derive(Debug)]
pub struct InvoicRecord {
pub status: &'static str,
pub pruefidentifikator: Option<Pruefidentifikator>,
pub event_count: usize,
}
impl Default for InvoicRecord {
fn default() -> Self {
Self {
status: "New",
pruefidentifikator: None,
event_count: 0,
}
}
}
#[derive(Debug, Default)]
pub struct InvoicProjection {
pub records: HashMap<String, InvoicRecord>,
pub last_seq: u64,
}
impl Projection for InvoicProjection {
fn name(&self) -> &'static str {
"InvoicProjection"
}
fn handle_event(&mut self, envelope: &EventEnvelope) {
self.last_seq = self.last_seq.max(envelope.sequence_number);
let record = self
.records
.entry(envelope.stream_id.as_str().to_owned())
.or_default();
record.event_count += 1;
let Ok(event) = envelope.decode::<InvoicEvent>() else {
return;
};
match event {
InvoicEvent::InvoicReceived {
pruefidentifikator, ..
} => {
record.status = "InvoicReceived";
record.pruefidentifikator = Some(pruefidentifikator);
}
InvoicEvent::ValidationPassed { .. } => record.status = "ValidationPassed",
InvoicEvent::InvoiceSettled => record.status = "Settled",
InvoicEvent::InvoiceDisputed { .. } => record.status = "Disputed",
InvoicEvent::Rejected { .. } | InvoicEvent::DeadlineExpired { .. } => {
record.status = "Rejected";
}
InvoicEvent::InvoicSent {
pruefidentifikator, ..
} => {
record.status = "InvoicSent";
record.pruefidentifikator = Some(pruefidentifikator);
}
InvoicEvent::RemadvReceived { is_confirmed, .. } => {
record.status = if is_confirmed {
"PaymentConfirmed"
} else {
"PaymentDisputed"
};
}
InvoicEvent::ComdisAbLehnungReceived { .. } => record.status = "ComdisRejected",
}
}
fn last_sequence(&self) -> Option<u64> {
if self.last_seq == 0 {
None
} else {
Some(self.last_seq)
}
}
}