gRPC-Web API & Chat Protocol

gRPC-Web API & Chat Protocol

The SkyConnect Rust streaming backend exposes its realtime chat, call-signaling, and presence features through a single gRPC service — chat.ChatService — defined in proto/chat.proto, implemented with the tonic framework, authenticated via JWT, and fronted by an nginx gRPC-Web gateway so browser clients can consume it.

Purpose and Scope

This page documents the gRPC-Web API surface of the Rust backend (streaming-backend/), treated as one coherent capability:

• The protocol contract in proto/chat.proto — service definition, RPC signatures, and every message type.
• The tonic service implementation (MyChatService in src/grpc/chat_service.rs) — how each RPC is wired to the domain layer.
• Authentication — how user identity is extracted from the JWT Authorization header on every RPC.
• The realtime event stream — the per-user broadcast relay that powers StreamMessages and SendSignal.
• Chat history, read receipts, and call signaling — how they map onto the five RPCs.

Related topics that belong to sibling pages and are only referenced here: the SQL schema in migrations/ (users, groups, messages, statuses, calls, user keys, push tokens, refresh tokens), the HTTP/auth middleware in src/middleware/ (JWT middleware, rate limiting), the domain modules under src/modules/ (chat message service/repository, call signal processor), and deployment artifacts (Dockerfile, docker-compose.yml, nginx.conf). See Related Links for pointers.

Overview

The backend is a Rust service built on tonic (the gRPC implementation for Tokio). build.rs compiles proto/chat.proto at build time; the generated code is pulled into the crate with tonic::include_proto!("chat") and the server trait chat_proto::chat_service_server::ChatService is implemented by MyChatService.

The API is deliberately small — five RPCs cover the entire realtime surface of the product:

Key design decisions visible in the source:

• Identity is never taken from the request body. StreamMessagesRequest.user_id is explicitly documented in the proto as “informational” — the authoritative identity comes from the JWT parsed by auth::extract_user_id. This prevents spoofing.
• One event stream for everything. Chat messages, group chat, WebRTC signaling (call_offer, call_answer, ice_candidate, …), and presence all flow through the same SignalEvent message and the same per-user broadcast channel. This keeps the client connection model simple: one long-lived server-streaming call.
• Zero-valued optional fields are normalized at the boundary. Proto3 has no optional here, so receiver_id/group_id use 0 to mean “not set”; the handler converts them with history::opt(...) before they reach the domain layer.
• gRPC-Web compatibility. The server-streaming RPC and unary RPCs are all expressible over gRPC-Web; the repository ships nginx.conf as the edge gateway that translates gRPC-Web traffic to the tonic server (its exact directives were not part of this page’s source review).

Architecture

Component roles:

• MyChatService (src/grpc/chat_service.rs) is the only gRPC-facing component. It is a thin adapter: every handler authenticates the caller, translates proto types into domain types, delegates to the domain layer, and maps errors to tonic::Status.
• state.call_relay is the realtime backbone. It is a per-user broadcast relay: SendSignal publishes CallSignals into it, and StreamMessages subscribes to it as a tokio_stream::wrappers::BroadcastStream. The service never writes events directly to client streams — it always goes through the relay, which decouples producers (signal processor) from consumers (stream subscribers).
• call_signal::process_signal (in modules::call) is the signaling state machine. It validates and normalizes an incoming CallSignal and returns an optional relayed signal; only when Some is returned does the event get broadcast.
• MessageService / MessageRepo (in modules::chat) own message persistence and read-receipt updates; they are the only components that touch PostgreSQL.
• grpc::history implements cursor-based history fetching and the opt() scalar-to-Option helper used across handlers.

The layering is intentional: the gRPC layer stays free of business logic so the same domain services could later be exposed over HTTP/WebSocket without changing chat or call logic.

Protocol Contract (proto/chat.proto)

The entire API surface is defined in a single proto3 file with package name chat. The full definition follows; it is the authoritative contract for every client (gRPC and gRPC-Web):

syntax = "proto3";

package chat;

