Chat, Groups & Messaging Services

Chat, Groups & Messaging Services

The Chat, Groups & Messaging Services page documents the real-time and store-and-forward messaging capability of the skyconnect/streaming Rust backend: the gRPC ChatService surface, the message persistence pipeline, cursor-based history retrieval, read receipts, group membership fan-out, and the signaling/relay infrastructure that carries messages and call signals to connected clients.

Purpose and Scope

This page covers the end-to-end messaging subsystem of the Rust backend (streaming-backend/):

• The gRPC chat API implemented in src/grpc/chat_service.rs — send_message, stream_messages, get_history, send_signal, mark_read.
• The message service and repository layer (src/modules/chat/) that persists messages and conversations.
• Cursor-based history pagination (src/grpc/history.rs, src/services/query_service.rs).
• Group membership and routing (src/modules/group/), including how group messages and group-call signals are fanned out without echo.
• The relay/subscription backbone (src/services/call_relay.rs, src/services/relay_subscriber.rs) that streams events to online users.

Related capabilities that live on sibling pages: authentication/JWT issuance (rust-backend.auth), call/media signaling specifics, file upload handling (/uploads static serving), and deployment/ops topics. This page treats those only as integration points, not as their own deep-dive material.

Overview

The backend exposes a single consolidated chat protocol over gRPC, generated at build time from proto/chat.proto by build.rs. The design deliberately separates three concerns:

1. API surface — a tonic-based gRPC service (MyChatService) that authenticates every request via JWT and delegates to domain services.
2. Domain logic — the chat module’s MessageService / MessageRepo (send, mark-read, conversations) and the group module’s GroupRepo (membership lookups used for fan-out).
3. Transport/fan-out — an in-process CallRelay with per-user tokio::sync::broadcast channels, fed by a Redis pub/sub subscriber on call_signals:* channels so that multiple backend instances can share one messaging backbone.

Messages are stored in PostgreSQL in a messages table and retrieved with keyset (cursor) pagination, never OFFSET-based paging, so history queries stay efficient as the table grows.

Architecture

The flow reads as follows: every gRPC request first passes through auth::extract_user_id, which validates the bearer token against state.jwt_secret and yields the acting user id. From there the handlers fan out to the domain services — MessageService::send_message for sending, MessageRepo::mark_read for read receipts, history::fetch_history for paging — or to the relay for streaming and signaling. The RelaySubscriber bridges Redis pub/sub (call_signals:*) into the in-process CallRelay, which distributes events to each online user’s broadcast channel; stream_messages wraps that channel in a BroadcastStream and maps each signal to a SignalEvent.

Source: chat_service.rs, relay_subscriber.rs, router.rs

The gRPC Chat Service

MyChatService is the concrete tonic implementation of the ChatService trait generated from proto/chat.proto (chat_proto::chat_service_server::ChatService). It holds a single field — AppState — which provides db (PostgreSQL pool), jwt_secret, and call_relay. Two small helpers frame the whole service:

pub struct MyChatService { state: AppState }
impl MyChatService { pub fn new(state: AppState) -> Self { Self { state } } }

fn status<E: std::fmt::Display>(e: E) -> Status { Status::internal(e.to_string()) }

fn now() -> i64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64
}

Source: chat_service.rs

Design intent: status() collapses every domain error into a gRPC INTERNAL status so that internal Result errors never leak details to clients, and now() produces Unix seconds for created_at / read_at timestamps consistent with the PostgreSQL bigint columns.

send_message

async fn send_message(&self, request: Request<SendMessageRequest>) -> Result<Response<SendMessageResponse>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let req = request.into_inner();
    let payload = SendMessagePayload {
        receiver_id: history::opt(req.receiver_id),
        group_id: history::opt(req.group_id),
        payload: req.payload,
        file_url: None,
        file_type: None,
    };
    let json = MessageService::send_message(&self.state, user_id, payload).await.map_err(status)?;
    let message_id = json.get("id").and_then(|v| v.as_str()).and_then(|s| s.parse::<i64>().ok()).unwrap_or(0);
    Ok(Response::new(SendMessageResponse { message_id, created_at: now() }))
}

Source: chat_service.rs

Notes on intent:

• The caller is never allowed to spoof the sender: user_id comes from the JWT, not from the request body.
• history::opt() converts the proto convention of 0 (unset) into Option<i64>; exactly one of receiver_id (direct message) or group_id (group message) is expected to be non-zero.
• MessageService::send_message returns a JSON value; the service extracts the persisted id (falling back to 0 if the shape is unexpected) and pairs it with a server-generated created_at, so clients get an immediate, authoritative message identifier.

