adapter-event-publisher: broadcast channel event bus

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files"]
+members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher"]
 exclude = ["k-tv-backend", "k-tv-frontend"]
 resolver = "2"
 

crates/adapters/event-publisher/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "adapter-event-publisher"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+domain = { workspace = true }
+async-trait = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }

crates/adapters/event-publisher/src/lib.rs

@@ -0,0 +1,45 @@
+use async_trait::async_trait;
+use domain::errors::{DomainError, DomainResult};
+use domain::events::DomainEvent;
+use domain::ports::events::{EventConsumer, EventPublisher};
+use tokio::sync::broadcast;
+
+pub struct ChannelEventBus {
+    tx: broadcast::Sender<DomainEvent>,
+}
+
+impl ChannelEventBus {
+    pub fn new(capacity: usize) -> Self {
+        let (tx, _) = broadcast::channel(capacity);
+        Self { tx }
+    }
+
+    pub fn subscriber(&self) -> broadcast::Receiver<DomainEvent> {
+        self.tx.subscribe()
+    }
+
+    pub fn sender(&self) -> broadcast::Sender<DomainEvent> {
+        self.tx.clone()
+    }
+}
+
+#[async_trait]
+impl EventPublisher for ChannelEventBus {
+    async fn publish(&self, event: DomainEvent) -> DomainResult<()> {
+        let _ = self.tx.send(event); // Ok to drop if no receivers
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl EventConsumer for ChannelEventBus {
+    async fn recv(&self) -> DomainResult<DomainEvent> {
+        // Note: This creates a new subscriber each call — for real use,
+        // the presentation layer should hold a receiver from subscriber()
+        // This impl exists to satisfy the port trait
+        let mut rx = self.tx.subscribe();
+        rx.recv()
+            .await
+            .map_err(|e| DomainError::InfrastructureError(e.to_string()))
+    }
+}