service ChatService {
  rpc SendMessage (SendMessageRequest) returns (SendMessageResponse);
  rpc StreamMessages (StreamMessagesRequest) returns (stream SignalEvent);
  rpc GetHistory (GetHistoryRequest) returns (GetHistoryResponse);
  rpc SendSignal (SendSignalRequest) returns (SendSignalResponse);
  rpc MarkRead (MarkReadRequest) returns (MarkReadResponse);
}

message SendMessageRequest {
  int64 receiver_id = 1;
  int64 group_id = 2;
  string payload = 3;
  string file_url = 4;
  string file_type = 5;
}

message SendMessageResponse {
  int64 message_id = 1;
  int64 created_at = 2;
}

// user_id is informational; identity comes from the JWT in the Authorization header
message StreamMessagesRequest {
  int64 user_id = 1;
}

// One realtime event: chat (1-1 or group), call signaling, or presence
message SignalEvent {
  string event_type = 1; // chat_message | group_chat_message | call_offer | call_answer | ice_candidate | call_busy | call_ended | call_timeout | presence
  int64 from_id = 2;     // sender / caller user id
  int64 group_id = 3;    // group scope when > 0
  string payload = 4;    // JSON payload (SDP, chat text, etc.)
  int64 created_at = 5;
}

message SendSignalRequest {
  int64 target_id = 1;
  int64 group_id = 2;
  string signal_type = 3;
  string payload = 4;
}

message SendSignalResponse {
  bool ok = 1;
}

message GetHistoryRequest {
  int64 peer_id = 1;
  int64 group_id = 2;
  int64 cursor_id = 3;
  int32 limit = 4;
}

message GetHistoryResponse {
  repeated MessageStreamItem messages = 1;
  int64 next_cursor_id = 2;
}

message MessageStreamItem {
  int64 id = 1;
  int64 sender_id = 2;
  int64 receiver_id = 3;
  int64 group_id = 4;
  string payload = 5;
  string file_url = 6;
  string file_type = 7;
  int64 created_at = 8;
}

message MarkReadRequest {
  int64 peer_id = 1;
  int64 group_id = 2;
}

message MarkReadResponse {
  int64 marked = 1;
}

Source: chat.proto

Service: ChatService

Message semantics

SendMessageRequest — a chat message. receiver_id (tag 1) selects a 1-1 peer; group_id (tag 2) selects a group; exactly one of the two should be non-zero. payload (tag 3) is the message body. file_url (tag 4) and file_type (tag 5) are declared for attachments, but note that the current handler does not propagate them into the domain SendMessagePayload (they are passed as None — see the implementation walkthrough).

SendMessageResponse — message_id (tag 1) is the persisted row id (parsed from the JSON returned by MessageService), and created_at (tag 2) is the server-side epoch-seconds timestamp produced by the now() helper.

StreamMessagesRequest — carries only an informational user_id; the comment in the proto is explicit that the JWT in the Authorization header is the source of identity. The handler ignores the field entirely and uses auth::extract_user_id instead.

SignalEvent — the single realtime envelope. event_type (tag 1) is a string enum with the documented values: chat_message, group_chat_message, call_offer, call_answer, ice_candidate, call_busy, call_ended, call_timeout, presence. from_id (tag 2) is the sender/caller; group_id (tag 3) scopes the event to a group when > 0; payload (tag 4) is a JSON string (e.g., WebRTC SDP/ICE data); created_at (tag 5) is epoch seconds. Using a string event_type instead of a proto enum is a deliberate extensibility choice — new event kinds can be shipped without a protocol change or regeneration.

SendSignalRequest — the inverse of the event stream. target_id (tag 1) is the recipient user, group_id (tag 2) a group scope, signal_type (tag 3) one of the same event kinds, and payload (tag 4) the JSON body. The handler builds a CallSignal with caller_id taken from the JWT, not from the request — the caller identity is never client-supplied.

SendSignalResponse — a single ok bool; true means the signal was accepted (and relayed when the processor returned a signal to broadcast).

GetHistoryRequest — cursor pagination: peer_id (tag 1) for a 1-1 conversation, group_id (tag 2) for a group conversation, cursor_id (tag 3) as the exclusive cursor (pass the previous response’s next_cursor_id, or 0 to start from the newest), and limit (tag 4) as the page size. A limit <= 0 falls back to 30 in the handler.