stream_messages

async fn stream_messages(&self, request: Request<StreamMessagesRequest>) -> Result<Response<Self::StreamMessagesStream>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let rx = self.state.call_relay.subscribe(user_id);
    let stream = BroadcastStream::new(rx)
        .filter_map(|item| item.ok())
        .map(|sig| Ok::<SignalEvent, Status>(signals::call_signal_to_event(sig)));
    Ok(Response::new(Box::pin(stream)))
}

Source: chat_service.rs

This is the real-time half of the system. call_relay.subscribe(user_id) hands out a tokio::sync::broadcast receiver scoped to that user; BroadcastStream turns it into an async stream; filter_map(|item| item.ok()) silently drops lagged/closed channel errors (a RecvError::Lagged means the client was too slow and is re-synchronized via get_history); and call_signal_to_event converts the internal CallSignal into the wire SignalEvent. The stream is boxed because the trait requires Pin<Box<dyn Stream<Item = Result<SignalEvent, Status>> + Send>>.

get_history and mark_read

async fn get_history(&self, request: Request<GetHistoryRequest>) -> Result<Response<GetHistoryResponse>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let req = request.into_inner();
    let limit = if req.limit > 0 { req.limit as i64 } else { 30 };
    let res = history::fetch_history(&self.state, user_id, req.group_id, req.peer_id, req.cursor_id, limit).await.map_err(status)?;
    Ok(Response::new(res))
}

async fn mark_read(&self, request: Request<MarkReadRequest>) -> Result<Response<MarkReadResponse>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let req = request.into_inner();
    let marked = MessageRepo::mark_read(&self.state.db, user_id, history::opt(req.peer_id), history::opt(req.group_id), now()).await.map_err(status)?;
    Ok(Response::new(MarkReadResponse { marked }))
}

Source: chat_service.rs

get_history defaults the page size to 30 when the client sends limit <= 0 and delegates the entire paging concern to history::fetch_history. mark_read stamps every matching inbound message with the current time, returning the count of rows marked — both a direct-message (peer_id) and a group (group_id) variant are supported through the same opt() convention.

Message Persistence and the send pipeline

Sending is handled by MessageService::send_message (in src/modules/chat/message_service.rs), which persists via MessageRepo and returns a JSON document containing the new message id. The messages table is queried through QueryService (src/services/query_service.rs), a thin sqlx wrapper that builds parameterized queries with PgArguments and executes them against the pool or a transaction:

let res = sqlx::query_with(sql, args).execute(db).await?;
Ok(res.rows_affected())

Source: query_service.rs

The chat module is organized as a REST-style module too — controller.rs, message_service.rs, conversation_service.rs, list.rs, push.rs, repo.rs, models.rs — and is wired into the HTTP router alongside the group module:

.merge(group::router())
.merge(chat::router())

Source: router.rs

The REST routes and the gRPC service share the same MessageRepo / MessageService, so REST and gRPC clients observe the same conversation state and read receipts.

Group routing and fan-out

Group messaging and group-call signaling rely on GroupRepo::fetch_member_ids to enumerate recipients. CallRelay::send_signal implements the fan-out with two deliberate rules that prevent signaling bugs:

if let Some(gid) = signal.group_id {
    // group_call_answer is routed only to the caller (target_id), not all
    // members — otherwise every member would treat it as accepted.
    if signal.signal_type == "group_call_answer" && signal.target_id != 0 {
        self.route_to_user(signal.target_id, &signal).await;
    } else {
        for member in GroupRepo::fetch_member_ids(&self.db, gid).await? {
            // Don't echo a signal back to the user who sent it — otherwise
            // ...
        }
    }
}

Source: call_relay.rs

Design intent of the two rules:

1. Targeted answers — group_call_answer is semantically an answer to the caller, so it is routed only to target_id; broadcasting it would make every member believe the call was accepted.
2. No echo — every other group signal is fanned out to all members except the sender, avoiding self-delivery loops in group call flows.

The relay’s event source is RelaySubscriber, which subscribes to the Redis channel pattern call_signals:* and forwards every message into the in-process relay:

