arminius_lib_message_bus/

event.rs

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

//! Fire-and-forget event messaging: traits for publishing events and
//! subscribing to them as acknowledgeable streams.

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

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

/// Errors that can occur during message bus operations.
#[derive(Debug, thiserror::Error)]
pub enum Error {
	/// JSON serialization or deserialization failure.
	#[error("Failed to serialize/deserialize payload: {0}")]
	Payload(#[from] serde_json::Error),

	/// Failed to publish a message to the specified subject.
	#[error("Failed to send message: {error}")]
	SendFailed {
		/// Underlying error description.
		error: String,
	},

	/// Failed to subscribe to requested subjects.
	#[error("Failed to subscribe to subjects '{subject}': {error}")]
	SubscriptionFailed {
		/// The subject that could not be subscribed to.
		subject: String,
		/// Underlying error description.
		error: String,
	},

	/// Failed while receiving a message from a subscribed subject.
	#[error("Failure while receiving message from subject '{subject}': {error}")]
	ReceptionFailed {
		/// The subject on which the failure occurred.
		subject: String,
		/// Underlying error description.
		error: String,
	},

	/// Failed to acknowledge handling of a message.
	/// The message may be reprocessed by this or another service instance.
	#[error("Failed to acknowledge message: {error}")]
	AckFailed {
		/// Underlying error description.
		error: String,
	},
}

/// Generic event type that can be sent between services.
///
/// Events must be serializable and define a subject (topic) where they will be
/// published. The subject typically includes version information for
/// compatibility handling.
///
/// # Example
///
/// ```rust
/// use arminius_lib_message_bus::Event;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Serialize, Deserialize)]
/// struct RoomCreated {
/// 	room_id: String,
/// }
///
/// impl Event for RoomCreated {
/// 	const SUBJECT: &str = "RoomCreated.v1";
/// }
/// ```
///
/// # Derive macro
///
/// Use `#[derive(Event)]` with a `#[subject(...)]` attribute to avoid the
/// boilerplate:
///
/// ```rust
/// use arminius_lib_message_bus::Event;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Serialize, Deserialize, Event)]
/// #[subject(version = "1")]
/// struct RoomCreated {
/// 	room_id: String,
/// }
/// ```
///
/// Supported `#[subject(...)]` forms:
///
/// 1. `#[subject("MyEvent.v1")]`
/// 2. `#[subject(name = "MyEvent", version = 1)]`
/// 2. `#[subject(name = "MyEvent", version = "1")]`
/// 3. `#[subject(name = "MyEvent", version("1"))]`
/// 4. `#[subject(name = "MyEvent", version = "2", compatible = [v0::MyEvent])]`
///
/// When `name` is omitted the struct name is used. The `compatible` list names
/// older event types whose subjects the subscriber will also consume,
/// converting each to `Self` via the required `From<OldType> for Self`
/// implementation.
///
/// # Per-instance (dynamic) subjects
///
/// A subject template may contain `{field}` placeholders that name fields of
/// the struct. Each placeholder becomes a NATS `*` wildcard in
/// [`Event::SUBJECT`] (the subscribe/stream pattern) and is interpolated via
/// [`Display`] in [`Event::subject`] (the concrete publish subject):
///
/// ```rust
/// use arminius_lib_message_bus::Event;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Serialize, Deserialize, Event)]
/// #[subject(name = "rooms.{room_id}.events", version = "1")]
/// struct RoomCreated {
/// 	room_id: u64,
/// }
/// // RoomCreated::SUBJECT             == "rooms.*.events.v1"   (subscribe to all)
/// // RoomCreated { room_id: 7 }.subject() == "rooms.7.events.v1"   (publish target)
/// ```
///
/// [`Display`]: std::fmt::Display
pub trait Event: Serialize + DeserializeOwned + Send + Sync {
	/// The NATS subject *pattern* this event type lives under.
	///
	/// Used for stream creation and subscription, so it may contain NATS
	/// wildcards (`*`) when the event uses per-instance subjects. For static
	/// events it equals the concrete publish subject. Should include version
	/// information for backward compatibility, e.g., `"RoomCreated.v1"`,
	/// `"RoomCreated.v2"`, `"rooms.*.events.v1"`.
	const SUBJECT: &str;