GetHistoryResponse — messages (tag 1) is the page of MessageStreamItems and next_cursor_id (tag 2) is the cursor for the next page; a value of 0 typically signals the end of history.

MessageStreamItem — the wire representation of a stored message: id (tag 1), sender_id (tag 2), receiver_id (tag 3), group_id (tag 4), payload (tag 5), file_url (tag 6), file_type (tag 7), created_at (tag 8).

MarkReadRequest — peer_id (tag 1) or group_id (tag 2) selects the conversation whose messages should be marked read for the authenticated user. MarkReadResponse returns the number of rows updated in marked (tag 1).

Service Implementation Walkthrough

The implementation lives in MyChatService in src/grpc/chat_service.rs. It holds the shared AppState (database pool, JWT secret, and the call_relay broadcast hub) and is a thin, stateless adapter over the domain layer:

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

Two helpers are worth noting before the RPC walkthrough:

• status(e) wraps any domain error into tonic::Status::internal, so the gRPC layer never leaks internal error types to clients. The trade-off is that all failures surface as INTERNAL rather than granular gRPC codes.
• now() produces epoch-seconds timestamps for created_at fields, mirroring the TIMESTAMP-style values stored by the migration layer.

The stream type used by the server-streaming RPC is declared on the trait implementation:

type StreamMessagesStream = Pin<Box<dyn Stream<Item = Result<SignalEvent, Status>> + Send>>;

Source: chat_service.rs

Every handler begins with the same authentication step: auth::extract_user_id(&request, &self.state.jwt_secret)?. The ? operator propagates a failed JWT validation as a tonic::Status, which terminates the RPC with an authentication error before any domain work happens. Identity is therefore guaranteed to be server-verified on all five RPCs, and the StreamMessagesRequest.user_id field is intentionally never consulted.

SendMessage — persisting a chat 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

Control flow:

1. Authenticate — auth::extract_user_id parses and verifies the JWT; the caller’s user_id becomes the message sender.
2. Translate — the proto scalars are coerced into the domain SendMessagePayload: history::opt(...) converts 0 (proto3’s “unset”) into Option::None for receiver_id/group_id, and the raw payload string is passed through. Notably, the proto’s file_url/file_type fields are currently dropped (None) — the attachment fields exist on the wire but are not yet wired into the domain layer.
3. Persist — MessageService::send_message writes the message and returns a JSON value containing the new row’s id as a string.
4. Map back — the id string is parsed to i64; a parse failure silently yields 0 (unwrap_or(0)).
5. Respond — created_at is stamped server-side via now(), so the client gets a canonical timestamp rather than trusting its own clock.

StreamMessages — the realtime event stream

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 heart of the realtime model:

1. Authenticate, then subscribe to the per-user broadcast channel: state.call_relay.subscribe(user_id) returns a tokio::sync::broadcast::Receiver<CallSignal> scoped to that user.
2. Wrap the receiver in BroadcastStream so it can be composed with StreamExt.
3. filter_map(|item| item.ok()) drops broadcast::error::RecvError::Lagged variants — a slow subscriber that missed messages simply skips ahead instead of killing the stream. Note there is no reconnect/resync logic at this layer; a lagging client silently misses events (history via GetHistory is the recovery path).
4. map(...) converts each CallSignal to the wire SignalEvent via signals::call_signal_to_event, wrapping it in Ok because the stream item type is Result<SignalEvent, Status>.
5. The boxed stream is returned as the server-streaming response. The connection stays open, and events flow as they are published by SendSignal or internal producers.

GetHistory — cursor pagination

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))
}

Source: chat_service.rs

• The handler normalizes limit: non-positive values become the default page size of 30.
• It delegates entirely to history::fetch_history(&self.state, user_id, group_id, peer_id, cursor_id, limit), which returns a ready-made GetHistoryResponse (messages page + next_cursor_id).
• cursor_id is passed straight through; the domain layer treats it as the exclusive cursor so clients page backwards through a conversation by echoing next_cursor_id from the previous response. This is the primary recovery path for clients that lagged on the live stream.

SendSignal — publishing call signaling / presence

