feat: async microsvc transports

README.md

@@ -131,8 +131,8 @@ use sourced_rust::{microsvc, HashMapRepository, Queueable};
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let service = Arc::new(sourced_rust::register_handlers!(
         microsvc::Service::with_repo(HashMapRepository::new().queued()),
-        handlers::todo_create,
-        handlers::todo_complete,
+        command handlers::todo_create,
+        command handlers::todo_complete,
     ));
 
     // Direct dispatch
@@ -1031,8 +1031,8 @@ Register them with the `register_handlers!` macro:
 ```rust
 let service = sourced_rust::register_handlers!(
     microsvc::Service::with_repo(HashMapRepository::new().queued()),
-    handlers::counter_create,
-    handlers::counter_increment,
+    command handlers::counter_create,
+    command handlers::counter_increment,
 );
 ```
 

src/microsvc/context.rs

@@ -1,16 +1,18 @@
-//! Context passed to command handlers.
+//! Context passed to handlers.
 //!
-//! Carries the parsed input, session variables, and a reference to the service
-//! dependencies. Handlers access everything they need through the context.
+//! Carries the message, parsed JSON payload when available, session variables,
+//! and a reference to the service dependencies. Handlers access everything they
+//! need through the context.
 
 use serde::de::DeserializeOwned;
 use serde_json::Value;
 
 use super::dependencies::{HasReadModelStore, HasRepo};
 use super::error::HandlerError;
+use super::service::Message;
 use super::session::Session;
 
-/// The context passed to every command handler.
+/// The context passed to every handler.
 ///
 /// Generic over `D` (the service dependency type) so handlers can access the
 /// repository, read-model store, or custom dependencies the service is
@@ -28,9 +30,9 @@ use super::session::Session;
 /// }
 /// ```
 pub struct Context<'a, D> {
-    /// The command name being handled.
-    command_name: String,
-    /// Raw JSON input from the request.
+    /// Message being handled.
+    message: Message,
+    /// Raw JSON payload input, when the payload is JSON.
     input: Value,
     /// Session variables (user ID, role, etc.).
     session: Session,
