feat(chat): add chatlayer crate (actor, consumer, events, projection, producer, service, metrics)
zhenyi committed
Added chat/chatlayer/actor/mod.rs chat/chatlayer/actor/mod.rs +5 −0 View chat/chatlayer/actor/mod.rs
@@ -0,0 +1,5 @@
1pub mod room;
2pub mod runtime;
3pub mod user;
4
5pub use runtime::{ActorRegistry, Mailbox};
Added chat/chatlayer/actor/room.rs chat/chatlayer/actor/room.rs +97 −0 View chat/chatlayer/actor/room.rs
@@ -0,0 +1,97 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4use storagelayer::DynChatStorage;
5use tokio_util::sync::CancellationToken;
6
7use crate::actor::runtime::{ActorRegistry, run_actor};
8use crate::dedup;
9use crate::error::ChatLayerResult;
10use crate::events::envelope::EventEnvelope;
11use crate::events::room::ChatRoomEvent;
12use crate::media::MediaService;
13use crate::producer::ChatProducer;
14use crate::projection;
15use crate::storage_bridge::MessageIdempotency;
16
17#[derive(Clone)]
18pub struct RoomActorCtx {
19 pub store: DynChatStorage,
20 pub idempotency: MessageIdempotency,
21 pub producer: Arc<ChatProducer>,
22 pub kv: Arc<RedisCache>,
23 pub media: Option<Arc<MediaService>>,
24 pub dedup_ttl_seconds: u64,
25}
26
27impl std::fmt::Debug for RoomActorCtx {
28 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29 f.debug_struct("RoomActorCtx")
30 .field("media", &self.media.as_ref().map(|_| "<MediaService>"))
31 .field("dedup_ttl_seconds", &self.dedup_ttl_seconds)
32 .finish()
33 }
34}
35
36impl RoomActorCtx {
37 /// Errors are logged and swallowed; Kafka at-least-once redelivery is
38 pub async fn handle(&self, envelope: EventEnvelope<ChatRoomEvent>) {
39 match self.handle_inner(&envelope).await {
40 Ok(()) => crate::metrics::room_event_applied("success"),
41 Err(e) => {
42 crate::metrics::room_event_applied("failure");
43 tracing::error!(
44 event_id = %envelope.event_id,
45 channel = %envelope.payload.channel_id(),
46 error = %e,
47 "room event apply failed"
48 );
49 }
50 }
51 }
52
53 async fn handle_inner(&self, envelope: &EventEnvelope<ChatRoomEvent>) -> ChatLayerResult<()> {
54 if !dedup::is_new_event(&self.kv, envelope.event_id, self.dedup_ttl_seconds).await? {
55 crate::metrics::dedup_hit();
56 return Ok(());
57 }
58 let user_events = projection::room::apply_room_event(
59 &self.store,
60 &self.idempotency,
61 self.media.as_ref(),
62 envelope,
63 )
64 .await?;
65 for evt in user_events {
66 let sub = EventEnvelope::new(envelope.actor_id, evt);
67 if let Err(e) = self.producer.publish_user(&sub).await {
68 tracing::warn!(
69 event_id = %envelope.event_id,
70 error = %e,
71 "failed to fan out user event"
72 );
73 }
74 }
75 Ok(())
76 }
77}
78
79pub fn spawn_room_registry(
80 ctx: Arc<RoomActorCtx>,
81 shutdown: CancellationToken,
82) -> Arc<ActorRegistry<EventEnvelope<ChatRoomEvent>>> {
83 ActorRegistry::new(
84 "room",
85 move |reg, _key, rx| {
86 let ctx = ctx.clone();
87 async move {
88 run_actor(rx, reg.shutdown_token(), move |msg| {
89 let ctx = ctx.clone();
90 Box::pin(async move { ctx.handle(msg).await })
91 })
92 .await;
93 }
94 },
95 shutdown,
96 )
97}
Added chat/chatlayer/actor/runtime.rs chat/chatlayer/actor/runtime.rs +158 −0 View chat/chatlayer/actor/runtime.rs
@@ -0,0 +1,158 @@
1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3
4use futures::future::BoxFuture;
5use tokio::sync::mpsc;
6use tokio_util::sync::CancellationToken;
7use uuid::Uuid;
8
9#[derive(Debug, Clone)]
10pub struct Mailbox<M> {
11 tx: mpsc::UnboundedSender<M>,
12}
13
14impl<M> Mailbox<M> {
15 pub fn send(&self, msg: M) -> bool {
16 self.tx.send(msg).is_ok()
17 }
18
19 pub fn is_closed(&self) -> bool {
20 self.tx.is_closed()
21 }
22}
23
24type SpawnFn<M> = dyn Fn(Arc<ActorRegistry<M>>, Uuid, mpsc::UnboundedReceiver<M>) -> BoxFuture<'static, ()>
25 + Send
26 + Sync;
27
28pub struct ActorRegistry<M> {
29 name: &'static str,
30 actors: Mutex<HashMap<Uuid, mpsc::UnboundedSender<M>>>,
31 spawn: Arc<SpawnFn<M>>,
32 shutdown: CancellationToken,
33}
34
35impl<M> std::fmt::Debug for ActorRegistry<M> {
36 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37 let live = self.actors.lock().unwrap().len();
38 f.debug_struct("ActorRegistry")
39 .field("name", &self.name)
40 .field("live_actors", &live)
41 .finish()
42 }
43}
44
45impl<M> ActorRegistry<M>
46where
47 M: Send + 'static,
48{
49 pub fn new<F, Fut>(name: &'static str, spawn: F, shutdown: CancellationToken) -> Arc<Self>
50 where
51 F: Fn(Arc<Self>, Uuid, mpsc::UnboundedReceiver<M>) -> Fut + Send + Sync + 'static,
52 Fut: Future<Output = ()> + Send + 'static,
53 {
54 let spawn: Arc<SpawnFn<M>> = Arc::new(move |reg, key, rx| Box::pin(spawn(reg, key, rx)));
55 Arc::new(Self {
56 name,
57 actors: Mutex::new(HashMap::new()),
58 spawn,
59 shutdown,
60 })
61 }
62
63 pub fn shutdown_token(&self) -> CancellationToken {
64 self.shutdown.clone()
65 }
66
67 pub fn live_count(&self) -> usize {
68 self.actors.lock().unwrap().len()
69 }
70
71 pub fn dispatch(self: &Arc<Self>, key: Uuid, mut msg: M) {
72 {
73 let guard = self.actors.lock().unwrap();
74 if let Some(tx) = guard.get(&key) {
75 match tx.send(msg) {
76 Ok(()) => return,
77 Err(e) => msg = e.0,
78 }
79 }
80 }
81 let (tx, rx) = mpsc::unbounded_channel();
82 let mailbox = Mailbox { tx: tx.clone() };
83 {
84 let mut guard = self.actors.lock().unwrap();
85 if let Some(existing) = guard.get(&key) {
86 match existing.send(msg) {
87 Ok(()) => return,
88 Err(e) => msg = e.0,
89 }
90 }
91 guard.insert(key, tx);
92 }
93 let registry = self.clone();
94 let spawn = self.spawn.clone();
95 tokio::spawn(async move {
96 let fut = (spawn)(registry.clone(), key, rx);
97 fut.await;
98 registry.actors.lock().unwrap().remove(&key);
99 });
100 let _ = mailbox.send(msg);
101 crate::metrics::actor_spawned(self.name);
102 }
103}
104
105pub async fn run_actor<M>(
106 mut rx: mpsc::UnboundedReceiver<M>,
107 shutdown: CancellationToken,
108 mut apply: impl FnMut(M) -> BoxFuture<'static, ()>,
109) {
110 loop {
111 tokio::select! {
112 _ = shutdown.cancelled() => break,
113 maybe = rx.recv() => {
114 match maybe {
115 Some(msg) => apply(msg).await,
116 None => break,
117 }
118 }
119 }
120 }
121}
122
123#[cfg(test)]
124mod tests {
125 use super::*;
126
127 #[tokio::test]
128 async fn dispatch_spawns_and_orders() {
129 let shutdown = CancellationToken::new();
130 let seen = Arc::new(Mutex::new(Vec::new()));
131 let reg = ActorRegistry::new(
132 "test",
133 {
134 let seen = seen.clone();
135 move |reg: Arc<ActorRegistry<u32>>, _key, rx| {
136 let seen = seen.clone();
137 async move {
138 run_actor(rx, reg.shutdown_token(), move |msg| {
139 let seen = seen.clone();
140 Box::pin(async move {
141 seen.lock().unwrap().push(msg);
142 })
143 })
144 .await;
145 }
146 }
147 },
148 shutdown,
149 );
150 let key = Uuid::new_v4();
151 reg.dispatch(key, 1);
152 reg.dispatch(key, 2);
153 reg.dispatch(key, 3);
154 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
155 let got = seen.lock().unwrap().clone();
156 assert_eq!(got, vec![1, 2, 3]);
157 }
158}
Added chat/chatlayer/actor/user.rs chat/chatlayer/actor/user.rs +87 −0 View chat/chatlayer/actor/user.rs
@@ -0,0 +1,87 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4use tokio_util::sync::CancellationToken;
5
6use crate::actor::runtime::{ActorRegistry, run_actor};
7use crate::config::AppChatSettings;
8use crate::dedup;
9use crate::error::ChatLayerResult;
10use crate::events::envelope::EventEnvelope;
11use crate::events::user::ChatUserEvent;
12use crate::producer::ChatProducer;
13use crate::projection;
14
15#[derive(Clone)]
16pub struct UserActorCtx {
17 pub kv: Arc<RedisCache>,
18 pub producer: Arc<ChatProducer>,
19 pub settings: AppChatSettings,
20}
21
22impl std::fmt::Debug for UserActorCtx {
23 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24 f.debug_struct("UserActorCtx")
25 .field("settings", &self.settings)
26 .finish()
27 }
28}
29
30impl UserActorCtx {
31 /// view events for the (future) gateway. Errors are logged and swallowed;
32 pub async fn handle(&self, envelope: EventEnvelope<ChatUserEvent>) {
33 match self.handle_inner(&envelope).await {
34 Ok(()) => crate::metrics::user_event_applied("success"),
35 Err(e) => {
36 crate::metrics::user_event_applied("failure");
37 tracing::error!(
38 event_id = %envelope.event_id,
39 user = %envelope.payload.user_id(),
40 error = %e,
41 "user event apply failed"
42 );
43 }
44 }
45 }
46
47 async fn handle_inner(&self, envelope: &EventEnvelope<ChatUserEvent>) -> ChatLayerResult<()> {
48 if !dedup::is_new_event(&self.kv, envelope.event_id, self.settings.dedup_ttl_seconds)
49 .await?
50 {
51 crate::metrics::dedup_hit();
52 return Ok(());
53 }
54 let views = projection::user::apply_user_event(&self.kv, &self.settings, envelope).await?;
55 for view in views {
56 let sub = EventEnvelope::new(envelope.actor_id, view);
57 if let Err(e) = self.producer.publish_user_view(&sub).await {
58 tracing::warn!(
59 event_id = %envelope.event_id,
60 error = %e,
61 "failed to publish user view event"
62 );
63 }
64 }
65 Ok(())
66 }
67}
68
69pub fn spawn_user_registry(
70 ctx: Arc<UserActorCtx>,
71 shutdown: CancellationToken,
72) -> Arc<ActorRegistry<EventEnvelope<ChatUserEvent>>> {
73 ActorRegistry::new(
74 "user",
75 move |reg, _key, rx| {
76 let ctx = ctx.clone();
77 async move {
78 run_actor(rx, reg.shutdown_token(), move |msg| {
79 let ctx = ctx.clone();
80 Box::pin(async move { ctx.handle(msg).await })
81 })
82 .await;
83 }
84 },
85 shutdown,
86 )
87}
Added chat/chatlayer/consumer/mod.rs chat/chatlayer/consumer/mod.rs +62 −0 View chat/chatlayer/consumer/mod.rs
@@ -0,0 +1,62 @@
1pub mod room_worker;
2pub mod user_worker;
3
4use queue::{AutoOffsetReset, SubscribeHandle, SubscriptionMode};
5
6use crate::config::AppChatSettings;
7
8pub fn group_mode(group_id: &str) -> SubscriptionMode {
9 SubscriptionMode::Group {
10 group_id: group_id.to_string(),
11 auto_offset_reset: AutoOffsetReset::Earliest,
12 }
13}
14
15#[derive(Debug, Default)]
16pub struct ConsumerRuntime {
17 handles: Vec<SubscribeHandle>,
18}
19
20impl ConsumerRuntime {
21 pub fn push(&mut self, handle: SubscribeHandle) {
22 self.handles.push(handle);
23 }
24
25 pub fn len(&self) -> usize {
26 self.handles.len()
27 }
28
29 pub fn is_empty(&self) -> bool {
30 self.handles.is_empty()
31 }
32
33 pub async fn shutdown(self) {
34 for handle in self.handles {
35 if let Err(e) = handle.shutdown().await {
36 tracing::warn!(error = %e, "consumer shutdown failed");
37 }
38 }
39 }
40}
41
42#[derive(Debug, Clone)]
43pub struct WorkerSettings {
44 pub topic: String,
45 pub group_id: String,
46}
47
48impl WorkerSettings {
49 pub fn room(cfg: &AppChatSettings) -> Self {
50 Self {
51 topic: crate::topics::room_topic(&cfg.topic_prefix),
52 group_id: cfg.room_group_id.clone(),
53 }
54 }
55
56 pub fn user(cfg: &AppChatSettings) -> Self {
57 Self {
58 topic: crate::topics::user_topic(&cfg.topic_prefix),
59 group_id: cfg.user_group_id.clone(),
60 }
61 }
62}
Added chat/chatlayer/consumer/room_worker.rs chat/chatlayer/consumer/room_worker.rs +78 −0 View chat/chatlayer/consumer/room_worker.rs
@@ -0,0 +1,78 @@
1use std::sync::Arc;
2
3use queue::{Queue, QueueError, SubscribeHandle};
4use tokio_util::sync::CancellationToken;
5
6use crate::actor::room::RoomActorCtx;
7use crate::actor::runtime::ActorRegistry;
8use crate::consumer::WorkerSettings;
9use crate::events::envelope::EventEnvelope;
10use crate::events::room::ChatRoomEvent;
11
12pub struct RoomWorker {
13 queue: Arc<Queue>,
14 settings: WorkerSettings,
15 registry: Arc<ActorRegistry<EventEnvelope<ChatRoomEvent>>>,
16}
17
18impl std::fmt::Debug for RoomWorker {
19 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20 f.debug_struct("RoomWorker")
21 .field("settings", &self.settings)
22 .finish_non_exhaustive()
23 }
24}
25
26impl RoomWorker {
27 pub fn new(
28 queue: Arc<Queue>,
29 settings: WorkerSettings,
30 registry: Arc<ActorRegistry<EventEnvelope<ChatRoomEvent>>>,
31 ) -> Self {
32 Self {
33 queue,
34 settings,
35 registry,
36 }
37 }
38
39 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
40 let mode = crate::consumer::group_mode(&self.settings.group_id);
41 let topics = [self.settings.topic.as_str()];
42 let registry = self.registry.clone();
43 self.queue.subscribe_raw(mode, &topics, move |msg| {
44 let registry = registry.clone();
45 Box::pin(async move {
46 match serde_json::from_slice::<EventEnvelope<ChatRoomEvent>>(&msg.payload) {
47 Ok(envelope) => {
48 registry.dispatch(envelope.payload.channel_id(), envelope);
49 Ok(())
50 }
51 Err(e) => {
52 tracing::warn!(
53 topic = %msg.topic,
54 offset = msg.offset,
55 error = %e,
56 "malformed room event dropped"
57 );
58 Ok(())
59 }
60 }
61 })
62 })
63 }
64
65 pub fn registry(&self) -> Arc<ActorRegistry<EventEnvelope<ChatRoomEvent>>> {
66 self.registry.clone()
67 }
68}
69
70pub fn build_room_worker(
71 queue: Arc<Queue>,
72 settings: WorkerSettings,
73 ctx: Arc<RoomActorCtx>,
74 shutdown: CancellationToken,
75) -> RoomWorker {
76 let registry = crate::actor::room::spawn_room_registry(ctx, shutdown);
77 RoomWorker::new(queue, settings, registry)
78}
Added chat/chatlayer/consumer/user_worker.rs chat/chatlayer/consumer/user_worker.rs +78 −0 View chat/chatlayer/consumer/user_worker.rs
@@ -0,0 +1,78 @@
1use std::sync::Arc;
2
3use queue::{Queue, QueueError, SubscribeHandle};
4use tokio_util::sync::CancellationToken;
5
6use crate::actor::runtime::ActorRegistry;
7use crate::actor::user::UserActorCtx;
8use crate::consumer::WorkerSettings;
9use crate::events::envelope::EventEnvelope;
10use crate::events::user::ChatUserEvent;
11
12pub struct UserWorker {
13 queue: Arc<Queue>,
14 settings: WorkerSettings,
15 registry: Arc<ActorRegistry<EventEnvelope<ChatUserEvent>>>,
16}
17
18impl std::fmt::Debug for UserWorker {
19 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20 f.debug_struct("UserWorker")
21 .field("settings", &self.settings)
22 .finish_non_exhaustive()
23 }
24}
25
26impl UserWorker {
27 pub fn new(
28 queue: Arc<Queue>,
29 settings: WorkerSettings,
30 registry: Arc<ActorRegistry<EventEnvelope<ChatUserEvent>>>,
31 ) -> Self {
32 Self {
33 queue,
34 settings,
35 registry,
36 }
37 }
38
39 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
40 let mode = crate::consumer::group_mode(&self.settings.group_id);
41 let topics = [self.settings.topic.as_str()];
42 let registry = self.registry.clone();
43 self.queue.subscribe_raw(mode, &topics, move |msg| {
44 let registry = registry.clone();
45 Box::pin(async move {
46 match serde_json::from_slice::<EventEnvelope<ChatUserEvent>>(&msg.payload) {
47 Ok(envelope) => {
48 registry.dispatch(envelope.payload.user_id(), envelope);
49 Ok(())
50 }
51 Err(e) => {
52 tracing::warn!(
53 topic = %msg.topic,
54 offset = msg.offset,
55 error = %e,
56 "malformed user event dropped"
57 );
58 Ok(())
59 }
60 }
61 })
62 })
63 }
64
65 pub fn registry(&self) -> Arc<ActorRegistry<EventEnvelope<ChatUserEvent>>> {
66 self.registry.clone()
67 }
68}
69
70pub fn build_user_worker(
71 queue: Arc<Queue>,
72 settings: WorkerSettings,
73 ctx: Arc<UserActorCtx>,
74 shutdown: CancellationToken,
75) -> UserWorker {
76 let registry = crate::actor::user::spawn_user_registry(ctx, shutdown);
77 UserWorker::new(queue, settings, registry)
78}
Added chat/chatlayer/events/envelope.rs chat/chatlayer/events/envelope.rs +42 −0 View chat/chatlayer/events/envelope.rs
@@ -0,0 +1,42 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct EventEnvelope<T> {
7 pub event_id: Uuid,
8 pub occurred_at: OffsetDateTime,
9 pub actor_id: Uuid,
10 pub payload: T,
11}
12
13impl<T> EventEnvelope<T> {
14 pub fn new(actor_id: Uuid, payload: T) -> Self {
15 Self {
16 event_id: Uuid::new_v4(),
17 occurred_at: OffsetDateTime::now_utc(),
18 actor_id,
19 payload,
20 }
21 }
22
23 pub fn with_timestamp(actor_id: Uuid, occurred_at: OffsetDateTime, payload: T) -> Self {
24 Self {
25 event_id: Uuid::new_v4(),
26 occurred_at,
27 actor_id,
28 payload,
29 }
30 }
31}
32
33impl<T: Clone> EventEnvelope<&T> {
34 pub fn into_owned(self) -> EventEnvelope<T> {
35 EventEnvelope {
36 event_id: self.event_id,
37 occurred_at: self.occurred_at,
38 actor_id: self.actor_id,
39 payload: self.payload.clone(),
40 }
41 }
42}
Added chat/chatlayer/events/mod.rs chat/chatlayer/events/mod.rs +72 −0 View chat/chatlayer/events/mod.rs
@@ -0,0 +1,72 @@
1pub mod envelope;
2pub mod room;
3pub mod user;
4pub mod user_view;
5
6pub use envelope::EventEnvelope;
7pub use room::{ChatRoomEvent, MentionSpec, MessageCommand};
8pub use user::ChatUserEvent;
9pub use user_view::ChatUserViewEvent;
10
11#[cfg(test)]
12mod tests {
13 use super::*;
14 use uuid::Uuid;
15
16 #[test]
17 fn room_event_serde_roundtrip() {
18 let cmd = MessageCommand {
19 channel_id: Uuid::new_v4(),
20 sender_id: Uuid::new_v4(),
21 body: "hi".into(),
22 reply_to: None,
23 thread_root: None,
24 is_thread_start: false,
25 attachment_ids: vec![],
26 mentions: vec![MentionSpec::everyone()],
27 client_msg_id: "c-1".into(),
28 };
29 let evt = ChatRoomEvent::MessageCreated(cmd);
30 let json = serde_json::to_string(&evt).unwrap();
31 let back: ChatRoomEvent = serde_json::from_str(&json).unwrap();
32 assert_eq!(back, evt);
33 assert!(json.contains("\"type\":\"message_created\""));
34 }
35
36 #[test]
37 fn room_event_channel_id() {
38 let channel = Uuid::new_v4();
39 let evt = ChatRoomEvent::MessageDeleted {
40 channel_id: channel,
41 message_id: Uuid::new_v4(),
42 by: Uuid::new_v4(),
43 };
44 assert_eq!(evt.channel_id(), channel);
45 }
46
47 #[test]
48 fn user_event_serde_roundtrip() {
49 let evt = ChatUserEvent::ReadReceiptAck {
50 user_id: Uuid::new_v4(),
51 channel_id: Uuid::new_v4(),
52 last_read_message_id: Uuid::new_v4(),
53 ts: time::OffsetDateTime::now_utc(),
54 };
55 let json = serde_json::to_string(&evt).unwrap();
56 let back: ChatUserEvent = serde_json::from_str(&json).unwrap();
57 assert_eq!(back, evt);
58 }
59
60 #[test]
61 fn envelope_key_derivation() {
62 let channel = Uuid::new_v4();
63 let evt = ChatRoomEvent::PinToggle {
64 channel_id: channel,
65 message_id: Uuid::new_v4(),
66 by: Uuid::new_v4(),
67 pinned: true,
68 };
69 assert_eq!(evt.channel_id(), channel);
70 assert_eq!(crate::topics::room_key(channel), channel.to_string());
71 }
72}
Added chat/chatlayer/events/room.rs chat/chatlayer/events/room.rs +100 −0 View chat/chatlayer/events/room.rs
@@ -0,0 +1,100 @@
1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
5pub struct MentionSpec {
6 pub user_id: Option<Uuid>,
7 pub role_id: Option<Uuid>,
8 pub channel_id: Option<Uuid>,
9 pub is_everyone: bool,
10 pub is_here: bool,
11}
12
13impl MentionSpec {
14 pub fn user(user_id: Uuid) -> Self {
15 Self {
16 user_id: Some(user_id),
17 role_id: None,
18 channel_id: None,
19 is_everyone: false,
20 is_here: false,
21 }
22 }
23
24 pub fn everyone() -> Self {
25 Self {
26 user_id: None,
27 role_id: None,
28 channel_id: None,
29 is_everyone: true,
30 is_here: false,
31 }
32 }
33}
34
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36pub struct MessageCommand {
37 pub channel_id: Uuid,
38 pub sender_id: Uuid,
39 pub body: String,
40 pub reply_to: Option<Uuid>,
41 pub thread_root: Option<Uuid>,
42 pub is_thread_start: bool,
43 pub attachment_ids: Vec<Uuid>,
44 pub mentions: Vec<MentionSpec>,
45 pub client_msg_id: String,
46}
47
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49#[serde(tag = "type", rename_all = "snake_case")]
50pub enum ChatRoomEvent {
51 MessageCreated(MessageCommand),
52 MessageEdited {
53 channel_id: Uuid,
54 message_id: Uuid,
55 editor_id: Uuid,
56 new_body: String,
57 },
58 MessageDeleted {
59 channel_id: Uuid,
60 message_id: Uuid,
61 by: Uuid,
62 },
63 ReactionAdded {
64 channel_id: Uuid,
65 message_id: Uuid,
66 user_id: Uuid,
67 emoji: String,
68 },
69 ReactionRemoved {
70 channel_id: Uuid,
71 message_id: Uuid,
72 user_id: Uuid,
73 emoji: String,
74 },
75 MentionAdded {
76 channel_id: Uuid,
77 message_id: Uuid,
78 mentions: Vec<MentionSpec>,
79 },
80 PinToggle {
81 channel_id: Uuid,
82 message_id: Uuid,
83 by: Uuid,
84 pinned: bool,
85 },
86}
87
88impl ChatRoomEvent {
89 pub fn channel_id(&self) -> Uuid {
90 match self {
91 Self::MessageCreated(cmd) => cmd.channel_id,
92 Self::MessageEdited { channel_id, .. }
93 | Self::MessageDeleted { channel_id, .. }
94 | Self::ReactionAdded { channel_id, .. }
95 | Self::ReactionRemoved { channel_id, .. }
96 | Self::MentionAdded { channel_id, .. }
97 | Self::PinToggle { channel_id, .. } => *channel_id,
98 }
99 }
100}
Added chat/chatlayer/events/user.rs chat/chatlayer/events/user.rs +43 −0 View chat/chatlayer/events/user.rs
@@ -0,0 +1,43 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6#[serde(tag = "type", rename_all = "snake_case")]
7pub enum ChatUserEvent {
8 MessageDelivered {
9 user_id: Uuid,
10 channel_id: Uuid,
11 message_id: Uuid,
12 ts: OffsetDateTime,
13 },
14 MentionReceived {
15 user_id: Uuid,
16 channel_id: Uuid,
17 message_id: Uuid,
18 by: Uuid,
19 ts: OffsetDateTime,
20 },
21 ReadReceiptAck {
22 user_id: Uuid,
23 channel_id: Uuid,
24 last_read_message_id: Uuid,
25 ts: OffsetDateTime,
26 },
27 PresenceHeartbeat {
28 user_id: Uuid,
29 device_id: String,
30 ts: OffsetDateTime,
31 },
32}
33
34impl ChatUserEvent {
35 pub fn user_id(&self) -> Uuid {
36 match self {
37 Self::MessageDelivered { user_id, .. }
38 | Self::MentionReceived { user_id, .. }
39 | Self::ReadReceiptAck { user_id, .. }
40 | Self::PresenceHeartbeat { user_id, .. } => *user_id,
41 }
42 }
43}
Added chat/chatlayer/events/user_view.rs chat/chatlayer/events/user_view.rs +52 −0 View chat/chatlayer/events/user_view.rs
@@ -0,0 +1,52 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6#[serde(tag = "type", rename_all = "snake_case")]
7pub enum ChatUserViewEvent {
8 NewMessage {
9 user_id: Uuid,
10 channel_id: Uuid,
11 message_id: Uuid,
12 sender_id: Uuid,
13 body: String,
14 ts: OffsetDateTime,
15 },
16 MessageEditedView {
17 user_id: Uuid,
18 channel_id: Uuid,
19 message_id: Uuid,
20 new_body: String,
21 ts: OffsetDateTime,
22 },
23 MessageDeletedView {
24 user_id: Uuid,
25 channel_id: Uuid,
26 message_id: Uuid,
27 ts: OffsetDateTime,
28 },
29 UnreadChanged {
30 user_id: Uuid,
31 channel_id: Uuid,
32 unread: u64,
33 ts: OffsetDateTime,
34 },
35 PresenceChanged {
36 user_id: Uuid,
37 online: bool,
38 ts: OffsetDateTime,
39 },
40}
41
42impl ChatUserViewEvent {
43 pub fn user_id(&self) -> Uuid {
44 match self {
45 Self::NewMessage { user_id, .. }
46 | Self::MessageEditedView { user_id, .. }
47 | Self::MessageDeletedView { user_id, .. }
48 | Self::UnreadChanged { user_id, .. }
49 | Self::PresenceChanged { user_id, .. } => *user_id,
50 }
51 }
52}
Added chat/chatlayer/projection/mod.rs chat/chatlayer/projection/mod.rs +5 −0 View chat/chatlayer/projection/mod.rs
@@ -0,0 +1,5 @@
1pub mod room;
2pub mod user;
3
4pub use room::apply_room_event;
5pub use user::apply_user_event;
Added chat/chatlayer/projection/room.rs chat/chatlayer/projection/room.rs +319 −0 View chat/chatlayer/projection/room.rs
@@ -0,0 +1,319 @@
1use std::sync::Arc;
2
3use storagelayer::DynChatStorage;
4use storagelayer::models::mention::NewMention;
5use storagelayer::models::message::NewMessage;
6use storagelayer::models::pinned::NewPinnedMessage;
7use storagelayer::models::reaction::NewReaction;
8use storagelayer::repository::ChatStorageCtx;
9use uuid::Uuid;
10
11use crate::error::ChatLayerResult;
12use crate::events::envelope::EventEnvelope;
13use crate::events::room::{ChatRoomEvent, MentionSpec, MessageCommand};
14use crate::events::user::ChatUserEvent;
15use crate::media::MediaService;
16use crate::storage_bridge::MessageIdempotencyStore;
17
18pub async fn apply_room_event(
19 store: &DynChatStorage,
20 idempotency: &dyn MessageIdempotencyStore,
21 media: Option<&Arc<MediaService>>,
22 envelope: &EventEnvelope<ChatRoomEvent>,
23) -> ChatLayerResult<Vec<ChatUserEvent>> {
24 let ctx = ChatStorageCtx::new(envelope.actor_id);
25 match &envelope.payload {
26 ChatRoomEvent::MessageCreated(cmd) => {
27 apply_message_created(store, idempotency, media, &ctx, envelope.actor_id, cmd).await
28 }
29 ChatRoomEvent::MessageEdited {
30 message_id,
31 new_body,
32 ..
33 } => {
34 store
35 .messages()
36 .edit(&ctx, *message_id, new_body.clone())
37 .await?;
38 Ok(Vec::new())
39 }
40 ChatRoomEvent::MessageDeleted { message_id, .. } => {
41 store.messages().soft_delete(&ctx, *message_id).await?;
42 Ok(Vec::new())
43 }
44 ChatRoomEvent::ReactionAdded {
45 message_id,
46 user_id,
47 emoji,
48 ..
49 } => {
50 store
51 .reactions()
52 .add_reaction(
53 &ctx,
54 NewReaction {
55 message_id: *message_id,
56 user_id: *user_id,
57 emoji: emoji.clone(),
58 },
59 )
60 .await?;
61 Ok(Vec::new())
62 }
63 ChatRoomEvent::ReactionRemoved {
64 message_id, emoji, ..
65 } => {
66 store
67 .reactions()
68 .remove_reaction(&ctx, *message_id, emoji)
69 .await?;
70 Ok(Vec::new())
71 }
72 ChatRoomEvent::MentionAdded {
73 message_id,
74 mentions,
75 ..
76 } => {
77 apply_mentions(store, &ctx, *message_id, mentions).await?;
78 Ok(Vec::new())
79 }
80 ChatRoomEvent::PinToggle {
81 channel_id,
82 message_id,
83 pinned,
84 ..
85 } => {
86 if *pinned {
87 store
88 .pinned()
89 .pin(
90 &ctx,
91 NewPinnedMessage {
92 chat_id: *channel_id,
93 message_id: *message_id,
94 pinned_by: envelope.actor_id,
95 },
96 )
97 .await?;
98 } else {
99 store.pinned().unpin(&ctx, *channel_id, *message_id).await?;
100 }
101 Ok(Vec::new())
102 }
103 }
104}
105
106async fn apply_message_created(
107 store: &DynChatStorage,
108 idempotency: &dyn MessageIdempotencyStore,
109 media: Option<&Arc<MediaService>>,
110 ctx: &ChatStorageCtx,
111 actor_id: Uuid,
112 cmd: &MessageCommand,
113) -> ChatLayerResult<Vec<ChatUserEvent>> {
114 if idempotency.is_resolved(&cmd.client_msg_id).await? {
115 crate::metrics::idempotency_hit();
116 return Ok(Vec::new());
117 }
118
119 let message = store
120 .messages()
121 .create(
122 ctx,
123 NewMessage {
124 chat_id: cmd.channel_id,
125 sender_id: cmd.sender_id,
126 body: cmd.body.clone(),
127 reply_to_message_id: cmd.reply_to,
128 thread_id: cmd.thread_root,
129 is_thread_start: cmd.is_thread_start,
130 flags: None,
131 attachment_ids: Vec::new(),
132 },
133 )
134 .await?;
135
136 idempotency.resolve(&cmd.client_msg_id, message.id).await?;
137 crate::metrics::message_persisted("success");
138
139 if let Some(media) = media {
140 for att_id in &cmd.attachment_ids {
141 if let Err(e) = media.attach_draft(actor_id, *att_id, message.id).await {
142 tracing::warn!(
143 message_id = %message.id,
144 attachment_id = %att_id,
145 error = %e,
146 "failed to attach draft to message"
147 );
148 }
149 }
150 }
151
152 let mut out = Vec::new();
153 if cmd.sender_id != ctx.user_id.unwrap_or(Uuid::nil()) {
154 out.push(ChatUserEvent::MessageDelivered {
155 user_id: cmd.sender_id,
156 channel_id: cmd.channel_id,
157 message_id: message.id,
158 ts: message.created_at,
159 });
160 }
161 for mention in &cmd.mentions {
162 if let Some(target) = mention_user(mention) {
163 out.push(ChatUserEvent::MentionReceived {
164 user_id: target,
165 channel_id: cmd.channel_id,
166 message_id: message.id,
167 by: cmd.sender_id,
168 ts: message.created_at,
169 });
170 }
171 }
172 Ok(out)
173}
174
175fn mention_user(spec: &MentionSpec) -> Option<Uuid> {
176 if spec.is_everyone || spec.is_here {
177 return None;
178 }
179 spec.user_id
180}
181
182async fn apply_mentions(
183 store: &DynChatStorage,
184 ctx: &ChatStorageCtx,
185 message_id: Uuid,
186 mentions: &[MentionSpec],
187) -> ChatLayerResult<()> {
188 for spec in mentions {
189 store
190 .mentions()
191 .add_mention(
192 ctx,
193 NewMention {
194 message_id,
195 user_id: spec.user_id,
196 role_id: spec.role_id,
197 channel_id: spec.channel_id,
198 is_everyone: spec.is_everyone,
199 is_here: spec.is_here,
200 },
201 )
202 .await?;
203 }
204 Ok(())
205}
206
207#[cfg(test)]
208mod tests {
209 use super::*;
210 use std::collections::HashMap;
211 use std::future::Future;
212 use std::pin::Pin;
213 use std::sync::Mutex;
214 use storagelayer::ChatStorageFactory;
215 use storagelayer::models::message::MessageQuery;
216 use storagelayer::repository::ChatStorageCtx;
217
218 struct MemIdempotency(Mutex<HashMap<String, Uuid>>);
219
220 impl MessageIdempotencyStore for MemIdempotency {
221 fn is_resolved<'a>(
222 &'a self,
223 client_msg_id: &'a str,
224 ) -> Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>> {
225 let map = &self.0;
226 Box::pin(async move { Ok(map.lock().unwrap().contains_key(client_msg_id)) })
227 }
228
229 fn resolve<'a>(
230 &'a self,
231 client_msg_id: &'a str,
232 message_id: Uuid,
233 ) -> Pin<Box<dyn Future<Output = ChatLayerResult<()>> + Send + 'a>> {
234 let map = &self.0;
235 Box::pin(async move {
236 map.lock()
237 .unwrap()
238 .insert(client_msg_id.to_string(), message_id);
239 Ok(())
240 })
241 }
242 }
243
244 fn cmd(channel: Uuid, sender: Uuid, body: &str, client_msg_id: &str) -> MessageCommand {
245 MessageCommand {
246 channel_id: channel,
247 sender_id: sender,
248 body: body.to_string(),
249 reply_to: None,
250 thread_root: None,
251 is_thread_start: false,
252 attachment_ids: Vec::new(),
253 mentions: Vec::new(),
254 client_msg_id: client_msg_id.to_string(),
255 }
256 }
257
258 #[tokio::test]
259 async fn message_created_persists_and_is_idempotent() {
260 let store = ChatStorageFactory::in_memory();
261 let idem = MemIdempotency(Mutex::new(HashMap::new()));
262 let sender = Uuid::new_v4();
263 let channel = Uuid::new_v4();
264
265 let envelope = EventEnvelope::new(
266 sender,
267 ChatRoomEvent::MessageCreated(cmd(channel, sender, "hi", "c-1")),
268 );
269 let _fanout = apply_room_event(&store, &idem, None, &envelope)
270 .await
271 .unwrap();
272
273 let ctx = ChatStorageCtx::new(sender);
274 let page = store
275 .messages()
276 .list_by_conversation(&ctx, channel, MessageQuery::default())
277 .await
278 .unwrap();
279 assert_eq!(page.items.len(), 1);
280 assert_eq!(page.items[0].body, "hi");
281
282 let again = EventEnvelope::new(
283 sender,
284 ChatRoomEvent::MessageCreated(cmd(channel, sender, "hi", "c-1")),
285 );
286 let _ = apply_room_event(&store, &idem, None, &again).await.unwrap();
287 let page = store
288 .messages()
289 .list_by_conversation(&ctx, channel, MessageQuery::default())
290 .await
291 .unwrap();
292 assert_eq!(
293 page.items.len(),
294 1,
295 "duplicate client_msg_id must be deduped"
296 );
297 }
298
299 #[tokio::test]
300 async fn mention_fans_out_to_target() {
301 let store = ChatStorageFactory::in_memory();
302 let idem = MemIdempotency(Mutex::new(HashMap::new()));
303 let sender = Uuid::new_v4();
304 let target = Uuid::new_v4();
305 let channel = Uuid::new_v4();
306
307 let mut mcmd = cmd(channel, sender, "hello @bob", "c-2");
308 mcmd.mentions = vec![MentionSpec::user(target)];
309 let envelope = EventEnvelope::new(sender, ChatRoomEvent::MessageCreated(mcmd));
310 let fanout = apply_room_event(&store, &idem, None, &envelope)
311 .await
312 .unwrap();
313
314 assert!(fanout.iter().any(|e| matches!(
315 e,
316 ChatUserEvent::MentionReceived { user_id, .. } if *user_id == target
317 )));
318 }
319}
Added chat/chatlayer/projection/user.rs chat/chatlayer/projection/user.rs +137 −0 View chat/chatlayer/projection/user.rs
@@ -0,0 +1,137 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4use time::OffsetDateTime;
5use uuid::Uuid;
6
7use crate::config::AppChatSettings;
8use crate::error::ChatLayerResult;
9use crate::events::envelope::EventEnvelope;
10use crate::events::user::ChatUserEvent;
11use crate::events::user_view::ChatUserViewEvent;
12
13pub const UNREAD_PREFIX: &str = "chat:unread:";
14pub const PRESENCE_PREFIX: &str = "chat:presence:";
15pub const MENTION_QUEUE_PREFIX: &str = "chat:mentions:";
16pub const READ_RECEIPT_PREFIX: &str = "chat:read:";
17
18pub fn unread_key(user_id: Uuid, channel_id: Uuid) -> String {
19 format!("{UNREAD_PREFIX}{user_id}:{channel_id}")
20}
21
22pub fn presence_key(user_id: Uuid) -> String {
23 format!("{PRESENCE_PREFIX}{user_id}")
24}
25
26pub fn mention_queue_key(user_id: Uuid) -> String {
27 format!("{MENTION_QUEUE_PREFIX}{user_id}")
28}
29
30fn read_receipt_key(user_id: Uuid, channel_id: Uuid) -> String {
31 format!("{READ_RECEIPT_PREFIX}{user_id}:{channel_id}")
32}
33
34pub async fn apply_user_event(
35 kv: &Arc<RedisCache>,
36 cfg: &AppChatSettings,
37 envelope: &EventEnvelope<ChatUserEvent>,
38) -> ChatLayerResult<Vec<ChatUserViewEvent>> {
39 match &envelope.payload {
40 ChatUserEvent::MessageDelivered {
41 user_id,
42 channel_id,
43 message_id,
44 ts,
45 } => {
46 let count = kv
47 .incr_with_ttl(&unread_key(*user_id, *channel_id), cfg.unread_ttl_seconds)
48 .await?;
49 Ok(vec![
50 ChatUserViewEvent::NewMessage {
51 user_id: *user_id,
52 channel_id: *channel_id,
53 message_id: *message_id,
54 sender_id: envelope.actor_id,
55 body: String::new(),
56 ts: *ts,
57 },
58 ChatUserViewEvent::UnreadChanged {
59 user_id: *user_id,
60 channel_id: *channel_id,
61 unread: count,
62 ts: *ts,
63 },
64 ])
65 }
66 ChatUserEvent::MentionReceived {
67 user_id,
68 channel_id,
69 message_id,
70 by,
71 ts,
72 } => {
73 let count = kv
74 .incr_with_ttl(&unread_key(*user_id, *channel_id), cfg.unread_ttl_seconds)
75 .await?;
76 kv.rpush(
77 &mention_queue_key(*user_id),
78 serde_json::to_vec(&envelope.payload)?,
79 )
80 .await?;
81 Ok(vec![
82 ChatUserViewEvent::NewMessage {
83 user_id: *user_id,
84 channel_id: *channel_id,
85 message_id: *message_id,
86 sender_id: *by,
87 body: String::new(),
88 ts: *ts,
89 },
90 ChatUserViewEvent::UnreadChanged {
91 user_id: *user_id,
92 channel_id: *channel_id,
93 unread: count,
94 ts: *ts,
95 },
96 ])
97 }
98 ChatUserEvent::ReadReceiptAck {
99 user_id,
100 channel_id,
101 last_read_message_id,
102 ts,
103 } => {
104 kv.set_bytes(
105 &read_receipt_key(*user_id, *channel_id),
106 last_read_message_id.as_bytes().to_vec(),
107 cfg.unread_ttl_seconds,
108 )
109 .await?;
110 kv.delete(&unread_key(*user_id, *channel_id)).await?;
111 Ok(vec![ChatUserViewEvent::UnreadChanged {
112 user_id: *user_id,
113 channel_id: *channel_id,
114 unread: 0,
115 ts: *ts,
116 }])
117 }
118 ChatUserEvent::PresenceHeartbeat {
119 user_id,
120 device_id,
121 ts,
122 } => {
123 let key = presence_key(*user_id);
124 let score = ts.unix_timestamp() as f64;
125 kv.zadd(&key, device_id, score, cfg.presence_ttl_seconds)
126 .await?;
127 let cutoff = (OffsetDateTime::now_utc().unix_timestamp() as f64)
128 - cfg.presence_ttl_seconds as f64;
129 let _ = kv.zremrange_by_score_lt(&key, cutoff).await;
130 Ok(vec![ChatUserViewEvent::PresenceChanged {
131 user_id: *user_id,
132 online: true,
133 ts: *ts,
134 }])
135 }
136 }
137}
Added chat/chatlayer/auth_bridge.rs chat/chatlayer/auth_bridge.rs +78 −0 View chat/chatlayer/auth_bridge.rs
@@ -0,0 +1,78 @@
1use std::sync::Arc;
2
3use jwt::JwtError;
4use jwt::signer::{decode_access_token, extract_kid};
5use storagelayer::repository::ChatStorageCtx;
6use uuid::Uuid;
7
8use crate::error::{ChatLayerError, ChatLayerResult};
9
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub struct AuthIdentity {
12 pub user_id: Uuid,
13 pub session_id: Option<String>,
14}
15
16pub struct AuthBridge {
17 jwt: Arc<jwt::JwtState>,
18}
19
20impl std::fmt::Debug for AuthBridge {
21 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22 f.debug_struct("AuthBridge").finish_non_exhaustive()
23 }
24}
25
26impl AuthBridge {
27 pub fn new(jwt: Arc<jwt::JwtState>) -> Self {
28 Self { jwt }
29 }
30
31 pub async fn verify(&self, token: &str) -> ChatLayerResult<AuthIdentity> {
32 let kid = extract_kid(token).ok_or(JwtError::InvalidHeader)?;
33 let cache = self.jwt.cache.read().await;
34 let cached = cache
35 .get(&kid)
36 .ok_or_else(|| JwtError::UnknownKid(kid.to_string()))?;
37 let claims = decode_access_token(
38 token,
39 &cached.decoding,
40 cached.algorithm,
41 &self.jwt.settings.issuer,
42 &self.jwt.settings.audience,
43 self.jwt.settings.clock_skew_seconds,
44 )?;
45 Ok(AuthIdentity {
46 user_id: claims.sub,
47 session_id: Some(claims.jti.to_string()),
48 })
49 }
50
51 pub async fn ctx_for(&self, token: &str) -> ChatLayerResult<ChatStorageCtx> {
52 let identity = self.verify(token).await?;
53 Ok(ChatStorageCtx::new(identity.user_id))
54 }
55
56 pub async fn require_identity(&self, token: &str) -> ChatLayerResult<AuthIdentity> {
57 self.verify(token).await.map_err(|e| match e {
58 ChatLayerError::Jwt(_) => {
59 ChatLayerError::PermissionDenied("valid bearer token required".to_string())
60 }
61 other => other,
62 })
63 }
64}
65
66#[cfg(test)]
67mod tests {
68 use super::*;
69
70 #[test]
71 fn identity_roundtrip_empty_session_is_none() {
72 let identity = AuthIdentity {
73 user_id: Uuid::new_v4(),
74 session_id: None,
75 };
76 assert_eq!(identity.session_id, None);
77 }
78}
Modified chat/chatlayer/Cargo.toml chat/chatlayer/Cargo.toml +22 −0 View chat/chatlayer/Cargo.toml
@@ -11,4 +11,26 @@
1111[lib]
1212path = "lib.rs"
1313name = "chatlayer"
14
1415[dependencies]
16storagelayer.workspace = true
17queue.workspace = true
18cache.workspace = true
19storage.workspace = true
20jwt.workspace = true
21
22tokio.workspace = true
23tokio-util = { version = "0.7", features = ["io", "rt"] }
24thiserror.workspace = true
25serde.workspace = true
26serde_json.workspace = true
27time.workspace = true
28uuid.workspace = true
29object_store.workspace = true
30tracing.workspace = true
31url.workspace = true
32metrics.workspace = true
33futures.workspace = true
34
35[dev-dependencies]
36storagelayer = { workspace = true, features = ["mock"] }
No newline at end of file
Added chat/chatlayer/config.rs chat/chatlayer/config.rs +42 −0 View chat/chatlayer/config.rs
@@ -0,0 +1,42 @@
1use serde::{Deserialize, Serialize};
2
3#[derive(Debug, Clone, Serialize, Deserialize)]
4#[serde(default, deny_unknown_fields)]
5pub struct AppChatSettings {
6 pub enabled: bool,
7
8 pub topic_prefix: String,
9
10 pub room_group_id: String,
11
12 pub user_group_id: String,
13
14 pub user_view_group_id: String,
15
16 pub dedup_ttl_seconds: u64,
17
18 pub idempotency_ttl_seconds: u64,
19
20 pub upload_ttl_seconds: u64,
21
22 pub presence_ttl_seconds: u64,
23
24 pub unread_ttl_seconds: u64,
25}
26
27impl Default for AppChatSettings {
28 fn default() -> Self {
29 Self {
30 enabled: true,
31 topic_prefix: "chat".to_string(),
32 room_group_id: "chat.room.v1".to_string(),
33 user_group_id: "chat.user.v1".to_string(),
34 user_view_group_id: "chat.user.view.v1".to_string(),
35 dedup_ttl_seconds: 24 * 3600,
36 idempotency_ttl_seconds: 24 * 3600,
37 upload_ttl_seconds: 3600,
38 presence_ttl_seconds: 120,
39 unread_ttl_seconds: 30 * 24 * 3600,
40 }
41 }
42}
Added chat/chatlayer/dedup.rs chat/chatlayer/dedup.rs +30 −0 View chat/chatlayer/dedup.rs
@@ -0,0 +1,30 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4
5use crate::error::ChatLayerResult;
6
7pub const EVT_DEDUP_PREFIX: &str = "chat:evt:";
8
9pub async fn is_new_event(
10 kv: &Arc<RedisCache>,
11 event_id: uuid::Uuid,
12 ttl_seconds: u64,
13) -> ChatLayerResult<bool> {
14 let key = format!("{EVT_DEDUP_PREFIX}{event_id}");
15 Ok(kv.set_nx(&key, b"1".to_vec(), ttl_seconds).await?)
16}
17
18#[cfg(test)]
19mod tests {
20 use super::*;
21
22 #[test]
23 fn dedup_key_format() {
24 let id = uuid::Uuid::nil();
25 assert_eq!(
26 format!("{EVT_DEDUP_PREFIX}{id}"),
27 "chat:evt:00000000-0000-0000-0000-000000000000"
28 );
29 }
30}
Added chat/chatlayer/error.rs chat/chatlayer/error.rs +53 −0 View chat/chatlayer/error.rs
@@ -0,0 +1,53 @@
1use thiserror::Error;
2
3#[derive(Debug, Error)]
4pub enum ChatLayerError {
5 #[error("{0}")]
6 Storage(#[from] storagelayer::ChatStorageError),
7
8 #[error("{0}")]
9 Queue(#[from] queue::QueueError),
10
11 #[error("{0}")]
12 Redis(#[from] cache::redis::RedisError),
13
14 #[error("{0}")]
15 Jwt(#[from] jwt::JwtError),
16
17 #[error("{0}")]
18 Blob(#[from] storage::StorageError),
19
20 #[error("{0}")]
21 Serde(#[from] serde_json::Error),
22
23 #[error("invalid input: {0}")]
24 InvalidInput(String),
25
26 #[error("permission denied: {0}")]
27 PermissionDenied(String),
28
29 #[error("conflict: {0}")]
30 Conflict(String),
31
32 #[error("not found: {0}")]
33 NotFound(String),
34
35 #[error("internal error: {0}")]
36 Internal(String),
37}
38
39pub type ChatLayerResult<T> = Result<T, ChatLayerError>;
40
41impl ChatLayerError {
42 pub fn invalid(msg: impl Into<String>) -> Self {
43 Self::InvalidInput(msg.into())
44 }
45
46 pub fn not_found(msg: impl Into<String>) -> Self {
47 Self::NotFound(msg.into())
48 }
49
50 pub fn conflict(msg: impl Into<String>) -> Self {
51 Self::Conflict(msg.into())
52 }
53}
Modified chat/chatlayer/lib.rs chat/chatlayer/lib.rs +28 −12 View chat/chatlayer/lib.rs
@@ -1,14 +1,30 @@
1pub fn add(left: u64, right: u64) -> u64 {
2 left + right
3}
1#![warn(missing_debug_implementations)]
2#![warn(unreachable_pub)]
43
5#[cfg(test)]
6mod tests {
7 use super::*;
4pub mod actor;
5pub mod auth_bridge;
6pub mod config;
7pub mod consumer;
8pub mod dedup;
9pub mod error;
10pub mod events;
11pub mod media;
12pub mod metrics;
13pub mod producer;
14pub mod projection;
15pub mod service;
16pub mod storage_bridge;
17pub mod topics;
818
9 #[test]
10 fn it_works() {
11 let result = add(2, 2);
12 assert_eq!(result, 4);
13 }
14}
19pub use auth_bridge::{AuthBridge, AuthIdentity};
20pub use config::AppChatSettings;
21pub use error::{ChatLayerError, ChatLayerResult};
22pub use events::{
23 envelope::EventEnvelope,
24 room::{ChatRoomEvent, MentionSpec, MessageCommand},
25 user::ChatUserEvent,
26 user_view::ChatUserViewEvent,
27};
28pub use media::{MediaService, PresignedUpload, UploadDraft};
29pub use producer::ChatProducer;
30pub use service::{AcceptedMsg, ChatService, WorkerHandle};
Added chat/chatlayer/media.rs chat/chatlayer/media.rs +323 −0 View chat/chatlayer/media.rs
@@ -0,0 +1,323 @@
1use std::sync::Arc;
2use std::time::Duration;
3
4use cache::redis::RedisCache;
5use object_store::path::Path as ObjectPath;
6use serde::{Deserialize, Serialize};
7use storagelayer::DynChatStorage;
8use storagelayer::blob::AttachmentPath;
9use storagelayer::models::attachment::{AttachmentOwner, NewAttachment};
10use storagelayer::repository::ChatStorageCtx;
11use url::Url;
12use uuid::Uuid;
13
14use crate::auth_bridge::AuthBridge;
15use crate::error::ChatLayerResult;
16
17const BLOB_PREFIX: &str = "chat";
18
19const DRAFT_PREFIX: &str = "chat:draft:";
20
21const UPLOAD_NONCE_PREFIX: &str = "chat:upload:";
22
23pub const DEFAULT_MAX_UPLOAD_BYTES: i64 = 100 * 1024 * 1024;
24
25#[derive(Debug, Clone)]
26pub struct PresignedUpload {
27 pub attachment_id: Uuid,
28 pub upload_url: Url,
29 pub key: String,
30 pub filename: String,
31 pub content_type: String,
32 pub size_hint: Option<i64>,
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize)]
36pub struct AttachmentMeta {
37 pub attachment_id: Uuid,
38 pub key: String,
39 pub filename: String,
40 pub content_type: String,
41 pub size: i64,
42}
43
44#[derive(Debug, Clone, Serialize, Deserialize)]
45pub struct UploadDraft {
46 pub sender_id: Uuid,
47 pub client_msg_id: Option<String>,
48 pub key: String,
49 pub filename: String,
50 pub content_type: String,
51 pub size: i64,
52}
53
54pub struct MediaService {
55 store: DynChatStorage,
56 storage: Arc<dyn storage::StorageRepository>,
57 kv: Arc<RedisCache>,
58 upload_ttl_seconds: u64,
59 max_upload_bytes: i64,
60}
61
62impl std::fmt::Debug for MediaService {
63 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64 f.debug_struct("MediaService")
65 .field("max_upload_bytes", &self.max_upload_bytes)
66 .finish_non_exhaustive()
67 }
68}
69
70impl MediaService {
71 pub fn new(
72 store: DynChatStorage,
73 storage: Arc<dyn storage::StorageRepository>,
74 kv: Arc<RedisCache>,
75 upload_ttl_seconds: u64,
76 ) -> Self {
77 Self {
78 store,
79 storage,
80 kv,
81 upload_ttl_seconds,
82 max_upload_bytes: DEFAULT_MAX_UPLOAD_BYTES,
83 }
84 }
85
86 pub fn with_max_upload_bytes(mut self, max: i64) -> Self {
87 self.max_upload_bytes = max;
88 self
89 }
90
91 fn draft_key(attachment_id: Uuid) -> String {
92 format!("{DRAFT_PREFIX}{attachment_id}")
93 }
94
95 fn object_path(key: &str) -> ObjectPath {
96 ObjectPath::from(format!("{BLOB_PREFIX}/{key}"))
97 }
98
99 #[allow(clippy::too_many_arguments)]
100 pub async fn presign_upload(
101 &self,
102 auth: &AuthBridge,
103 token: &str,
104 owner: AttachmentOwner,
105 owner_hint: Option<Uuid>,
106 filename: &str,
107 content_type: &str,
108 size_hint: Option<i64>,
109 ttl: Duration,
110 ) -> ChatLayerResult<PresignedUpload> {
111 let identity = auth.require_identity(token).await?;
112 if let Some(hint) = owner_hint {
113 let claimed = self
114 .kv
115 .set_nx(
116 &format!("{UPLOAD_NONCE_PREFIX}{hint}"),
117 b"1".to_vec(),
118 self.upload_ttl_seconds,
119 )
120 .await?;
121 if !claimed {
122 return Err(crate::error::ChatLayerError::conflict(format!(
123 "upload nonce {hint} already in use"
124 )));
125 }
126 }
127 if let Some(size) = size_hint
128 && size > self.max_upload_bytes
129 {
130 return Err(crate::error::ChatLayerError::invalid(format!(
131 "file too large: {size} bytes > {max}",
132 max = self.max_upload_bytes
133 )));
134 }
135
136 let attachment_id = Uuid::new_v4();
137 let path = AttachmentPath::new(owner, Uuid::nil(), filename);
138 let key = path.as_key();
139 let object_path = Self::object_path(&key);
140 let upload_url = self.storage.presigned_upload(&object_path, ttl).await?;
141
142 let draft = UploadDraft {
143 sender_id: identity.user_id,
144 client_msg_id: owner_hint.map(|h| h.to_string()),
145 key: key.clone(),
146 filename: filename.to_string(),
147 content_type: content_type.to_string(),
148 size: size_hint.unwrap_or(0),
149 };
150 self.kv
151 .set_bytes(
152 &Self::draft_key(attachment_id),
153 serde_json::to_vec(&draft)?,
154 self.upload_ttl_seconds,
155 )
156 .await?;
157
158 let ctx = ChatStorageCtx::new(identity.user_id);
159 self.store
160 .attachments()
161 .create_metadata(
162 &ctx,
163 NewAttachment {
164 owner,
165 owner_id: Uuid::nil(),
166 filename: filename.to_string(),
167 content_type: content_type.to_string(),
168 size: size_hint.unwrap_or(0),
169 url: object_path.to_string(),
170 uploaded_by: Some(identity.user_id),
171 },
172 )
173 .await?;
174
175 crate::metrics::media_op("presign", "success");
176 Ok(PresignedUpload {
177 attachment_id,
178 upload_url,
179 key,
180 filename: filename.to_string(),
181 content_type: content_type.to_string(),
182 size_hint,
183 })
184 }
185
186 pub async fn finalize_upload(
187 &self,
188 auth: &AuthBridge,
189 token: &str,
190 attachment_id: Uuid,
191 observed_size: i64,
192 _content_type: &str,
193 ) -> ChatLayerResult<AttachmentMeta> {
194 let identity = auth.require_identity(token).await?;
195 let mut draft = self.load_draft(attachment_id).await?;
196 if draft.sender_id != identity.user_id {
197 return Err(crate::error::ChatLayerError::PermissionDenied(
198 "only the uploader may finalize this attachment".to_string(),
199 ));
200 }
201
202 let meta = self.storage.head(&Self::object_path(&draft.key)).await?;
203 if meta.size > self.max_upload_bytes as u64 {
204 return Err(crate::error::ChatLayerError::invalid(format!(
205 "object too large: {} bytes > {max}",
206 meta.size,
207 max = self.max_upload_bytes
208 )));
209 }
210 if observed_size > self.max_upload_bytes || observed_size < 0 {
211 return Err(crate::error::ChatLayerError::invalid(format!(
212 "observed size {observed_size} out of range"
213 )));
214 }
215
216 draft.size = observed_size;
217 self.kv
218 .set_bytes(
219 &Self::draft_key(attachment_id),
220 serde_json::to_vec(&draft)?,
221 self.upload_ttl_seconds,
222 )
223 .await?;
224
225 crate::metrics::media_op("finalize", "success");
226 Ok(AttachmentMeta {
227 attachment_id,
228 key: draft.key.clone(),
229 filename: draft.filename.clone(),
230 content_type: draft.content_type.clone(),
231 size: observed_size,
232 })
233 }
234
235 pub async fn abort_upload(
236 &self,
237 auth: &AuthBridge,
238 token: &str,
239 attachment_id: Uuid,
240 ) -> ChatLayerResult<()> {
241 let identity = auth.require_identity(token).await?;
242 let draft = self.load_draft(attachment_id).await?;
243 if draft.sender_id != identity.user_id {
244 return Err(crate::error::ChatLayerError::PermissionDenied(
245 "only the uploader may abort this attachment".to_string(),
246 ));
247 }
248 let _ = self.storage.delete(&Self::object_path(&draft.key)).await;
249 let ctx = ChatStorageCtx::new(identity.user_id);
250 self.store.attachments().delete(&ctx, attachment_id).await?;
251 let _ = self.kv.delete(&Self::draft_key(attachment_id)).await;
252 crate::metrics::media_op("abort", "success");
253 Ok(())
254 }
255
256 pub async fn attach_to_message(
257 &self,
258 auth: &AuthBridge,
259 token: &str,
260 attachment_id: Uuid,
261 message_id: Uuid,
262 ) -> ChatLayerResult<()> {
263 let identity = auth.require_identity(token).await?;
264 self.attach_draft(identity.user_id, attachment_id, message_id)
265 .await
266 }
267
268 pub async fn attach_draft(
269 &self,
270 actor_id: Uuid,
271 attachment_id: Uuid,
272 message_id: Uuid,
273 ) -> ChatLayerResult<()> {
274 let draft = self.load_draft(attachment_id).await?;
275 if draft.sender_id != actor_id {
276 return Err(crate::error::ChatLayerError::PermissionDenied(format!(
277 "attachment {attachment_id} was uploaded by {}, not {actor_id}",
278 draft.sender_id
279 )));
280 }
281 let ctx = ChatStorageCtx::new(actor_id);
282 self.store
283 .attachments()
284 .attach_to_message(&ctx, attachment_id, message_id)
285 .await?;
286 let _ = self.kv.delete(&Self::draft_key(attachment_id)).await;
287 crate::metrics::media_op("attach", "success");
288 Ok(())
289 }
290
291 async fn load_draft(&self, attachment_id: Uuid) -> ChatLayerResult<UploadDraft> {
292 let raw = self
293 .kv
294 .get_bytes(&Self::draft_key(attachment_id))
295 .await?
296 .ok_or_else(|| {
297 crate::error::ChatLayerError::not_found(format!(
298 "attachment {attachment_id} is not a draft upload"
299 ))
300 })?;
301 Ok(serde_json::from_slice(&raw)?)
302 }
303}
304
305#[cfg(test)]
306mod tests {
307 use super::*;
308
309 #[test]
310 fn object_path_prefixes_chat() {
311 let p = MediaService::object_path("chat_message/00000000-0000-0000-0000-000000000000/x");
312 assert_eq!(
313 p.to_string(),
314 "chat/chat_message/00000000-0000-0000-0000-000000000000/x"
315 );
316 }
317
318 #[test]
319 fn draft_key_format() {
320 let id = Uuid::nil();
321 assert_eq!(MediaService::draft_key(id), format!("{DRAFT_PREFIX}{id}"));
322 }
323}
Added chat/chatlayer/metrics.rs chat/chatlayer/metrics.rs +29 −0 View chat/chatlayer/metrics.rs
@@ -0,0 +1,29 @@
1use metrics::counter;
2
3pub fn actor_spawned(role: &'static str) {
4 counter!("chat.actors_spawned_total", "role" => role).increment(1);
5}
6
7pub fn room_event_applied(result: &'static str) {
8 counter!("chat.room_events_applied_total", "result" => result).increment(1);
9}
10
11pub fn user_event_applied(result: &'static str) {
12 counter!("chat.user_events_applied_total", "result" => result).increment(1);
13}
14
15pub fn idempotency_hit() {
16 counter!("chat.idempotency_hits_total").increment(1);
17}
18
19pub fn dedup_hit() {
20 counter!("chat.event_dedup_hits_total").increment(1);
21}
22
23pub fn message_persisted(result: &'static str) {
24 counter!("chat.messages_persisted_total", "result" => result).increment(1);
25}
26
27pub fn media_op(op: &'static str, result: &'static str) {
28 counter!("chat.media_ops_total", "op" => op, "result" => result).increment(1);
29}
Added chat/chatlayer/producer.rs chat/chatlayer/producer.rs +88 −0 View chat/chatlayer/producer.rs
@@ -0,0 +1,88 @@
1use std::sync::Arc;
2
3use crate::error::{ChatLayerError, ChatLayerResult};
4use crate::events::envelope::EventEnvelope;
5use crate::events::room::ChatRoomEvent;
6use crate::events::user::ChatUserEvent;
7use crate::events::user_view::ChatUserViewEvent;
8use crate::topics;
9
10#[derive(Clone)]
11pub struct ChatProducer {
12 queue: Option<Arc<queue::Queue>>,
13 topic_prefix: Arc<str>,
14}
15
16impl std::fmt::Debug for ChatProducer {
17 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
18 f.debug_struct("ChatProducer")
19 .field("topic_prefix", &self.topic_prefix)
20 .field("enabled", &self.queue.is_some())
21 .finish()
22 }
23}
24
25impl ChatProducer {
26 pub fn new(queue: Option<Arc<queue::Queue>>, topic_prefix: impl Into<Arc<str>>) -> Self {
27 Self {
28 queue,
29 topic_prefix: topic_prefix.into(),
30 }
31 }
32
33 pub fn enabled(&self) -> bool {
34 self.queue.is_some()
35 }
36
37 pub fn queue(&self) -> Option<Arc<queue::Queue>> {
38 self.queue.clone()
39 }
40
41 fn require_queue(&self) -> ChatLayerResult<Arc<queue::Queue>> {
42 self.queue.clone().ok_or_else(|| {
43 ChatLayerError::InvalidInput("chat queue disabled; publishing unavailable".to_string())
44 })
45 }
46
47 pub async fn publish_room(
48 &self,
49 envelope: &EventEnvelope<ChatRoomEvent>,
50 ) -> ChatLayerResult<()> {
51 let queue = self.require_queue()?;
52 let topic = topics::room_topic(&self.topic_prefix);
53 let key = topics::room_key(envelope.payload.channel_id());
54 let payload = serde_json::to_vec(envelope)?;
55 queue
56 .publish_raw(&topic, Some(key.as_bytes()), &payload)
57 .await
58 .map_err(ChatLayerError::from)
59 }
60
61 pub async fn publish_user(
62 &self,
63 envelope: &EventEnvelope<ChatUserEvent>,
64 ) -> ChatLayerResult<()> {
65 let queue = self.require_queue()?;
66 let topic = topics::user_topic(&self.topic_prefix);
67 let key = topics::user_key(envelope.payload.user_id());
68 let payload = serde_json::to_vec(envelope)?;
69 queue
70 .publish_raw(&topic, Some(key.as_bytes()), &payload)
71 .await
72 .map_err(ChatLayerError::from)
73 }
74
75 pub async fn publish_user_view(
76 &self,
77 envelope: &EventEnvelope<ChatUserViewEvent>,
78 ) -> ChatLayerResult<()> {
79 let queue = self.require_queue()?;
80 let topic = topics::user_view_topic(&self.topic_prefix);
81 let key = topics::user_key(envelope.payload.user_id());
82 let payload = serde_json::to_vec(envelope)?;
83 queue
84 .publish_raw(&topic, Some(key.as_bytes()), &payload)
85 .await
86 .map_err(ChatLayerError::from)
87 }
88}
Added chat/chatlayer/service.rs chat/chatlayer/service.rs +445 −0 View chat/chatlayer/service.rs
@@ -0,0 +1,445 @@
1use std::sync::Arc;
2
3use cache::redis::RedisCache;
4use serde::{Deserialize, Serialize};
5use storagelayer::DynChatStorage;
6use storagelayer::models::conversation::Conversation;
7use storagelayer::models::message::{MessagePage, MessageQuery};
8use storagelayer::repository::ChatStorageCtx;
9use tokio_util::sync::CancellationToken;
10use uuid::Uuid;
11
12use crate::actor::room::{RoomActorCtx, spawn_room_registry};
13use crate::actor::user::{UserActorCtx, spawn_user_registry};
14use crate::auth_bridge::AuthBridge;
15use crate::config::AppChatSettings;
16use crate::consumer::WorkerSettings;
17use crate::error::{ChatLayerError, ChatLayerResult};
18use crate::events::envelope::EventEnvelope;
19use crate::events::room::{ChatRoomEvent, MentionSpec};
20use crate::events::user::ChatUserEvent;
21use crate::media::MediaService;
22use crate::producer::ChatProducer;
23use crate::storage_bridge::{ClientMessageClaim, MessageIdempotency};
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct NewMessageCmd {
27 pub channel_id: Uuid,
28 pub body: String,
29 pub reply_to: Option<Uuid>,
30 pub thread_root: Option<Uuid>,
31 pub is_thread_start: bool,
32 pub attachment_ids: Vec<Uuid>,
33 pub mentions: Vec<MentionSpec>,
34 pub client_msg_id: String,
35}
36
37#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
38pub struct AcceptedMsg {
39 pub client_msg_id: String,
40 pub server_msg_id: Option<Uuid>,
41}
42
43pub struct WorkerHandle {
44 pub role: &'static str,
45 handle: queue::SubscribeHandle,
46 live_actors: Box<dyn Fn() -> usize + Send + Sync>,
47}
48
49impl std::fmt::Debug for WorkerHandle {
50 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51 f.debug_struct("WorkerHandle")
52 .field("role", &self.role)
53 .field("live_actors", &self.live_actors())
54 .finish()
55 }
56}
57
58impl WorkerHandle {
59 pub fn live_actors(&self) -> usize {
60 (self.live_actors)()
61 }
62
63 pub async fn shutdown(self) -> ChatLayerResult<()> {
64 self.handle.shutdown().await?;
65 Ok(())
66 }
67}
68
69pub struct ChatService {
70 store: DynChatStorage,
71 producer: Arc<ChatProducer>,
72 auth: AuthBridge,
73 media: Arc<MediaService>,
74 settings: AppChatSettings,
75 idempotency: MessageIdempotency,
76 shutdown: CancellationToken,
77 room_ctx: Arc<RoomActorCtx>,
78 user_ctx: Arc<UserActorCtx>,
79}
80
81impl std::fmt::Debug for ChatService {
82 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83 f.debug_struct("ChatService")
84 .field("settings", &self.settings)
85 .finish_non_exhaustive()
86 }
87}
88
89impl ChatService {
90 #[allow(clippy::too_many_arguments)]
91 pub fn new(
92 store: DynChatStorage,
93 producer: Arc<ChatProducer>,
94 kv: Arc<RedisCache>,
95 auth: AuthBridge,
96 media: Arc<MediaService>,
97 settings: AppChatSettings,
98 ) -> Self {
99 let idempotency = MessageIdempotency::new(kv.clone(), settings.idempotency_ttl_seconds);
100 let shutdown = CancellationToken::new();
101 let room_ctx = Arc::new(RoomActorCtx {
102 store: store.clone(),
103 idempotency: idempotency.clone(),
104 producer: producer.clone(),
105 kv: kv.clone(),
106 media: Some(media.clone()),
107 dedup_ttl_seconds: settings.dedup_ttl_seconds,
108 });
109 let user_ctx = Arc::new(UserActorCtx {
110 kv: kv.clone(),
111 producer: producer.clone(),
112 settings: settings.clone(),
113 });
114 Self {
115 store,
116 producer,
117 auth,
118 media,
119 settings,
120 idempotency,
121 shutdown,
122 room_ctx,
123 user_ctx,
124 }
125 }
126
127 pub fn storage(&self) -> DynChatStorage {
128 self.store.clone()
129 }
130
131 pub fn media(&self) -> Arc<MediaService> {
132 self.media.clone()
133 }
134
135 pub fn auth(&self) -> &AuthBridge {
136 &self.auth
137 }
138
139
140 pub async fn send_message(
141 &self,
142 auth: &AuthBridge,
143 token: &str,
144 cmd: NewMessageCmd,
145 ) -> ChatLayerResult<AcceptedMsg> {
146 let identity = auth.require_identity(token).await?;
147 if cmd.body.trim().is_empty() && cmd.attachment_ids.is_empty() {
148 return Err(ChatLayerError::invalid(
149 "message requires a body or at least one attachment",
150 ));
151 }
152 if cmd.client_msg_id.trim().is_empty() {
153 return Err(ChatLayerError::invalid("client_msg_id is required"));
154 }
155 match self.idempotency.claim(&cmd.client_msg_id).await? {
156 ClientMessageClaim::AlreadyResolved(id) => {
157 return Ok(AcceptedMsg {
158 client_msg_id: cmd.client_msg_id,
159 server_msg_id: Some(id),
160 });
161 }
162 ClientMessageClaim::Fresh | ClientMessageClaim::InFlight => {}
163 }
164
165 let command = crate::events::room::MessageCommand {
166 channel_id: cmd.channel_id,
167 sender_id: identity.user_id,
168 body: cmd.body,
169 reply_to: cmd.reply_to,
170 thread_root: cmd.thread_root,
171 is_thread_start: cmd.is_thread_start,
172 attachment_ids: cmd.attachment_ids,
173 mentions: cmd.mentions,
174 client_msg_id: cmd.client_msg_id.clone(),
175 };
176 let envelope = EventEnvelope::new(identity.user_id, ChatRoomEvent::MessageCreated(command));
177 self.producer.publish_room(&envelope).await?;
178 Ok(AcceptedMsg {
179 client_msg_id: cmd.client_msg_id,
180 server_msg_id: None,
181 })
182 }
183
184 async fn channel_for_message(
185 &self,
186 ctx: &ChatStorageCtx,
187 message_id: Uuid,
188 ) -> ChatLayerResult<Uuid> {
189 let msg = self
190 .store
191 .messages()
192 .find_by_id(ctx, message_id)
193 .await?
194 .ok_or_else(|| ChatLayerError::not_found(format!("message {message_id}")))?;
195 Ok(msg.chat_id)
196 }
197
198 pub async fn edit_message(
199 &self,
200 auth: &AuthBridge,
201 token: &str,
202 message_id: Uuid,
203 new_body: String,
204 ) -> ChatLayerResult<()> {
205 let identity = auth.require_identity(token).await?;
206 let ctx = ChatStorageCtx::new(identity.user_id);
207 let channel_id = self.channel_for_message(&ctx, message_id).await?;
208 let envelope = EventEnvelope::new(
209 identity.user_id,
210 ChatRoomEvent::MessageEdited {
211 channel_id,
212 message_id,
213 editor_id: identity.user_id,
214 new_body,
215 },
216 );
217 self.producer.publish_room(&envelope).await
218 }
219
220 pub async fn delete_message(
221 &self,
222 auth: &AuthBridge,
223 token: &str,
224 message_id: Uuid,
225 ) -> ChatLayerResult<()> {
226 let identity = auth.require_identity(token).await?;
227 let ctx = ChatStorageCtx::new(identity.user_id);
228 let channel_id = self.channel_for_message(&ctx, message_id).await?;
229 let envelope = EventEnvelope::new(
230 identity.user_id,
231 ChatRoomEvent::MessageDeleted {
232 channel_id,
233 message_id,
234 by: identity.user_id,
235 },
236 );
237 self.producer.publish_room(&envelope).await
238 }
239
240 pub async fn add_reaction(
241 &self,
242 auth: &AuthBridge,
243 token: &str,
244 message_id: Uuid,
245 emoji: String,
246 ) -> ChatLayerResult<()> {
247 let identity = auth.require_identity(token).await?;
248 if emoji.trim().is_empty() {
249 return Err(ChatLayerError::invalid("emoji is required"));
250 }
251 let ctx = ChatStorageCtx::new(identity.user_id);
252 let channel_id = self.channel_for_message(&ctx, message_id).await?;
253 let envelope = EventEnvelope::new(
254 identity.user_id,
255 ChatRoomEvent::ReactionAdded {
256 channel_id,
257 message_id,
258 user_id: identity.user_id,
259 emoji,
260 },
261 );
262 self.producer.publish_room(&envelope).await
263 }
264
265 pub async fn remove_reaction(
266 &self,
267 auth: &AuthBridge,
268 token: &str,
269 message_id: Uuid,
270 emoji: String,
271 ) -> ChatLayerResult<()> {
272 let identity = auth.require_identity(token).await?;
273 let ctx = ChatStorageCtx::new(identity.user_id);
274 let channel_id = self.channel_for_message(&ctx, message_id).await?;
275 let envelope = EventEnvelope::new(
276 identity.user_id,
277 ChatRoomEvent::ReactionRemoved {
278 channel_id,
279 message_id,
280 user_id: identity.user_id,
281 emoji,
282 },
283 );
284 self.producer.publish_room(&envelope).await
285 }
286
287 pub async fn add_mention(
288 &self,
289 auth: &AuthBridge,
290 token: &str,
291 message_id: Uuid,
292 mentions: Vec<MentionSpec>,
293 ) -> ChatLayerResult<()> {
294 let identity = auth.require_identity(token).await?;
295 let ctx = ChatStorageCtx::new(identity.user_id);
296 let channel_id = self.channel_for_message(&ctx, message_id).await?;
297 let envelope = EventEnvelope::new(
298 identity.user_id,
299 ChatRoomEvent::MentionAdded {
300 channel_id,
301 message_id,
302 mentions,
303 },
304 );
305 self.producer.publish_room(&envelope).await
306 }
307
308 pub async fn toggle_pin(
309 &self,
310 auth: &AuthBridge,
311 token: &str,
312 message_id: Uuid,
313 pinned: bool,
314 ) -> ChatLayerResult<()> {
315 let identity = auth.require_identity(token).await?;
316 let ctx = ChatStorageCtx::new(identity.user_id);
317 let channel_id = self.channel_for_message(&ctx, message_id).await?;
318 let envelope = EventEnvelope::new(
319 identity.user_id,
320 ChatRoomEvent::PinToggle {
321 channel_id,
322 message_id,
323 by: identity.user_id,
324 pinned,
325 },
326 );
327 self.producer.publish_room(&envelope).await
328 }
329
330 pub async fn mark_read(
331 &self,
332 auth: &AuthBridge,
333 token: &str,
334 channel_id: Uuid,
335 last_read_message_id: Uuid,
336 ) -> ChatLayerResult<()> {
337 let identity = auth.require_identity(token).await?;
338 let envelope = EventEnvelope::new(
339 identity.user_id,
340 ChatUserEvent::ReadReceiptAck {
341 user_id: identity.user_id,
342 channel_id,
343 last_read_message_id,
344 ts: time::OffsetDateTime::now_utc(),
345 },
346 );
347 self.producer.publish_user(&envelope).await
348 }
349
350 pub async fn heartbeat(
351 &self,
352 auth: &AuthBridge,
353 token: &str,
354 device_id: String,
355 ) -> ChatLayerResult<()> {
356 let identity = auth.require_identity(token).await?;
357 let envelope = EventEnvelope::new(
358 identity.user_id,
359 ChatUserEvent::PresenceHeartbeat {
360 user_id: identity.user_id,
361 device_id,
362 ts: time::OffsetDateTime::now_utc(),
363 },
364 );
365 self.producer.publish_user(&envelope).await
366 }
367
368
369 pub async fn list_messages(
370 &self,
371 auth: &AuthBridge,
372 token: &str,
373 channel_id: Uuid,
374 query: MessageQuery,
375 ) -> ChatLayerResult<MessagePage> {
376 let ctx = auth.ctx_for(token).await?;
377 Ok(self
378 .store
379 .messages()
380 .list_by_conversation(&ctx, channel_id, query)
381 .await?)
382 }
383
384 pub async fn get_conversation(
385 &self,
386 auth: &AuthBridge,
387 token: &str,
388 id: Uuid,
389 ) -> ChatLayerResult<Option<Conversation>> {
390 let ctx = auth.ctx_for(token).await?;
391 Ok(self.store.conversations().find_by_id(&ctx, id).await?)
392 }
393
394 pub async fn lookup_by_client_msg_id(
395 &self,
396 auth: &AuthBridge,
397 token: &str,
398 client_msg_id: &str,
399 ) -> ChatLayerResult<Option<Uuid>> {
400 let _ = auth.require_identity(token).await?;
401 self.idempotency.lookup(client_msg_id).await
402 }
403
404
405 pub fn queue(&self) -> Option<Arc<queue::Queue>> {
406 self.producer.queue()
407 }
408
409 pub fn spawn_room_worker(self: &Arc<Self>) -> ChatLayerResult<WorkerHandle> {
410 let queue = self
411 .queue()
412 .ok_or_else(|| ChatLayerError::invalid("chat queue disabled"))?;
413 let settings = WorkerSettings::room(&self.settings);
414 let registry = spawn_room_registry(self.room_ctx.clone(), self.shutdown.clone());
415 let worker =
416 crate::consumer::room_worker::RoomWorker::new(queue, settings, registry.clone());
417 let handle = worker.run()?;
418 Ok(WorkerHandle {
419 role: "room",
420 handle,
421 live_actors: Box::new(move || registry.live_count()),
422 })
423 }
424
425 pub fn spawn_user_worker(self: &Arc<Self>) -> ChatLayerResult<WorkerHandle> {
426 let queue = self
427 .queue()
428 .ok_or_else(|| ChatLayerError::invalid("chat queue disabled"))?;
429 let settings = WorkerSettings::user(&self.settings);
430 let registry = spawn_user_registry(self.user_ctx.clone(), self.shutdown.clone());
431 let worker =
432 crate::consumer::user_worker::UserWorker::new(queue, settings, registry.clone());
433 let handle = worker.run()?;
434 Ok(WorkerHandle {
435 role: "user",
436 handle,
437 live_actors: Box::new(move || registry.live_count()),
438 })
439 }
440
441 pub async fn shutdown(&self) -> ChatLayerResult<()> {
442 self.shutdown.cancel();
443 Ok(())
444 }
445}
Added chat/chatlayer/storage_bridge.rs chat/chatlayer/storage_bridge.rs +132 −0 View chat/chatlayer/storage_bridge.rs
@@ -0,0 +1,132 @@
1use std::future::Future;
2use std::sync::Arc;
3
4use cache::redis::RedisCache;
5use uuid::Uuid;
6
7use crate::error::ChatLayerResult;
8
9pub const MSG_IDEMPOTENCY_PREFIX: &str = "chat:msg:";
10
11const PENDING: &[u8] = b"pending";
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum ClientMessageClaim {
15 Fresh,
16 AlreadyResolved(Uuid),
17 InFlight,
18}
19
20pub trait MessageIdempotencyStore: Send + Sync {
21 fn is_resolved<'a>(
22 &'a self,
23 client_msg_id: &'a str,
24 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>>;
25
26 fn resolve<'a>(
27 &'a self,
28 client_msg_id: &'a str,
29 message_id: Uuid,
30 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<()>> + Send + 'a>>;
31}
32
33#[derive(Clone)]
34pub struct MessageIdempotency {
35 kv: Arc<RedisCache>,
36 ttl_seconds: u64,
37}
38
39impl std::fmt::Debug for MessageIdempotency {
40 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41 f.debug_struct("MessageIdempotency")
42 .field("ttl_seconds", &self.ttl_seconds)
43 .finish()
44 }
45}
46
47impl MessageIdempotency {
48 pub fn new(kv: Arc<RedisCache>, ttl_seconds: u64) -> Self {
49 Self { kv, ttl_seconds }
50 }
51
52 fn key(client_msg_id: &str) -> String {
53 format!("{MSG_IDEMPOTENCY_PREFIX}{client_msg_id}")
54 }
55
56 pub async fn claim(&self, client_msg_id: &str) -> ChatLayerResult<ClientMessageClaim> {
57 let key = Self::key(client_msg_id);
58 if self
59 .kv
60 .set_nx(&key, PENDING.to_vec(), self.ttl_seconds)
61 .await?
62 {
63 return Ok(ClientMessageClaim::Fresh);
64 }
65 match self.kv.get_bytes(&key).await? {
66 Some(v) if v.len() == 16 => {
67 let mut b = [0u8; 16];
68 b.copy_from_slice(&v);
69 Ok(ClientMessageClaim::AlreadyResolved(Uuid::from_bytes(b)))
70 }
71 _ => Ok(ClientMessageClaim::InFlight),
72 }
73 }
74
75 pub async fn lookup(&self, client_msg_id: &str) -> ChatLayerResult<Option<Uuid>> {
76 let key = Self::key(client_msg_id);
77 match self.kv.get_bytes(&key).await? {
78 Some(v) if v.len() == 16 => {
79 let mut b = [0u8; 16];
80 b.copy_from_slice(&v);
81 Ok(Some(Uuid::from_bytes(b)))
82 }
83 _ => Ok(None),
84 }
85 }
86}
87
88impl MessageIdempotencyStore for MessageIdempotency {
89 fn is_resolved<'a>(
90 &'a self,
91 client_msg_id: &'a str,
92 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<bool>> + Send + 'a>> {
93 let key = Self::key(client_msg_id);
94 let kv = self.kv.clone();
95 Box::pin(async move {
96 Ok(matches!(
97 kv.get_bytes(&key).await?,
98 Some(v) if v.len() == 16
99 ))
100 })
101 }
102
103 fn resolve<'a>(
104 &'a self,
105 client_msg_id: &'a str,
106 message_id: Uuid,
107 ) -> std::pin::Pin<Box<dyn Future<Output = ChatLayerResult<()>> + Send + 'a>> {
108 let key = Self::key(client_msg_id);
109 let kv = self.kv.clone();
110 let ttl = self.ttl_seconds;
111 Box::pin(async move {
112 kv.set_bytes(&key, message_id.as_bytes().to_vec(), ttl)
113 .await?;
114 Ok(())
115 })
116 }
117}
118
119#[cfg(test)]
120mod tests {
121 use super::*;
122
123 #[test]
124 fn key_prefix() {
125 assert_eq!(MessageIdempotency::key("c-1"), "chat:msg:c-1");
126 }
127
128 #[test]
129 fn pending_bytes_are_16() {
130 assert_ne!(PENDING.len(), 16);
131 }
132}
Added chat/chatlayer/topics.rs chat/chatlayer/topics.rs +25 −0 View chat/chatlayer/topics.rs
@@ -0,0 +1,25 @@
1use uuid::Uuid;
2
3pub const ROOM_SUFFIX: &str = "room.v1";
4pub const USER_SUFFIX: &str = "user.v1";
5pub const USER_VIEW_SUFFIX: &str = "user.view.v1";
6
7pub fn room_topic(prefix: &str) -> String {
8 format!("{prefix}.{ROOM_SUFFIX}")
9}
10
11pub fn user_topic(prefix: &str) -> String {
12 format!("{prefix}.{USER_SUFFIX}")
13}
14
15pub fn user_view_topic(prefix: &str) -> String {
16 format!("{prefix}.{USER_VIEW_SUFFIX}")
17}
18
19pub fn room_key(channel_id: Uuid) -> String {
20 channel_id.to_string()
21}
22
23pub fn user_key(user_id: Uuid) -> String {
24 user_id.to_string()
25}