async fn send_signal(&self, request: Request<SendSignalRequest>) -> Result<Response<SendSignalResponse>, Status> {
    let user_id = auth::extract_user_id(&request, &self.state.jwt_secret)?;
    let req = request.into_inner();
    let sig = CallSignal {
        target_id: req.target_id,
        caller_id: user_id,
        signal_type: req.signal_type,
        payload: req.payload,
        group_id: history::opt(req.group_id),
    };
    let processed = call_signal::process_signal(&self.state, user_id, sig).await.map_err(status)?;
    if let Some(signal) = processed {
        self.state.call_relay.send_signal(signal).await.map_err(status)?;
    }
    Ok(Response::new(SendSignalResponse { ok: true }))
}

Source: chat_service.rs

Control flow:

1. Authenticate — the JWT identity becomes caller_id. The client cannot forge who is calling.
2. Build a domain CallSignal with the request’s target_id, signal_type, payload, and normalized group_id.
3. Process — call_signal::process_signal runs the signaling logic (validation/state transitions for offers, answers, ICE candidates, busy/ended/timeout, presence). It returns Option<CallSignal>: Some when a signal should be relayed to the target, None when the signal is consumed internally (e.g., a rejected or superseded state).
4. Relay — only if Some is returned does the handler push the signal into state.call_relay.send_signal(...), which fans it out to the target user’s StreamMessages subscribers.
5. Respond — ok: true is returned unconditionally once processing succeeded; the boolean does not encode whether a relay happened.

MarkRead — read receipts

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

• Authenticates the caller, then calls MessageRepo::mark_read(db, user_id, peer, group, now()) directly — this RPC skips MessageService and talks straight to the repository.
• The conversation is selected by peer_id or group_id (normalized via history::opt); the timestamp now() is the read watermark.
• The returned marked count tells the client how many messages were updated, which the UI can use to update unread badges.

Usage Examples

Example 1: Sending a 1-1 message (client view)

The wire contract for the most common operation — a 1-1 chat message with no attachment:

message SendMessageRequest {
  int64 receiver_id = 1;   // set the peer
  int64 group_id = 2;      // leave 0 for 1-1
  string payload = 3;      // message body
  string file_url = 4;     // currently not propagated by the handler
  string file_type = 5;
}

Source: chat.proto

The client sends receiver_id = 42, payload = "hello" and receives SendMessageResponse { message_id, created_at }. The handler-side counterpart that turns this request into a domain call shows how the adapter layer works:

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

Example 2: Subscribing to the realtime stream

The server-streaming RPC returns a lazily-evaluated, boxed stream of SignalEvents. The implementation composes a Tokio broadcast receiver with stream adapters:

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

Every event type a client can observe is enumerated in the proto’s SignalEvent.event_type comment — chat messages, group chat, and the full WebRTC signaling cycle:

string event_type = 1; // chat_message | group_chat_message | call_offer | call_answer | ice_candidate | call_busy | call_ended | call_timeout | presence

Source: chat.proto

Example 3: Relaying a WebRTC offer

A caller relays an SDP offer to a callee by publishing a SendSignalRequest; the server stamps the caller identity from the JWT and broadcasts to the callee’s stream:

let sig = CallSignal {
    target_id: req.target_id,
    caller_id: user_id,            // from JWT, never from the request
    signal_type: req.signal_type,
    payload: req.payload,
    group_id: history::opt(req.group_id),
};
let processed = call_signal::process_signal(&self.state, user_id, sig).await.map_err(status)?;
if let Some(signal) = processed {
    self.state.call_relay.send_signal(signal).await.map_err(status)?;
}

Source: chat_service.rs

Example 4: Paging through history

History uses an opaque cursor. The client requests limit = 50, cursor_id = 0 for the newest page, then echoes next_cursor_id from each response:

message GetHistoryRequest {
  int64 peer_id = 1;    // conversation peer
  int64 group_id = 2;   // or group conversation
  int64 cursor_id = 3;  // 0 = newest page; else previous next_cursor_id
  int32 limit = 4;      // <= 0 falls back to 30
}

Source: chat.proto

Configuration Options

The gRPC layer itself has few tunables; the verified ones are:

Other runtime behavior (database pool, server bind address, relay channel capacity) is configured in src/config.rs / docker-compose.yml, which are covered by the deployment sibling page.

API Reference

The server trait is generated by tonic from chat.proto and implemented in src/grpc/chat_service.rs. All five methods take tonic::Request<T> and return Result<tonic::Response<U>, tonic::Status>.

send_message(&self, request: Request<SendMessageRequest>) -> Result<Response<SendMessageResponse>, Status>

Persists a 1-1 or group chat message and returns its id and server timestamp.

• Auth: required (JWT in Authorization header); the JWT subject becomes the sender.
• Request fields: receiver_id (i64, peer, 0 = unset), group_id (i64, 0 = unset), payload (string, body), file_url/file_type (string, currently ignored).
• Returns: message_id (i64, parsed from the persistence layer; 0 if parsing fails) and created_at (i64, epoch seconds from now()).
• Throws: Status from failed JWT validation; Status::internal wrapping any MessageService::send_message error.

stream_messages(&self, request: Request<StreamMessagesRequest>) -> Result<Response<Self::StreamMessagesStream>, Status>

Opens a server-streaming channel of SignalEvents for the authenticated user.

• Auth: required. The user_id request field is ignored (documented as informational).
• Returns: a Pin<Box<dyn Stream<Item = Result<SignalEvent, Status>> + Send>> fed by state.call_relay.subscribe(user_id) through a BroadcastStream with Lagged errors filtered out.
• Throws: Status from JWT validation only — stream item errors are always Ok after subscription.

get_history(&self, request: Request<GetHistoryRequest>) -> Result<Response<GetHistoryResponse>, Status>

Fetches a cursor-paginated page of messages for a 1-1 or group conversation.

• Auth: required.
• Request fields: peer_id (i64), group_id (i64), cursor_id (i64, exclusive cursor, 0 = newest page), limit (int32, clamped to default 30 when <= 0).
• Returns: GetHistoryResponse { messages: Vec<MessageStreamItem>, next_cursor_id: i64 }.
• Throws: Status from JWT validation; Status::internal from history::fetch_history.

send_signal(&self, request: Request<SendSignalRequest>) -> Result<Response<SendSignalResponse>, Status>

Publishes a call-signaling or presence event toward a target user or group.

• Auth: required; JWT subject becomes caller_id (never client-supplied).
• Request fields: target_id (i64), group_id (i64, 0 = unset), signal_type (string, one of the SignalEvent event kinds), payload (string, JSON body).
• Returns: SendSignalResponse { ok: true } once processing succeeds — ok does not indicate whether the signal was relayed (relay happens only when call_signal::process_signal returns Some).
• Throws: Status from JWT validation; Status::internal from process_signal or from call_relay.send_signal failures.

mark_read(&self, request: Request<MarkReadRequest>) -> Result<Response<MarkReadResponse>, Status>

Marks all messages in a conversation as read up to the current server time.

• Auth: required.
• Request fields: peer_id (i64) or group_id (i64) — exactly one conversation selector.
• Returns: MarkReadResponse { marked: i64 } — the number of rows updated by MessageRepo::mark_read.
• Throws: Status from JWT validation; Status::internal from repository errors.

Failure Modes, Edge Cases & Concurrency

Error handling strategy

• Uniform error mapping. The status() helper converts every domain error into tonic::Status::internal(...). Clients therefore see a single INTERNAL code for persistence, signaling, and history failures. This keeps implementation details private but sacrifices granular error codes — a deliberate simplicity trade-off visible at chat_service.rs.
• Auth failures short-circuit every handler via ? on auth::extract_user_id(...), so unauthenticated requests never reach the domain layer and never touch the database.

Edge cases

• message_id parse failure silently yields 0 (unwrap_or(0)) — a client receiving 0 cannot distinguish “newest message id” from “unknown id”. This is a known blunt edge rather than a designed contract.
• Attachment fields are dropped. SendMessageRequest.file_url/file_type exist on the wire but are passed as None to SendMessagePayload. Clients should not expect attachments to persist via this RPC yet.
• StreamMessagesRequest.user_id mismatch is ignored. The proto warns the field is informational; the handler never compares it with the JWT identity. Sending a different user_id does not subscribe to another user’s stream — the JWT decides the subscription, which is the security-relevant behavior.
• limit <= 0 → 30. History page size is silently clamped.
• cursor_id = 0 denotes “start from the newest page”; paging proceeds by echoing next_cursor_id.