@@ -41,13 +43,13 @@ pub struct Context<'a, D> {
 impl<'a, D> Context<'a, D> {
     /// Create a new context.
     pub(crate) fn new(
-        command_name: String,
+        message: Message,
         input: Value,
         session: Session,
         dependencies: &'a D,
     ) -> Self {
         Self {
-            command_name,
+            message,
             input,
             session,
             dependencies,
@@ -56,8 +58,7 @@ impl<'a, D> Context<'a, D> {
 
     /// Deserialize the input payload into a typed struct.
     pub fn input<T: DeserializeOwned>(&self) -> Result<T, HandlerError> {
-        serde_json::from_value(self.input.clone())
-            .map_err(|e| HandlerError::DecodeFailed(e.to_string()))
+        self.message.payload_json()
     }
 
     /// Get the raw JSON input.
@@ -67,7 +68,17 @@ impl<'a, D> Context<'a, D> {
 
     /// Get the command name.
     pub fn command_name(&self) -> &str {
-        &self.command_name
+        self.message.name()
+    }
+
+    /// Get the message name.
+    pub fn message_name(&self) -> &str {
+        self.message.name()
+    }
+
+    /// Get the full message, including id, raw payload bytes, and metadata.
+    pub fn message(&self) -> &Message {
+        &self.message
     }
 
     /// Get the session.

src/microsvc/grpc.rs

@@ -16,7 +16,8 @@
 //!
 //! let service = Arc::new(
 //!     microsvc::Service::with_repo(HashMapRepository::new())
-//!         .command("counter.create", |ctx| { /* ... */ })
+//!         .command("counter.create")
+//!         .handle(|ctx| { /* ... */ })
 //! );
 //!
 //! // Get the server to compose with other tonic routes
@@ -180,7 +181,7 @@ impl<D: Send + Sync + 'static> CommandService for GrpcHandler<D> {
     ) -> Result<Response<HealthResponse>, Status> {
         let commands: Vec<String> = self
             .service
-            .commands()
+            .command_names()
             .into_iter()
             .map(|s| s.to_string())
             .collect();

src/microsvc/http.rs

@@ -15,7 +15,8 @@
 //!
 //! let service = Arc::new(
 //!     microsvc::Service::with_repo(HashMapRepository::new())
-//!         .command("counter.create", |ctx| { /* ... */ })
+//!         .command("counter.create")
+//!         .handle(|ctx| { /* ... */ })
 //! );
 //!
 //! // Get the router to compose with other axum routes
@@ -60,7 +61,7 @@ pub async fn serve<D: Send + Sync + 'static>(
 async fn health_handler<D: Send + Sync + 'static>(
     State(service): State<Arc<Service<D>>>,
 ) -> impl IntoResponse {
-    let commands: Vec<&str> = service.commands();
+    let commands: Vec<&str> = service.command_names();
     Json(json!({ "ok": true, "commands": commands }))
 }
 

src/microsvc/mod.rs

@@ -1,6 +1,6 @@
 //! microsvc — Convention-based microservice command handler framework.
 //!
-//! Build microservices by registering command handlers on a `Service`.
+//! Build microservices by registering command and event handlers on a `Service`.
 //! Each handler receives a `Context<D>` with access to the input payload,
 //! session variables, and the service dependencies.
 //!
@@ -13,7 +13,8 @@
 //!
 //! let service = Arc::new(
 //!     microsvc::Service::with_repo(HashMapRepository::new())
-//!         .command("order.create", |ctx| {
+//!         .command("order.create")
+//!         .handle(|ctx| {
 //!             let input = ctx.input::<CreateOrderInput>()?;
 //!             Ok(json!({ "id": input.id }))
 //!         })
@@ -64,7 +65,10 @@ pub use dependencies::{
     RepoReadModelDependencies,
 };
 pub use error::HandlerError;
-pub use service::{CommandRequest, CommandResponse, Service};
+pub use service::{
+    CommandRequest, CommandResponse, DeliveryKind, HandlerBuilder, HandlerNames, HandlerSpec,
+    Message, MessageKind, Service, SubscriptionPlan,
+};
 pub use session::Session;
 
 // Bus transports (requires "bus" feature)
@@ -85,29 +89,85 @@ pub use grpc::{grpc_server, serve_grpc, GrpcServeError};
 
 /// Register handler modules with a service using the convention pattern.
 ///
-/// Each handler module must export:
+/// Each handler entry must be prefixed with `command`, `event`, or `events`.
+///
+/// Command handler modules must export:
 /// - `COMMAND: &str` — the command name
 /// - `guard(ctx) -> bool` — input validation
 /// - `handle(ctx) -> Result<Value, HandlerError>` — the handler
 ///
+/// Event handler modules must export:
+/// - `EVENT: &str` or `EVENTS: &[&str]` — event names
+/// - `guard(ctx) -> bool` — input validation
+/// - `handle(ctx) -> Result<Value, HandlerError>` — the handler
+///
 /// # Example
 /// ```ignore
 /// let service = sourced_rust::register_handlers!(
 ///     microsvc::Service::with_repo(HashMapRepository::new()),
-///     handlers::counter_create,
-///     handlers::counter_increment,
+///     command handlers::counter_create,
+///     command handlers::counter_increment,
+///     event handlers::counter_rebuilt,
+///     events handlers::counter_projection,
 /// );
 /// ```
 #[macro_export]
 macro_rules! register_handlers {
-    ($service:expr, $( $($seg:ident)::+ ),+ $(,)?) => {
+    ($service:expr $(,)?) => {
         $service
-        $(
-            .command_guarded(
-                $($seg)::+::COMMAND,
+    };
+    ($service:expr, $($rest:tt)+) => {
+        $crate::__register_handlers!($service, $($rest)+)
+    };
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __register_handlers {
+    ($service:expr, command $($seg:ident)::+ $(, $($rest:tt)*)?) => {
+        $crate::__register_handlers_continue!(
+            $service.command($($seg)::+::COMMAND).guarded(
+                $($seg)::+::guard,
+                $($seg)::+::handle,
+            )
+            $(, $($rest)*)?
+        )
+    };
+    ($service:expr, event $($seg:ident)::+ $(, $($rest:tt)*)?) => {
+        $crate::__register_handlers_continue!(
+            $service.event($($seg)::+::EVENT).guarded(
                 $($seg)::+::guard,
                 $($seg)::+::handle,
             )
-        )+
+            $(, $($rest)*)?
+        )
+    };
+    ($service:expr, events $($seg:ident)::+ $(, $($rest:tt)*)?) => {
+        $crate::__register_handlers_continue!(
+            $service.events($($seg)::+::EVENTS).guarded(
+                $($seg)::+::guard,
+                $($seg)::+::handle,
+            )
+            $(, $($rest)*)?
+        )
+    };
+    ($service:expr, $($seg:ident)::+ $(, $($rest:tt)*)?) => {
+        compile_error!(
+            "register_handlers! entries must be prefixed with `command`, `event`, or `events`"
+        )
+    };
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __register_handlers_continue {
+    ($service:expr) => {
+        $service
+    };
+    ($service:expr,) => {
+        $service
+    };
+    ($service:expr, $($rest:tt)+) => {
+        $crate::__register_handlers!($service, $($rest)+)
     };
 }