let _ = sub_client.psubscribe("call_signals:*").await;
let mut msg_stream = sub_client.message_rx();
while let Ok(msg) = msg_stream.recv().await {

Source: relay_subscriber.rs

This is what makes the system horizontally scalable: any backend instance that produces a signal publishes to Redis, every instance’s subscriber ingests it, and the local CallRelay delivers it only to the users connected to that instance.

Core Flow: Sending a Message End-to-End

The same stream_messages stream also carries WebRTC call signals (send_signal RPC → call::signal::process_signal → CallRelay::send_signal), so messaging and signaling share one connection and one event stream per user.

Data Model: the messages table

The messages table (modeled by src/models/message.rs) is the single persistence point for both DMs and group messages. Its columns are visible in the exact select_cols list used by history queries:

Source: history.rs

receiver_id and group_id are mutually exclusive at the application level — exactly one is set per row — which is why the history query branches on group_id != 0. read_at is written by MessageRepo::mark_read and status/edited_at/deleted_at support future delivery-state and edit/delete features.

Cursor-Based History Pagination

history::fetch_history builds a parameterized query and delegates paging to QueryService::fetch_cursor_page with a keyset cursor on the id column:

let (where_clause, args) = if group_id != 0 {
    let mut a = PgArguments::default();
    let _ = a.add(group_id);
    ("group_id = $1", a)
} else {
    let mut a = PgArguments::default();
    let _ = a.add(user_id);
    let _ = a.add(peer_id);
    let _ = a.add(peer_id);
    let _ = a.add(user_id);
    ("(sender_id = $1 AND receiver_id = $2) OR (sender_id = $3 AND receiver_id = $4)", a)
};

let opts = CursorQueryOptions {
    select_cols: Some("id, sender_id, receiver_id, group_id, payload, file_url, file_type, status, read_at, edited_at, deleted_at, created_at"),
    table_or_join: "messages",
    where_clause,
    cursor_cols: vec!["id"],
    cursor_val: if cursor_id != 0 { Some(cursor_id) } else { None },
    limit,
};

let rows = QueryService::fetch_cursor_page::<Message>(&state.db, opts, args).await?;
let next_cursor_id = if rows.len() as i64 == limit { rows.last().map(|m| m.id).unwrap_or(0) } else { 0 };

Source: history.rs

Two decisions are worth calling out:

1. Keyset pagination, not OFFSET. The cursor is the last seen message id (cursor_cols: vec!["id"]), so paging is WHERE id < cursor ORDER BY id DESC LIMIT n — an index-friendly scan that stays O(page) regardless of how deep into history the client pages. The next_cursor_id is only emitted when a full page was returned (rows.len() == limit), signalling “there may be more”; a short page means the end of history.
2. Parameterized filters. Group history is a single group_id = $1 predicate; DM history is the symmetric pair (sender_id = $1 AND receiver_id = $2) OR (sender_id = $3 AND receiver_id = $4) so either direction of a conversation is returned regardless of who initiated it. Note the peer’s id is bound in both directions, and user_id is enforced in the predicate so a user can only page their own conversations.

Rows are mapped into MessageStreamItem (with Option fields flattened to 0/empty defaults for the wire format) and returned as GetHistoryResponse { messages, next_cursor_id }.

Usage Examples

Opening a live message stream (server side)

async fn stream_messages(&self, request: Request<StreamMessagesRequest>) -> Result<Response<Self::StreamMessagesStream>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let rx = self.state.call_relay.subscribe(user_id);
    let stream = BroadcastStream::new(rx)
        .filter_map(|item| item.ok())
        .map(|sig| Ok::<SignalEvent, Status>(signals::call_signal_to_event(sig)));
    Ok(Response::new(Box::pin(stream)))
}

Source: chat_service.rs

Sending a direct or group message (server side)

let payload = SendMessagePayload {
    receiver_id: history::opt(req.receiver_id),
    group_id: history::opt(req.group_id),
    payload: req.payload,
    file_url: None,
    file_type: None,
};
let json = MessageService::send_message(&self.state, user_id, payload).await.map_err(status)?;

Source: chat_service.rs

Fetching history with the cursor loop (server side)

let limit = if req.limit > 0 { req.limit as i64 } else { 30 };
let res = history::fetch_history(&self.state, user_id, req.group_id, req.peer_id, req.cursor_id, limit).await.map_err(status)?;
// client repeats with GetHistoryRequest { cursor_id: res.next_cursor_id } until next_cursor_id == 0

Source: chat_service.rs

API Reference (gRPC ChatService)

