arminius_lib_message_bus/

command.rs

// SPDX-FileCopyrightText: 2025 BWI GmbH
//
// SPDX-License-Identifier: Apache-2.0

//! Request-reply command messaging: traits for sending commands and awaiting
//! typed replies, and for responding to received commands.

use std::{borrow::Cow, future::Future, ops::Deref};

pub use futures::Stream;
use serde::{Serialize, de::DeserializeOwned};

use crate::event::Error;

/// Errors during command payload serialization or subject parsing.
#[derive(Debug, thiserror::Error)]
pub enum CommandError {
	/// JSON serialization or deserialization failure.
	#[error("Failed to serialize/deserialize payload: {0}")]
	Payload(#[from] serde_json::Error),
	/// Base64url decoding of a subject segment failed.
	#[error("Failed to decode subject: {0}")]
	SubjectDecode(String),
	/// Parsed string could not be converted to the expected type.
	#[error("Failed to parse subject: {0}")]
	SubjectParse(String),
}

/// How often a command may reach its handler.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Delivery {
	/// Redelivered until it is acknowledged, so handling it twice must be
	/// harmless.
	AtLeastOnce,
	/// Acknowledged before it is handled and never redelivered, so a handler
	/// that does not finish loses the command.
	AtMostOnce,
}

/// Typed command endpoint for request-reply over a message bus.
///
/// Each struct represents one command endpoint.
pub trait Command: Serialize + DeserializeOwned + Send + Sync {
	/// NATS subject pattern, e.g. `"rooms.*.create"`.
	const SUBJECT: &'static str;
	/// JetStream stream filter (auto-inferred by derive), e.g. `"rooms.>"`.
	const SUBJECT_FILTER: &'static str;
	/// How often this command may reach its handler.
	///
	/// Defaults to [`Delivery::AtLeastOnce`], which suits any handler that can
	/// run twice without harm.
	const DELIVERY: Delivery = Delivery::AtLeastOnce;
	/// Wire response type — `Requester::request` returns `Self::Reply`.
	type Reply: Serialize + DeserializeOwned + Send;

	/// The concrete subject *this instance* is sent to.
	///
	/// Encodes any wildcard fields of [`Command::SUBJECT`]. Defaults to the
	/// static `SUBJECT` for commands without wildcard fields.
	fn subject(&self) -> Cow<'static, str> {
		Cow::Borrowed(Self::SUBJECT)
	}

	/// Serialize this command into the transport payload.
	///
	/// The wire format is owned here, not by the transport, so the encoding can
	/// change (e.g. JSON to protobuf) without the bus implementation knowing.
	/// The full struct is encoded, including any fields also carried in the
	/// subject (so the two copies cannot drift); the reply address travels
	/// out-of-band (e.g. a NATS header), not in these bytes.
	fn into_payload(self) -> Result<Vec<u8>, CommandError> {
		Ok(serde_json::to_vec(&self)?)
	}

	/// Reconstruct `Self` from a received command's `payload`.
	///
	/// The full struct is serialized on the wire (including any fields also
	/// encoded in the subject, so the two copies cannot drift), and the reply
	/// address is carried out-of-band by the transport (e.g. a NATS header).
	/// The payload therefore suffices to recover every field; `subject` is
	/// retained for transports that wish to validate routing.
	fn from_payload(subject: &str, payload: &[u8]) -> Result<Self, CommandError> {
		let _ = subject;
		Ok(serde_json::from_slice(payload)?)
	}
}

/// A received command that must be acknowledged with a typed reply.
pub trait ReceivedCommand<C: Command>: Deref<Target = C> {
	/// Returns a reference to the received command data.
	fn command(&self) -> &C {
		self
	}

	/// Take responsibility for the command without answering it yet.
	///
	/// A handler that acknowledges up front trades the certainty of
	/// eventually receiving an answer for the certainty that nothing can hand
	/// the command out a second time: a crash or a stall past this point costs
	/// the reply rather than repeating whatever the handler had already done.
	/// This is what a [`Delivery::AtMostOnce`] command asks for:
	///
	/// ```ignore
	/// received.ack().await?;
	/// let reply = claim_keys(received.command()).await;
	/// received.reply(reply).await?;
	/// ```
	///
	/// [`reply`](Self::reply) acknowledges as well, so a handler that is happy
	/// to be run again should never call this.
	fn ack(&self) -> impl Future<Output = Result<(), Error>> + Send;

	/// Send `response` back to the requester and acknowledge the command.
	fn reply(&self, response: C::Reply) -> impl Future<Output = Result<(), Error>> + Send;

	/// Returns the OpenTelemetry context propagated with this command.
	fn context(&self) -> &opentelemetry::Context;
}

/// Sends commands and awaits typed replies. Mirrors
/// [`Publisher`](crate::Publisher).
pub trait Requester {
	/// Send `cmd` and wait for a typed reply.
	fn request<C: Command>(&self, cmd: C) -> impl Future<Output = Result<C::Reply, Error>> + Send;
}

impl<T: Requester + Sync> Requester for std::sync::Arc<T> {
	fn request<C: Command>(&self, cmd: C) -> impl Future<Output = Result<C::Reply, Error>> + Send {
		(**self).request(cmd)
	}
}

/// Subscribes to incoming commands. Mirrors [`Subscriber`](crate::Subscriber).
///
/// `respond` takes `self` by value so the returned future is `'static` with no
/// borrow of the implementor. Callers that need to keep a handle should clone
/// before calling (e.g. `conn.clone().respond::<C>()`).
pub trait Responder {
	/// Subscribe to commands of type `C` and return a stream of received
	/// commands.
	fn respond<C: Command + 'static>(
		self,
	) -> impl Future<
		Output = Result<
			impl Stream<Item = Result<impl ReceivedCommand<C> + Send + Sync + 'static, Error>>
			+ Send
			+ 'static,
			Error,
		>,
	> + Send
	+ 'static;
}

impl<T: Responder + Clone + Send + Sync + 'static> Responder for std::sync::Arc<T> {
	fn respond<C: Command + 'static>(
		self,
	) -> impl Future<
		Output = Result<
			impl Stream<Item = Result<impl ReceivedCommand<C> + Send + Sync + 'static, Error>>
			+ Send
			+ 'static,
			Error,
		>,
	> + Send
	+ 'static {
		let inner = (*self).clone();
		async move { inner.respond::<C>().await }
	}
}