refactor: extract the bus into a standalone crate::bus module

[patrickleet]

What

Extracts the message bus out of src/microsvc/transport/ into a standalone top-level src/bus/ module that imports nothing from microsvc. microsvc now depends on the bus, not the reverse. Behavior-preserving — pure restructuring.

Implements specs/bus-module-decomposition (hardened after an adversarial design review) / tasks/bus-decomposition-phase1.

Why

Three goals, all delivered:

• Standalone bus usage — consume the bus with no Service via the new dependency-free Handlers builder.
• Inline handlers — bus.subscribe(Handlers::new().on_event("seat.reserved", |msg| async { … }), opts).
• The bus is its own module, not embedded in microsvc.

How (phased, each commit compiles + green)

• MessageRouter seam — run_source + all five BusConsumer adapters consume a small MessageRouter { handles, subscription_plan, dispatch } trait instead of the concrete Service<D>. Service<D> implements it; classification of HandlerError happens on the microsvc side.
• Handlers — a second, dependency-free MessageRouter impl (the Rust analog of Node's bus.listen('x', fn)); idempotent-only by design.
• Message/MessageKind/SubscriptionPlan → bus core, plus a bus-core PayloadDecodeError so payload decoding no longer returns microsvc::HandlerError (mapped back via From on the microsvc side).
• Error coupling inverted — bus::error names no microsvc types; From<HandlerError>/classification live in microsvc, From<RepositoryError> with the outbox bridge.
• Outbox bridge → outbox_worker (OutboxDispatcher/OutboxSource, re-exported at the crate root).
• Knative split — the Service-coupled HTTP ingress (cloud_events_router) → microsvc::knative_ingress; the Message/SubscriptionPlan-only manifest helpers stay in the bus.
• Clean break — the transitional microsvc::transport alias is removed; all call sites (incl. tests + docs) use crate::bus.

Verification

• 229 lib + 493 default-feature integration tests green; cargo test --all-features --no-run clean (all broker/Knative test crates compile).
• All 9 single-feature combinations (--no-default-features [--features …]) compile.
• grep confirms src/bus/ imports nothing from crate::microsvc.
• cargo doc at parity with main (4 pre-existing warnings, zero new).
• A multi-agent review of the full branch diff (behavior-preservation, dependency purity, public-API completeness, feature-combo compilation) returned GO with only cosmetic doc-link fixes (applied).

Public API moves (intended)

• cloud_events_router → sourced_rust::microsvc::cloud_events_router (was under the transport module).
• OutboxDispatcher / OutboxSource → sourced_rust:: crate root.
• Message / MessageKind / SubscriptionPlan / PayloadDecodeError re-exported from both sourced_rust::bus and sourced_rust::microsvc.

Deferred (tracked in the spec)

The per-transport crate split (sourced-bus-*) is deferred — feature-gating already provides compile isolation, so its only added benefit is independent versioning. Revisit when a consumer needs it.

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Fluent Handlers API for registering async command/event handlers
  • MessageRouter abstraction for flexible routing and dispatch
  • Core Message model (kinds, payload/metadata, subscription plans)
  • Knative trigger manifest generation utilities

• Documentation

  • README and transport docs updated to the new bus facade and import paths

• Refactor

  • Consolidated transport surface under a unified bus module
  • Transport consumer APIs migrated to router-driven model
  • Transport error conversion/classification refined

• Tests

  • Integration tests updated to use reorganized bus APIs

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: fe2118fc-1c4b-4165-9184-40530891de29

📥 Commits

Reviewing files that changed from the base of the PR and between 10a75d8 and 31de327.

📒 Files selected for processing (5)

• src/bus/knative.rs
• src/bus/knative_bus.rs
• src/bus/mod.rs
• tests/distributed_read_model/main.rs
• tests/knative_cloudevents/main.rs

✅ Files skipped from review due to trivial changes (1)

• tests/knative_cloudevents/main.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• tests/distributed_read_model/main.rs

---

📝 Walkthrough

Walkthrough

This PR refactors the async transport layer to decouple bus consumers from a concrete Service<D> type, replacing it with a MessageRouter trait abstraction. The change introduces bus-core message types, an in-memory handler registry, and updates all transport implementations and error handling accordingly.

Changes

Bus Transport Refactor

Layer / File(s)	Summary
Message & Router Core Contracts src/bus/message.rs, src/bus/router.rs	Message, MessageKind, SubscriptionPlan, and PayloadDecodeError define the shared transport vocabulary. MessageRouter trait specifies handler detection, subscription planning, and async dispatch returning classified TransportError.
MessageRouter Implementations src/bus/handlers.rs, src/microsvc/message_router.rs	Handlers struct provides a dependency-free MessageRouter with fluent on_command/on_event registration APIs. Service<D> gains a MessageRouter impl that bridges legacy service dispatch logic.
Run Source Core Loop Refactor src/bus/runner.rs	run_source now accepts Arc<MessageRouter> instead of Arc<Service<D>>, routing through router.handles() and router.dispatch() with preserved failure-policy semantics.
Transport Consumer Implementations src/bus/{in_memory_bus,kafka_bus,nats_bus,postgres_bus,rabbit_bus,knative_bus}.rs	BusConsumer::listen/subscribe now accept MessageRouter instead of Service, deriving command/event subscription names from router.subscription_plan() instead of service methods.
Error Handling Reclassification src/bus/error.rs, src/microsvc/error.rs, src/outbox_worker/outbox_dispatch.rs	bus/error.rs removes handler classification; microsvc/error.rs adds PayloadDecodeError conversion, transport_error_kind() method, and From<HandlerError> conversion; outbox dispatcher gains From<RepositoryError> for transport retries.
Knative Transport Utilities src/bus/knative.rs, src/microsvc/knative_ingress.rs	Kubernetes manifest helpers (sanitize_k8s_name, knative_triggers) move to bus/knative.rs; knative_bus uses unique_k8s_name to avoid trigger-name collisions; ingress updates error classification to use transport_error_kind().
Module Organization & Public API Surface src/lib.rs, src/microsvc/mod.rs, src/bus/mod.rs, src/microsvc/service.rs, src/outbox_worker/mod.rs	Bus module declared at crate level; microsvc re-exports bus types from crate::bus; outbox worker exposes dispatch/source items; microsvc::transport module removed; local Message/MessageKind/SubscriptionPlan definitions removed from microsvc.
Context & Message Type Migrations src/microsvc/context.rs, src/bus/{kafka,nats,rabbitmq}.rs, src/bus/{publisher,run_options}.rs	Message imported from crate::bus; transport modules update imports to reference message types from parent/bus modules; Context::input maps payload decode errors via HandlerError::from.
Outbox Worker Bus Integration src/outbox_worker/mod.rs, src/outbox_worker/outbox_source.rs	Outbox worker declares and re-exports outbox_dispatch/outbox_source submodules; imports migrate to crate::bus for transport traits.
Documentation & Test Imports Migration README.md, docs/async-transports.md, tests/**/*.rs	Documentation examples and all test files updated to import from sourced_rust::bus instead of sourced_rust::microsvc::transport; distributed_read_model matrix helpers and transport-specific tests use new bus module paths.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• patrickleet/sourced_rust#15: Overlaps at error plumbing and HandlerError conversions used by transport error classification.
• patrickleet/sourced_rust#43: Related work on service/Message subscription planning and handler wiring that this change centralizes into the bus layer.
• patrickleet/sourced_rust#48: Overlapping test/integration changes touching distributed_read_model and bus facade imports.

"A rabbit hops, I code away,
Routers hum where Services lay,
Messages hop and handlers play,
Errors sorted, buses sway. 🐇"

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'refactor: extract the bus into a standalone crate::bus module' clearly and accurately summarizes the main change of extracting the bus from microsvc/transport into a top-level module.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/bus-module-decomposition

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (3)

src/bus/handlers.rs (1)

110-122: ⚡ Quick win

Make subscription_plan() deterministic and linear-time.

HashMap iteration order is nondeterministic, and the bucket.iter().any(...) check is redundant here because keys are already unique by (kind, name). Build buckets directly and sort before returning.

♻️ Proposed refactor

     fn subscription_plan(&self) -> SubscriptionPlan {
-        let mut plan = SubscriptionPlan::default();
-        for (kind, name) in self.handlers.keys() {
-            let bucket = match kind {
-                MessageKind::Command => &mut plan.commands,
-                MessageKind::Event => &mut plan.events,
-            };
-            if !bucket.iter().any(|existing| existing == name) {
-                bucket.push(name.clone());
-            }
-        }
-        plan
+        let mut plan = SubscriptionPlan::default();
+        for (kind, name) in self.handlers.keys() {
+            match kind {
+                MessageKind::Command => plan.commands.push(name.clone()),
+                MessageKind::Event => plan.events.push(name.clone()),
+            }
+        }
+        plan.commands.sort();
+        plan.events.sort();
+        plan
     }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/bus/handlers.rs` around lines 110 - 122, The subscription_plan()
currently iterates a HashMap and does an O(n^2) duplicate check which yields
nondeterministic order; change it to build the command and event buckets
directly by iterating self.handlers.keys() and pushing names into the
appropriate bucket (no .any() duplicate check since keys are unique by
(kind,name)), then sort the buckets (e.g., plan.commands.sort();
plan.events.sort()) before returning so the result is deterministic and overall
linearithmic instead of quadratic; update the function subscription_plan,
operate on SubscriptionPlan::default()'s commands and events, and match on
MessageKind to decide which bucket to push into.

tests/microsvc/transport_subscribe.rs (1)

8-10: ⚡ Quick win

Import Message and MessageKind from sourced_rust::bus in this bus-facing test.

This test still builds transport messages through sourced_rust::microsvc, so it doesn't fully cover the new top-level bus API. Pulling those types from sourced_rust::bus here would make the migration complete and keep this test aligned with the boundary the PR is establishing.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/microsvc/transport_subscribe.rs` around lines 8 - 10, Update the test
imports so Message and MessageKind are pulled from the top-level bus API instead
of microsvc: remove Message and MessageKind from the sourced_rust::microsvc
import and import them from sourced_rust::bus (alongside Bus, BusConsumer,
InMemoryBus, RunOptions) while leaving Service (and other existing imports)
intact; this ensures the test constructs transport messages using
sourced_rust::bus types (symbols: Message, MessageKind, Service, Bus,
BusConsumer).

src/bus/rabbit_bus.rs (1)

148-157: ⚡ Quick win

Short-circuit empty event plans before declaring RabbitMQ resources.

ensure_subscription and subscribe still perform broker setup on a no-op event plan. Move the empty-plan check ahead of declarations to avoid unnecessary AMQP writes and connection work.

♻️ Suggested change

 pub async fn ensure_subscription<R: MessageRouter>(
     &self,
     router: &R,
 ) -> Result<(), TransportError> {
+    let plan = router.subscription_plan();
+    if plan.events.is_empty() {
+        return Ok(());
+    }
     self.declare_events_exchange(&self.channel).await?;
     let queue = self.group_queue();
     self.declare_queue(&self.channel, &queue).await?;
-    let plan = router.subscription_plan();
     for name in &plan.events {
         self.channel
             .queue_bind(
                 &queue,
                 &self.events_exchange,
@@
     async fn subscribe<R: MessageRouter>(
         &self,
         router: Arc<R>,
         options: RunOptions,
     ) -> Result<(), TransportError> {
         self.ensure_subscription(router.as_ref()).await?;
-        if router.subscription_plan().events.is_empty() {
-            return Ok(());
-        }
         let channel = connect_channel(&self.uri).await?;
         let source = RabbitBusSource {
             channel,
             queues: vec![self.group_queue()],
             // Events are published with routing key == the bare event name.

Also applies to: 230-233

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/bus/rabbit_bus.rs` around lines 148 - 157, The functions
ensure_subscription and subscribe currently perform RabbitMQ setup even when the
router.subscription_plan() has no events; to fix, compute let plan =
router.subscription_plan() early in both ensure_subscription and subscribe,
return Ok(()) immediately if plan.events.is_empty() (short-circuit), and only
proceed to call declare_events_exchange, declare_queue, and iterate plan.events
when non-empty — update references in ensure_subscription and subscribe to check
plan.events before any channel or broker declaration to avoid unnecessary AMQP
work.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/bus/knative.rs`:
- Around line 40-44: The current knative_triggers function can produce duplicate
metadata.name after sanitize_k8s_name because different event strings can
normalize to the same DNS label; change the trigger name generation so you
append a short deterministic suffix derived from the original event (e.g., a 6–8
char hex of a stable hash like SHA-1 or CRC32 of event) to the base
"{subscriber_service}-{event}" before calling sanitize_k8s_name, ensuring
uniqueness while keeping names deterministic; update the trigger_name creation
in knative_triggers to compute that short hash suffix and include it in the
formatted name.

In `@src/bus/mod.rs`:
- Line 141: The module docs still describe run_source dispatching via
Service::dispatch_message but the public API now exports MessageRouter; update
the header docs to describe router-based dispatch instead: replace or augment
the overview that references run_source and Service::dispatch_message with
language that explains messages are dispatched through the MessageRouter
abstraction (mention how run_source interacts with or delegates to MessageRouter
if applicable), and update any examples/links in the doc comments to reference
MessageRouter rather than Service::dispatch_message so rustdoc reflects the new
public surface.

---

Nitpick comments:
In `@src/bus/handlers.rs`:
- Around line 110-122: The subscription_plan() currently iterates a HashMap and
does an O(n^2) duplicate check which yields nondeterministic order; change it to
build the command and event buckets directly by iterating self.handlers.keys()
and pushing names into the appropriate bucket (no .any() duplicate check since
keys are unique by (kind,name)), then sort the buckets (e.g.,
plan.commands.sort(); plan.events.sort()) before returning so the result is
deterministic and overall linearithmic instead of quadratic; update the function
subscription_plan, operate on SubscriptionPlan::default()'s commands and events,
and match on MessageKind to decide which bucket to push into.

In `@src/bus/rabbit_bus.rs`:
- Around line 148-157: The functions ensure_subscription and subscribe currently
perform RabbitMQ setup even when the router.subscription_plan() has no events;
to fix, compute let plan = router.subscription_plan() early in both
ensure_subscription and subscribe, return Ok(()) immediately if
plan.events.is_empty() (short-circuit), and only proceed to call
declare_events_exchange, declare_queue, and iterate plan.events when non-empty —
update references in ensure_subscription and subscribe to check plan.events
before any channel or broker declaration to avoid unnecessary AMQP work.

In `@tests/microsvc/transport_subscribe.rs`:
- Around line 8-10: Update the test imports so Message and MessageKind are
pulled from the top-level bus API instead of microsvc: remove Message and
MessageKind from the sourced_rust::microsvc import and import them from
sourced_rust::bus (alongside Bus, BusConsumer, InMemoryBus, RunOptions) while
leaving Service (and other existing imports) intact; this ensures the test
constructs transport messages using sourced_rust::bus types (symbols: Message,
MessageKind, Service, Bus, BusConsumer).

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 1ffc2cb8-6399-4324-a06b-1e830250d21f

📥 Commits

Reviewing files that changed from the base of the PR and between a1021c9 and 10a75d8.

📒 Files selected for processing (46)

• README.md
• docs/async-transports.md
• src/bus/bus.rs
• src/bus/capabilities.rs
• src/bus/error.rs
• src/bus/failure_policy.rs
• src/bus/handlers.rs
• src/bus/in_memory_bus.rs
• src/bus/kafka.rs
• src/bus/kafka_bus.rs
• src/bus/knative.rs
• src/bus/knative_bus.rs
• src/bus/message.rs
• src/bus/mod.rs
• src/bus/nats.rs
• src/bus/nats_bus.rs
• src/bus/postgres_bus.rs
• src/bus/publisher.rs
• src/bus/rabbit_bus.rs
• src/bus/rabbitmq.rs
• src/bus/router.rs
• src/bus/run_options.rs
• src/bus/runner.rs
• src/bus/source.rs
• src/bus/stable_id.rs
• src/lib.rs
• src/microsvc/context.rs
• src/microsvc/error.rs
• src/microsvc/knative_ingress.rs
• src/microsvc/message_router.rs
• src/microsvc/mod.rs
• src/microsvc/service.rs
• src/outbox_worker/mod.rs
• src/outbox_worker/outbox_dispatch.rs
• src/outbox_worker/outbox_source.rs
• tests/distributed_read_model/main.rs
• tests/distributed_read_model_board/main.rs
• tests/kafka_transport/main.rs
• tests/knative_cloudevents/main.rs
• tests/microsvc/transport_listen.rs
• tests/microsvc/transport_subscribe.rs
• tests/nats_transport/main.rs
• tests/postgres_transport/main.rs
• tests/rabbitmq_transport/main.rs
• tests/sagas/microsvc_saga.rs
• tests/transport_conformance/mod.rs

💤 Files with no reviewable changes (1)

• src/bus/error.rs