484 lines
16 KiB
Markdown
484 lines
16 KiB
Markdown
# event-transport Rename + Consumer Abstraction Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Rename `event-publisher` → `event-transport` and add the symmetric consumer abstraction (`MessageSource` trait + `EventConsumerAdapter<S>`) so both publish and subscribe are transport-agnostic.
|
|
|
|
**Architecture after this plan:**
|
|
```
|
|
event-transport/ ← Transport + EventPublisherAdapter<T> (existing)
|
|
← MessageSource + EventConsumerAdapter<S> (new)
|
|
← RawMessage { subject, payload, ack, nack } (new)
|
|
|
|
nats/ ← NatsTransport (existing, implements Transport)
|
|
← NatsMessageSource (new, implements MessageSource)
|
|
← NatsEventConsumer removed
|
|
|
|
worker/ ← EventConsumerAdapter::new(NatsMessageSource::new(client))
|
|
```
|
|
|
|
**Dependency chain:**
|
|
```
|
|
event-transport → domain, event-payload, serde_json, async-trait
|
|
nats → domain, event-payload, event-transport, async-nats
|
|
worker → domain, nats, event-transport, postgres
|
|
```
|
|
|
|
---
|
|
|
|
## File Map
|
|
|
|
```
|
|
Rename: crates/adapters/event-publisher/ → crates/adapters/event-transport/
|
|
Modify: Cargo.toml (root) ← update member path + workspace dep name
|
|
Modify: crates/adapters/event-transport/Cargo.toml ← name = "event-transport"
|
|
Modify: crates/adapters/nats/Cargo.toml ← event-publisher → event-transport
|
|
Modify: crates/adapters/nats/src/lib.rs ← use event_transport; add NatsMessageSource; remove NatsEventConsumer
|
|
Modify: crates/bootstrap/Cargo.toml ← event-publisher → event-transport
|
|
Modify: crates/bootstrap/src/factory.rs ← use event_transport; update EventConsumerAdapter wiring
|
|
Modify: crates/worker/Cargo.toml ← add event-transport dep
|
|
Modify: crates/worker/src/main.rs ← EventConsumerAdapter<NatsMessageSource>
|
|
Modify: crates/adapters/event-transport/src/lib.rs ← add RawMessage + MessageSource + EventConsumerAdapter
|
|
```
|
|
|
|
---
|
|
|
|
### Task 1: Rename crate + update all references
|
|
|
|
**Files:** root `Cargo.toml`, `event-publisher/Cargo.toml` (renamed), `nats/Cargo.toml`, `bootstrap/Cargo.toml`, `nats/src/lib.rs`, `bootstrap/src/factory.rs`
|
|
|
|
- [ ] **Rename the directory:**
|
|
|
|
```bash
|
|
git mv crates/adapters/event-publisher crates/adapters/event-transport
|
|
```
|
|
|
|
- [ ] **Update `crates/adapters/event-transport/Cargo.toml`** — change the package name:
|
|
|
|
```toml
|
|
[package]
|
|
name = "event-transport"
|
|
version = "0.1.0"
|
|
edition = "2021"
|
|
|
|
[dependencies]
|
|
domain = { workspace = true }
|
|
event-payload = { workspace = true }
|
|
serde_json = { workspace = true }
|
|
async-trait = { workspace = true }
|
|
tracing = { workspace = true }
|
|
|
|
[dev-dependencies]
|
|
tokio = { workspace = true, features = ["full"] }
|
|
```
|
|
|
|
- [ ] **Update root `Cargo.toml`:**
|
|
|
|
In `[workspace] members`, change:
|
|
```toml
|
|
"crates/adapters/event-publisher",
|
|
```
|
|
to:
|
|
```toml
|
|
"crates/adapters/event-transport",
|
|
```
|
|
|
|
In `[workspace.dependencies]`, change:
|
|
```toml
|
|
event-publisher = { path = "crates/adapters/event-publisher" }
|
|
```
|
|
to:
|
|
```toml
|
|
event-transport = { path = "crates/adapters/event-transport" }
|
|
```
|
|
|
|
- [ ] **Update `crates/adapters/nats/Cargo.toml`:**
|
|
|
|
Change `event-publisher = { workspace = true }` to `event-transport = { workspace = true }`.
|
|
|
|
- [ ] **Update `crates/adapters/nats/src/lib.rs`:**
|
|
|
|
Change `use event_publisher::Transport;` to `use event_transport::Transport;`.
|
|
|
|
- [ ] **Update `crates/bootstrap/Cargo.toml`:**
|
|
|
|
Change `event-publisher = { workspace = true }` to `event-transport = { workspace = true }`.
|
|
|
|
- [ ] **Update `crates/bootstrap/src/factory.rs`:**
|
|
|
|
Change `use event_publisher::EventPublisherAdapter;` to `use event_transport::EventPublisherAdapter;`.
|
|
|
|
- [ ] **Run:** `cargo check --workspace` — Expected: no errors.
|
|
|
|
- [ ] **Run:** `cargo test -p event-transport` — Expected: 2 tests pass (same tests as before, just crate renamed).
|
|
|
|
- [ ] **Commit:**
|
|
|
|
```bash
|
|
git add Cargo.toml \
|
|
crates/adapters/event-transport/ \
|
|
crates/adapters/nats/Cargo.toml \
|
|
crates/adapters/nats/src/lib.rs \
|
|
crates/bootstrap/Cargo.toml \
|
|
crates/bootstrap/src/factory.rs
|
|
git commit -m "refactor: rename event-publisher → event-transport"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: Add MessageSource + EventConsumerAdapter to event-transport
|
|
|
|
**Files:**
|
|
- Modify: `crates/adapters/event-transport/src/lib.rs`
|
|
|
|
- [ ] **Write failing tests** — append to the test module in `src/lib.rs`:
|
|
|
|
```rust
|
|
#[tokio::test]
|
|
async fn consumer_adapter_deserializes_and_yields_event() {
|
|
use domain::value_objects::ThoughtId;
|
|
use futures::StreamExt;
|
|
|
|
// Produce a serialized EventPayload for ThoughtCreated
|
|
let event = DomainEvent::ThoughtCreated {
|
|
thought_id: ThoughtId::new(),
|
|
user_id: UserId::new(),
|
|
in_reply_to_id: None,
|
|
};
|
|
let payload = EventPayload::from(&event);
|
|
let bytes = serde_json::to_vec(&payload).unwrap();
|
|
|
|
// A MessageSource that yields one message then ends
|
|
struct OneMessageSource { bytes: Vec<u8> }
|
|
#[async_trait]
|
|
impl MessageSource for OneMessageSource {
|
|
fn messages(&self) -> futures::stream::BoxStream<'_, Result<RawMessage, DomainError>> {
|
|
let msg = RawMessage {
|
|
subject: "thoughts.created".to_string(),
|
|
payload: self.bytes.clone(),
|
|
ack: Box::new(|| {}),
|
|
nack: Box::new(|| {}),
|
|
};
|
|
Box::pin(futures::stream::once(async { Ok(msg) }))
|
|
}
|
|
}
|
|
|
|
let adapter = EventConsumerAdapter::new(OneMessageSource { bytes });
|
|
let mut stream = adapter.consume();
|
|
let envelope = stream.next().await.unwrap().unwrap();
|
|
assert!(matches!(envelope.event, DomainEvent::ThoughtCreated { .. }));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn consumer_adapter_skips_invalid_payloads() {
|
|
use futures::StreamExt;
|
|
|
|
struct BadMessageSource;
|
|
#[async_trait]
|
|
impl MessageSource for BadMessageSource {
|
|
fn messages(&self) -> futures::stream::BoxStream<'_, Result<RawMessage, DomainError>> {
|
|
let msg = RawMessage {
|
|
subject: "bad".to_string(),
|
|
payload: b"not valid json".to_vec(),
|
|
ack: Box::new(|| {}),
|
|
nack: Box::new(|| {}),
|
|
};
|
|
Box::pin(futures::stream::once(async { Ok(msg) }))
|
|
}
|
|
}
|
|
|
|
let adapter = EventConsumerAdapter::new(BadMessageSource);
|
|
let mut stream = adapter.consume();
|
|
// Invalid JSON should be skipped — stream ends with no items
|
|
assert!(stream.next().await.is_none());
|
|
}
|
|
```
|
|
|
|
- [ ] **Run:** `cargo test -p event-transport` — Expected: FAIL (MessageSource, RawMessage, EventConsumerAdapter not defined).
|
|
|
|
- [ ] **Add to `crates/adapters/event-transport/src/lib.rs`** — append after the existing `EventPublisherAdapter` impl and before `#[cfg(test)]`:
|
|
|
|
```rust
|
|
use domain::{events::EventEnvelope, ports::EventConsumer};
|
|
use futures::stream::BoxStream;
|
|
|
|
/// A raw inbound message from a transport backend.
|
|
/// `ack` and `nack` are transport-level acknowledgements (e.g. Kafka offset commit).
|
|
/// For at-most-once transports (basic NATS), both are no-ops.
|
|
pub struct RawMessage {
|
|
pub subject: String,
|
|
pub payload: Vec<u8>,
|
|
pub ack: Box<dyn Fn() + Send + Sync>,
|
|
pub nack: Box<dyn Fn() + Send + Sync>,
|
|
}
|
|
|
|
/// Abstraction over any subscribe/consume backend.
|
|
/// Implement this for NATS, Kafka, Redis Streams, etc.
|
|
pub trait MessageSource: Send + Sync {
|
|
fn messages(&self) -> BoxStream<'_, Result<RawMessage, DomainError>>;
|
|
}
|
|
|
|
/// Deserializes raw transport messages into domain `EventEnvelope`s.
|
|
///
|
|
/// Converts: `RawMessage.payload` → `EventPayload` → `DomainEvent` → `EventEnvelope`
|
|
///
|
|
/// Invalid or unknown messages are skipped with a warning — the stream continues.
|
|
pub struct EventConsumerAdapter<S: MessageSource> {
|
|
source: S,
|
|
}
|
|
|
|
impl<S: MessageSource> EventConsumerAdapter<S> {
|
|
pub fn new(source: S) -> Self { Self { source } }
|
|
}
|
|
|
|
impl<S: MessageSource> EventConsumer for EventConsumerAdapter<S> {
|
|
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
|
|
use futures::StreamExt;
|
|
let stream = self.source.messages();
|
|
Box::pin(stream.filter_map(|result| async move {
|
|
match result {
|
|
Err(e) => {
|
|
tracing::warn!("transport error: {e}");
|
|
None
|
|
}
|
|
Ok(msg) => {
|
|
let payload = match serde_json::from_slice::<EventPayload>(&msg.payload) {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
tracing::warn!("failed to deserialize event payload: {e}");
|
|
return None;
|
|
}
|
|
};
|
|
let event = match DomainEvent::try_from(payload) {
|
|
Ok(e) => e,
|
|
Err(e) => {
|
|
tracing::warn!("unknown event type: {e}");
|
|
return None;
|
|
}
|
|
};
|
|
Some(Ok(EventEnvelope {
|
|
event,
|
|
ack: msg.ack,
|
|
nack: msg.nack,
|
|
}))
|
|
}
|
|
}
|
|
}))
|
|
}
|
|
}
|
|
```
|
|
|
|
Note: the existing imports at the top of `lib.rs` already have `use domain::...` — add `EventEnvelope` and `EventConsumer` to those imports. Also add `futures::stream::BoxStream` if not already present.
|
|
|
|
Also add `futures = { workspace = true }` to `event-transport/Cargo.toml` dependencies (needed for `BoxStream` and `StreamExt`).
|
|
|
|
- [ ] **Run:** `cargo test -p event-transport` — Expected: 4 tests pass (2 existing + 2 new).
|
|
|
|
- [ ] **Commit:**
|
|
|
|
```bash
|
|
git add crates/adapters/event-transport/
|
|
git commit -m "feat(event-transport): MessageSource trait + EventConsumerAdapter for transport-agnostic consuming"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: nats — add NatsMessageSource, remove NatsEventConsumer
|
|
|
|
**Files:**
|
|
- Modify: `crates/adapters/nats/src/lib.rs`
|
|
|
|
- [ ] **Rewrite `crates/adapters/nats/src/lib.rs`** — remove `NatsEventConsumer`, add `NatsMessageSource`:
|
|
|
|
```rust
|
|
use async_trait::async_trait;
|
|
use domain::errors::DomainError;
|
|
use event_payload::EventPayload;
|
|
use event_transport::{MessageSource, RawMessage, Transport};
|
|
use futures::stream::BoxStream;
|
|
|
|
// ── NatsTransport — raw NATS publish backend ────────────────────────────────
|
|
|
|
pub struct NatsTransport {
|
|
client: async_nats::Client,
|
|
}
|
|
|
|
impl NatsTransport {
|
|
pub fn new(client: async_nats::Client) -> Self { Self { client } }
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Transport for NatsTransport {
|
|
async fn publish_bytes(&self, subject: &str, bytes: &[u8]) -> Result<(), DomainError> {
|
|
self.client
|
|
.publish(subject, bytes.to_vec().into())
|
|
.await
|
|
.map_err(|e| DomainError::Internal(e.to_string()))
|
|
}
|
|
}
|
|
|
|
// ── NatsMessageSource — raw NATS subscribe backend ──────────────────────────
|
|
|
|
pub struct NatsMessageSource {
|
|
client: async_nats::Client,
|
|
}
|
|
|
|
impl NatsMessageSource {
|
|
pub fn new(client: async_nats::Client) -> Self { Self { client } }
|
|
}
|
|
|
|
impl MessageSource for NatsMessageSource {
|
|
fn messages(&self) -> BoxStream<'_, Result<RawMessage, DomainError>> {
|
|
let client = self.client.clone();
|
|
Box::pin(async_stream::try_stream! {
|
|
let mut sub = client
|
|
.subscribe(">")
|
|
.await
|
|
.map_err(|e| DomainError::Internal(e.to_string()))?;
|
|
|
|
use futures::StreamExt;
|
|
while let Some(msg) = sub.next().await {
|
|
let subject = msg.subject.to_string();
|
|
let payload = msg.payload.to_vec();
|
|
// Basic NATS: at-most-once delivery — ack/nack are no-ops.
|
|
// Replace with JetStream for at-least-once delivery.
|
|
yield RawMessage {
|
|
subject,
|
|
payload,
|
|
ack: Box::new(|| {}),
|
|
nack: Box::new(|| {}),
|
|
};
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use domain::{events::DomainEvent, value_objects::{LikeId, ThoughtId, UserId}};
|
|
|
|
#[test]
|
|
fn payload_from_domain_event_has_correct_subject() {
|
|
let event = DomainEvent::ThoughtCreated {
|
|
thought_id: ThoughtId::new(),
|
|
user_id: UserId::new(),
|
|
in_reply_to_id: None,
|
|
};
|
|
let payload = EventPayload::from(&event);
|
|
assert_eq!(payload.subject(), "thoughts.created");
|
|
}
|
|
|
|
#[test]
|
|
fn domain_event_roundtrip_via_payload() {
|
|
let uid = UserId::new();
|
|
let tid = ThoughtId::new();
|
|
let event = DomainEvent::LikeAdded {
|
|
like_id: LikeId::new(),
|
|
user_id: uid.clone(),
|
|
thought_id: tid.clone(),
|
|
};
|
|
let payload = EventPayload::from(&event);
|
|
let back = DomainEvent::try_from(payload).unwrap();
|
|
if let DomainEvent::LikeAdded { user_id, thought_id, .. } = back {
|
|
assert_eq!(user_id, uid);
|
|
assert_eq!(thought_id, tid);
|
|
} else {
|
|
panic!("wrong variant");
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
- [ ] **Run:** `cargo test -p nats` — Expected: 2 tests pass.
|
|
|
|
- [ ] **Run:** `cargo check --workspace` — Expected: one error in `worker` (uses removed `NatsEventConsumer`). That's expected — fixed in Task 4.
|
|
|
|
- [ ] **Commit:**
|
|
|
|
```bash
|
|
git add crates/adapters/nats/src/lib.rs
|
|
git commit -m "refactor(nats): replace NatsEventConsumer with NatsMessageSource implementing MessageSource"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 4: Update worker + full verification
|
|
|
|
**Files:**
|
|
- Modify: `crates/worker/Cargo.toml`
|
|
- Modify: `crates/worker/src/main.rs`
|
|
|
|
- [ ] **Add `event-transport = { workspace = true }` to `crates/worker/Cargo.toml`.**
|
|
|
|
- [ ] **Update `crates/worker/src/main.rs`** — find and update the consumer creation.
|
|
|
|
Current code in `main.rs`:
|
|
```rust
|
|
let consumer = nats::NatsEventConsumer::new(nats_client);
|
|
```
|
|
|
|
Replace with:
|
|
```rust
|
|
use event_transport::EventConsumerAdapter;
|
|
use nats::NatsMessageSource;
|
|
let consumer = EventConsumerAdapter::new(NatsMessageSource::new(nats_client));
|
|
```
|
|
|
|
Also add the `use` statements at the top of `main.rs` alongside existing imports.
|
|
|
|
- [ ] **Run:** `cargo check --workspace` — Expected: no errors.
|
|
|
|
- [ ] **Run full test suite:**
|
|
|
|
```bash
|
|
DATABASE_URL=postgres://postgres:postgres@localhost:5434/postgres cargo test --workspace 2>&1 | tail -3
|
|
```
|
|
|
|
Expected: all tests pass (79 existing + 2 new event-transport consumer tests = 81+).
|
|
|
|
- [ ] **Commit:**
|
|
|
|
```bash
|
|
git add crates/worker/
|
|
git commit -m "feat(worker): use EventConsumerAdapter<NatsMessageSource> — transport-agnostic consuming"
|
|
```
|
|
|
|
---
|
|
|
|
## Self-Review
|
|
|
|
**Spec coverage:**
|
|
- ✅ `event-publisher` renamed to `event-transport` everywhere (Task 1)
|
|
- ✅ `RawMessage { subject, payload, ack, nack }` in `event-transport` (Task 2)
|
|
- ✅ `MessageSource` trait with `messages() -> BoxStream<RawMessage>` (Task 2)
|
|
- ✅ `EventConsumerAdapter<S: MessageSource>` implementing `EventConsumer` (Task 2)
|
|
- ✅ Invalid messages skipped with warning, stream continues (Task 2)
|
|
- ✅ 2 new tests: valid deserialization + invalid JSON skip (Task 2)
|
|
- ✅ `NatsEventConsumer` removed from nats (Task 3)
|
|
- ✅ `NatsMessageSource` implementing `MessageSource` added to nats (Task 3)
|
|
- ✅ Worker uses `EventConsumerAdapter::new(NatsMessageSource::new(client))` (Task 4)
|
|
|
|
**Adding Kafka later:**
|
|
```toml
|
|
# kafka/Cargo.toml: event-transport = { workspace = true }
|
|
```
|
|
```rust
|
|
// kafka/src/lib.rs
|
|
pub struct KafkaMessageSource { ... }
|
|
impl MessageSource for KafkaMessageSource { ... } // yields RawMessage + real ack/nack
|
|
|
|
pub struct KafkaTransport { ... }
|
|
impl Transport for KafkaTransport { ... }
|
|
```
|
|
```rust
|
|
// bootstrap/src/factory.rs — two lines change:
|
|
EventPublisherAdapter::new(KafkaTransport::new(...))
|
|
EventConsumerAdapter::new(KafkaMessageSource::new(...))
|
|
```
|
|
|
|
**Type consistency:**
|
|
- `EventConsumerAdapter<NatsMessageSource>` — `NatsMessageSource` implements `MessageSource`, adapter implements `EventConsumer` ✓
|
|
- `RawMessage.ack` / `.nack` transferred to `EventEnvelope.ack` / `.nack` in consumer adapter ✓
|
|
- `event_transport::` (underscore) is the Rust module name for `event-transport` (dash) crate ✓
|