Concurrency and consistency

• Realtime fan-out is a Tokio broadcast channel (state.call_relay). Producers (SendSignal handlers) and consumers (StreamMessages subscribers) run concurrently; broadcast semantics mean each event is delivered to every subscriber of the target user unless the subscriber lags.
• Lagged subscribers are dropped, not killed. BroadcastStream::new(rx).filter_map(|item| item.ok()) filters out RecvError::Lagged, so a slow client silently misses events and continues from the newest. There is no replay or backfill on the stream itself — the designed recovery path is GetHistory with cursor pagination.
• Persistence and streaming are decoupled. SendMessage writes to PostgreSQL through MessageService/MessageRepo; StreamMessages only reads the in-memory relay. A chat message is therefore not guaranteed to appear on the stream of another client unless the sending flow also publishes it to the relay (the message service owns that responsibility outside this page’s boundary).
• Read receipts are watermark-based. MarkRead stamps a single server timestamp; concurrent MarkRead calls for the same conversation are idempotent in effect (rows already at or after the watermark are not counted again).

Performance & Operational Considerations

• One long-lived connection per client. The realtime model is a single server-streaming gRPC call per user, subscribed to a per-user broadcast channel. Fan-out cost scales with the number of subscribers to a given target user, not with total users.
• Backpressure is delegated to broadcast semantics. BroadcastStream pulls from the receiver; if the client TCP window stalls, the broadcast buffer is the only safety net — beyond it, Lagged drops occur (see above). For high-volume group traffic, monitoring relay capacity and lag rates is advisable.
• GetHistory is cursor-based, avoiding offset scans on large conversation tables; the actual query plan is owned by history::fetch_history / MessageRepo (migrations 003_messages defines the backing table).
• gRPC-Web edge. The repository ships nginx.conf as the gRPC-Web gateway in front of the tonic server, and docker-compose.yml for local orchestration — these are the operational surfaces that control TLS, connection pooling, and proxy timeouts (directives not reviewed in this source pass).
• Server timestamps. now() is evaluated per call using SystemTime, so created_at values share a single epoch-seconds convention across messages, signals, and read watermarks.

Extension Points

• New event kinds without a protocol change. SignalEvent.event_type and SendSignalRequest.signal_type are plain strings. Adding a new realtime event (e.g., typing, reaction) requires only: (1) handling it in call_signal::process_signal or the producing service, and (2) documenting the string in the proto comment. No regeneration or client recompile is strictly required.
• New RPCs require a proto change. Add the rpc to ChatService in chat.proto, recompile via build.rs (tonic::include_proto!("chat") regenerates the trait), implement the new method on MyChatService, and register it on the generated server in src/grpc/server.rs.
• Stream transformation pipeline. The BroadcastStream → filter_map → map chain in stream_messages is the single place to add per-event enrichment, filtering, or authorization checks before events reach clients.
• Transport swap. Because all handlers are thin adapters over MessageService, MessageRepo, call_signal, and history, the same domain logic could be re-exposed over WebSocket or HTTP without touching the chat/call modules.

Tests

No gRPC-specific test files were found in the source pass for this page. The behavior described above is verified from the implementation itself; unit-test coverage for the service trait and the relay pipeline would be a gap worth closing (the domain services under src/modules/ and the middleware in src/middleware/ are the natural test seams).

Related Links

• chat.proto — protocol definition (this page’s contract)
• chat_service.rs — service implementation
• build.rs — proto compilation
• nginx.conf — gRPC-Web edge gateway
• docker-compose.yml — local orchestration
• Migrations — backing schema (messages, users, groups)
• For authentication middleware and rate limiting, see the src/middleware/ pages (auth_middleware.rs, rate_limit.rs).
• For the chat domain services and call signaling state machine, see the src/modules/ pages (chat/, call/).
• For deployment, configuration (src/config.rs), and Docker images, see the deployment pages.