All RPCs authenticate via auth::extract_user_id (JWT bearer token validated against state.jwt_secret); authentication failure returns a non-OK Status before any domain work. Domain errors are flattened to Status::internal via the status() helper.

Conventions shared across the API:

• 0 is the “unset” sentinel for optional ids — converted with history::opt() to Option<i64> (None when 0). See history.rs.
• Timestamps are Unix seconds (now() in chat_service.rs).
• The proto contract is compiled at build time from proto/chat.proto with a generated client (build_client(true)), see build.rs.

Failure Modes, Edge Cases & Concurrency

• JWT failure — every RPC calls auth::extract_user_id first; an invalid/expired token aborts the call with a gRPC error status before any DB or relay access. Sender identity can never be client-supplied.
• Slow consumers — BroadcastStream::new(rx).filter_map(|item| item.ok()) drops RecvError::Lagged and RecvError::Closed variants silently. A lagged client misses events by design; it is expected to re-sync state through get_history, so the system trades per-client delivery guarantees for a simple, non-blocking fan-out.
• Group answer race — without the group_call_answer targeting rule, an answer signal would be broadcast to every member and each would treat the call as accepted; routing it only to target_id keeps acceptance unambiguous (call_relay.rs).
• Echo suppression — group fan-out skips the sender, preventing self-delivery of one’s own signals and avoiding loop amplification across multiple relay hops.
• Sentinel 0 ambiguity — receiver_id/group_id both default to 0 on the wire; opt() normalizes to None, and history picks the DM query only when group_id == 0. A client sending both fields would persist a DM-shaped row (group branch wins in history) — the API does not validate mutual exclusivity at the gRPC boundary.
• message_id parse fallback — send_message parses the JSON id with a chain of and_then/ok() and falls back to 0 if the shape is unexpected, degrading gracefully instead of failing the RPC after a successful insert.
• Cross-instance consistency — the Redis pub/sub bridge (call_signals:*) is at-least-once/best-effort: RelaySubscriber loops on msg_stream.recv() and a reconnect would re-subscribe to the pattern, but there is no durable per-user queue; offline delivery relies on get_history and mark_read rather than on stream replay.

Performance & Operational Notes

• No OFFSET paging — history uses keyset cursors on id, so deep pages stay index-friendly and O(page size) per request (history.rs).
• Per-user broadcast channels — call_relay.subscribe(user_id) gives O(1) fan-out per connected user in-process; the Redis hop is only crossed once per emitted signal, not per recipient.
• Single stream for messages + signals — stream_messages carries both chat events and call signaling, so clients keep one long-lived connection instead of two, reducing connection churn under load.
• Static uploads — file attachments are served by tower_http::services::ServeDir under /uploads and referenced via file_url/file_type on messages (router.rs); the gRPC send_message currently passes file_url: None, so attachment sends flow through the REST module.
• Default page size 30 — history requests without an explicit limit are capped at 30 rows, bounding per-request DB and payload cost.

Extension Points

• New RPCs — regenerate from proto/chat.proto via build.rs, then add a handler to MyChatService (chat_service.rs); domain work should live in the chat module services/repos so REST and gRPC stay in sync.
• New signal types — CallSignal (signal_type, target_id, group_id, payload) is the carrier for both chat events and call signaling; routing rules are centralized in call::signal::process_signal and CallRelay::send_signal, the natural place to add new group-aware behaviors.
• Delivery-state features — the messages.status, edited_at, and deleted_at columns are already present in the schema and selected by history queries; edit/delete/status features can be added in MessageRepo without schema changes.
• Multi-instance scaling — the call_signals:* Redis pattern is the pluggable bus between instances; replacing RelaySubscriber with another transport keeps the rest of the relay untouched.
• Rate limiting/auth — HTTP middleware (src/middleware/auth_middleware.rs, src/middleware/rate_limit.rs) guards REST routes and can be layered onto new endpoints consistently.

Related Links

• Source entry points: chat_service.rs, history.rs, query_service.rs, call_relay.rs, relay_subscriber.rs, router.rs
• Chat domain module: src/modules/chat/ (message_service.rs, conversation_service.rs, repo.rs, push.rs, list.rs, controller.rs)
• Group domain module: src/modules/group/ (membership via GroupRepo::fetch_member_ids)
• Data model: message.rs, group.rs
• For authentication and JWT issuance, see the Auth page; for call/media specifics, see the Calls & Signaling page; for deployment and Redis/PostgreSQL topology, see the Operations page.