	/// Subjects of older compatible versions of this event type.
	///
	/// When non-empty, a subscriber will also consume messages on these
	/// subjects and convert them to `Self` via the generated `From` impls.
	/// Populated automatically by `#[subject(version = ..., compatible =
	/// [OldType, ...])]`.
	const COMPATIBLE_SUBJECTS: &'static [&'static str] = &[];

	/// The concrete subject *this instance* is published to.
	///
	/// Defaults to [`Event::SUBJECT`]. Events that use `{field}` placeholders
	/// in their `#[subject(...)]` template get a generated override that
	/// interpolates the named fields, e.g. `rooms.7.events.v1`.
	fn subject(&self) -> std::borrow::Cow<'static, str> {
		std::borrow::Cow::Borrowed(Self::SUBJECT)
	}

	/// Serialize this event 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 publisher knowing.
	fn into_payload(self) -> Result<Vec<u8>, Error> {
		Ok(serde_json::to_vec(&self)?)
	}

	/// Deserialize a raw payload into `Self`, using `subject` to select the
	/// right schema when compatible older versions are present.
	///
	/// The default implementation ignores `subject` and deserializes directly
	/// as `Self`.
	fn from_payload(subject: &str, payload: &[u8]) -> Result<Self, Error> {
		let _ = subject;
		Ok(serde_json::from_slice(payload)?)
	}
}

/// An event received from the message bus that can be acknowledged.
///
/// Implements [`Deref`] to give direct access to the inner event data.
pub trait ReceivedEvent<E: Event>: Deref<Target = E> {
	/// Returns a reference to the received event data.
	fn event(&self) -> &E {
		self
	}

	/// Acknowledge processing of this event.
	///
	/// Tells the message broker that the event was processed successfully and
	/// should not be redelivered. If acknowledgment fails, the event may be
	/// reprocessed by this or another service instance.
	///
	/// # Errors
	///
	/// Returns [`Error::AckFailed`] if the acknowledgment could not be sent.
	fn ack(&self) -> impl Future<Output = Result<(), Error>> + Send;

	/// Returns the OpenTelemetry context propagated with this event.
	///
	/// Use this to continue a distributed trace across the service boundary:
	///
	/// ```rust,no_run
	/// # use arminius_lib_message_bus::ReceivedEvent;
	/// # use opentelemetry::trace::FutureExt;
	/// # async fn process<E: arminius_lib_message_bus::Event + std::fmt::Debug>(
	/// #     event: impl ReceivedEvent<E>
	/// # ) -> Result<(), arminius_lib_message_bus::Error> {
	/// let ctx = event.context().clone();
	/// async move {
	/// 	println!("{:?}", event.event());
	/// 	event.ack().await
	/// }
	/// .with_context(ctx)
	/// .await
	/// # }
	/// ```
	fn context(&self) -> &opentelemetry::Context;
}

/// Publishes events to the message bus.
pub trait Publisher {
	/// Publish an event to its subject.
	///
	/// # Errors
	///
	/// Returns [`Error::SendFailed`] if the message could not be published.
	fn publish<E>(&self, message: E) -> impl Future<Output = Result<(), Error>> + Send
	where
		E: Event + Clone + 'static;
}

/// Subscribes to events from the message bus.
///
/// Multiple service instances with the same consumer name will automatically
/// load-balance message processing.
pub trait Subscriber {
	/// Subscribe to events of type `E`.
	///
	/// Returns a stream that yields acknowledged-capable event wrappers.
	///
	/// # Errors
	///
	/// Returns [`Error::SubscriptionFailed`] if the subscription could not be
	/// created.
	fn subscribe<E: Event + 'static>(
		&self,
	) -> impl Future<
		Output = Result<
			impl Stream<Item = Result<impl ReceivedEvent<E> + Send + Sync + 'static, Error>>
			+ Send
			+ 'static,
			Error,
		>,
	> + Send;
}