Real-time chat with an in-memory backend

This guide builds a real-time chat room backed entirely by an in-memory service. It demonstrates the core observable query pattern: the frontend subscribes once and receives updates as they arrive, with no polling and no manual WebSocket setup.

The backend is shown in C#, Kotlin, Java, and TypeScript. Arc for TypeScript is a source preview: no package is published to npm, and it does not have full parity with Arc on .NET. How the backends differ summarizes the per-language differences this guide explains.

🎯What you'll build

• A ChatRoom that holds each room’s message history in a source with a current value
• A single, application-wide ChatService that creates and tracks rooms by name
• A ChatMessage read model with an observable ForRoom query that delivers the room’s current history as the initial payload, then pushes updates in real time
• A SendMessage command that posts to a room and triggers a push to all subscribers
• A React page component that joins a room and renders a live chat thread

Folder structure

Chat/
├── ChatRoom.cs           ← ChatRoom state holder + ChatService singleton
├── ChatRoomPage.cs       ← ChatMessage read model + SendMessage command
└── ChatRoomPage.tsx      ← React component

This is the C# layout. Kotlin and TypeScript keep the same two backend files as .kt or .ts. Java puts each public type in its own file in the same folder.

Step 1 — The backend

The backend has four pieces:

• The room keeps its history in an observable source that always has a current value. A new subscriber receives that value first, which is how a client joining mid-conversation gets the history without a separate call.
• The chat service is one instance for the whole application. It creates a room the first time its name is used and returns the same room afterwards.
• The read model and its query describe a chat message and expose the room’s source as an observable query. Arc calls the query once per subscriber, passes the room name from the request and supplies the chat service from the application’s services.
• The command sends a message to a room. The room appends it and publishes the new history, and every subscriber of that room receives it.

C#

using System.Collections.Concurrent;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Cratis.Arc.Commands.ModelBound;
using Cratis.Arc.Queries.ModelBound;
using Cratis.Concepts;

namespace MyApp.Chat;

// Chat/ChatRoom.cs
public class ChatRoom
{
    readonly object _lock = new();
    readonly List<ChatMessage> _history = [];

    // Holds the room's full history and hands it to every new subscriber.
    public BehaviorSubject<IEnumerable<ChatMessage>> Messages { get; } = new([]);

    public void Send(string user, string message)
    {
        lock (_lock)
        {
            _history.Add(new ChatMessage(ChatMessageId.New(), user, DateTimeOffset.UtcNow, message));
            Messages.OnNext([.. _history]);
        }
    }
}

// Register as a singleton: builder.Services.AddSingleton<ChatService>();
public class ChatService
{
    readonly ConcurrentDictionary<string, ChatRoom> _rooms = new();

    public ChatRoom GetChatRoom(string name) => _rooms.GetOrAdd(name, _ => new ChatRoom());
}

// Chat/ChatRoomPage.cs
public record ChatMessageId(Guid Value) : ConceptAs<Guid>(Value)
{
    public static ChatMessageId New() => new(Guid.NewGuid());
}

[ReadModel]
public record ChatMessage(ChatMessageId Id, string User, DateTimeOffset SentAt, string Message)
{
    public static ISubject<IEnumerable<ChatMessage>> ForRoom(string roomName, ChatService chatService)
    {
        var room = chatService.GetChatRoom(roomName);

        // A relay per subscriber, seeded with the room's current history. It follows the room only
        // while Arc is subscribed: Observable.Using ends the room subscription when the client leaves.
        var relay = new BehaviorSubject<IEnumerable<ChatMessage>>(room.Messages.Value);
        return Subject.Create<IEnumerable<ChatMessage>>(relay, Observable.Using(() => room.Messages.Subscribe(relay), _ => relay));
    }
}

[Command]
public record SendMessage(string RoomName, string User, string Message)
{
    public void Handle(ChatService chatService) => chatService.GetChatRoom(RoomName).Send(User, Message);
}

View C# snippet source on GitHub

Java

import io.cratis.arc.artifacts.Command;
import io.cratis.arc.artifacts.FromServices;
import io.cratis.arc.artifacts.ReadModel;
import io.cratis.arc.authorization.AllowAnonymous;
import io.cratis.arc.concepts.ConceptAs;
import io.cratis.arc.queries.ObservableState;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Flow;
import org.springframework.stereotype.Component;

// Chat/ChatRoom.java
public final class ChatRoom {
    private final Object lock = new Object();

    // Holds the room's full history and hands it to every new subscriber.
    private final ObservableState<List<ChatMessage>> messages = new ObservableState<>(List.of());

    public Flow.Publisher<List<ChatMessage>> messages() {
        return messages;
    }

    public void send(String user, String message) {
        synchronized (lock) {
            var history = new ArrayList<>(messages.get());
            history.add(new ChatMessage(ChatMessageId.newId(), user, OffsetDateTime.now(ZoneOffset.UTC), message));
            messages.set(List.copyOf(history));
        }
    }
}

// Chat/ChatService.java
// A Spring component, so the application has one ChatService.
@Component
public final class ChatService {
    private final Map<String, ChatRoom> rooms = new ConcurrentHashMap<>();

    public ChatRoom getChatRoom(String name) {
        return rooms.computeIfAbsent(name, key -> new ChatRoom());
    }
}

// Chat/ChatMessageId.java
public record ChatMessageId(UUID value) implements ConceptAs<UUID> {
    public static ChatMessageId newId() {
        return new ChatMessageId(UUID.randomUUID());
    }
}

// Chat/ChatMessage.java
@ReadModel
@AllowAnonymous
public record ChatMessage(ChatMessageId id, String user, OffsetDateTime sentAt, String message) {
    public static Flow.Publisher<List<ChatMessage>> forRoom(String roomName, @FromServices ChatService chatService) {
        return chatService.getChatRoom(roomName).messages();
    }
}

// Chat/SendMessage.java
@Command
@AllowAnonymous
public record SendMessage(String roomName, String user, String message) {
    public void handle(@FromServices ChatService chatService) {
        chatService.getChatRoom(roomName).send(user, message);
    }
}

View Java snippet source on GitHub

Kotlin

import io.cratis.arc.artifacts.Command
import io.cratis.arc.artifacts.FromServices
import io.cratis.arc.artifacts.ReadModel
import io.cratis.arc.authorization.AllowAnonymous
import io.cratis.arc.concepts.ConceptAs
import java.time.OffsetDateTime
import java.time.ZoneOffset
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import org.springframework.stereotype.Component

// Chat/ChatRoom.kt
class ChatRoom {
    private val history = MutableStateFlow<List<ChatMessage>>(emptyList())

    // Holds the room's full history and hands it to every new subscriber.
    val messages: StateFlow<List<ChatMessage>> = history.asStateFlow()

    fun send(user: String, message: String) {
        history.update { it + ChatMessage(ChatMessageId.new(), user, OffsetDateTime.now(ZoneOffset.UTC), message) }
    }
}

// A Spring component, so the application has one ChatService.
@Component
class ChatService {
    private val rooms = ConcurrentHashMap<String, ChatRoom>()

    fun getChatRoom(name: String): ChatRoom = rooms.computeIfAbsent(name) { ChatRoom() }
}

// Chat/ChatRoomPage.kt
data class ChatMessageId(private val id: UUID) : ConceptAs<UUID> {
    override fun value(): UUID = id

    companion object {
        fun new(): ChatMessageId = ChatMessageId(UUID.randomUUID())
    }
}

@ReadModel
@AllowAnonymous
data class ChatMessage(val id: ChatMessageId, val user: String, val sentAt: OffsetDateTime, val message: String) {
    companion object {
        @JvmStatic
        fun forRoom(roomName: String, @FromServices chatService: ChatService): Flow<List<ChatMessage>> =
            chatService.getChatRoom(roomName).messages
    }
}

@Command
@AllowAnonymous
data class SendMessage(val roomName: String, val user: String, val message: String) {
    fun handle(@FromServices chatService: ChatService) {
        chatService.getChatRoom(roomName).send(user, message)
    }
}

View Kotlin snippet source on GitHub

TypeScript (preview)

Chat/ChatRoom.ts

import { singleton } from '@cratis/arc.core';
import { BehaviorSubject } from 'rxjs';
// A type-only import: ChatRoomPage.ts imports ChatService from this file at runtime, so a runtime
// import back would be a cycle that fails when Arc's discovery loads the files.
import type { ChatMessage } from './ChatRoomPage.js';

export class ChatRoom {
    // Holds the room's full history and hands it to every new subscriber.
    readonly messages = new BehaviorSubject<ChatMessage[]>([]);

    send(message: ChatMessage): void {
        this.messages.next([...this.messages.value, message]);
    }
}

// One ChatService for the application.
@singleton()
export class ChatService {
    readonly #rooms = new Map<string, ChatRoom>();

    getChatRoom(name: string): ChatRoom {
        let room = this.#rooms.get(name);
        if (!room) {
            room = new ChatRoom();
            this.#rooms.set(name, room);
        }
        return room;
    }
}

// Chat/ChatRoomPage.ts
import { ConceptAs, field, Guid } from '@cratis/fundamentals';
import { argument, command, inject, query, readModel, service } from '@cratis/arc.core';
import { BehaviorSubject } from 'rxjs';
import { ChatService } from './ChatRoom.js';

export class ChatMessageId extends ConceptAs<Guid> {
    static readonly valueType = Guid;

    static create(): ChatMessageId {
        return new ChatMessageId(Guid.create());
    }
}

@readModel()
export class ChatMessage {
    @field(ChatMessageId) id: ChatMessageId;
    @field(String) user: string;
    @field(Date) sentAt: Date;
    @field(String) message: string;

    constructor(id = ChatMessageId.create(), user = '', sentAt = new Date(), message = '') {
        this.id = id;
        this.user = user;
        this.sentAt = sentAt;
        this.message = message;
    }

    @query({ observable: true }, argument('roomName', String), service(ChatService))
    static forRoom(roomName: string, chatService: ChatService): BehaviorSubject<ChatMessage[]> {
        return chatService.getChatRoom(roomName).messages;
    }
}

@command()
export class SendMessage {
    @field(String) roomName!: string;
    @field(String) user!: string;
    @field(String) message!: string;

    @inject(ChatService)
    handle(chatService: ChatService): void {
        chatService.getChatRoom(this.roomName).send(new ChatMessage(ChatMessageId.create(), this.user, new Date(), this.message));
    }
}

View TypeScript (preview) snippet source on GitHub

What is happening here?

The source with a current value is the only part that differs by backend:

• C#: a BehaviorSubject<IEnumerable<ChatMessage>> from System.Reactive.
• Kotlin: a MutableStateFlow<List<ChatMessage>>, returned from the query as a Flow.
• Java: Arc’s ObservableState<List<ChatMessage>>, returned as a Flow.Publisher.
• TypeScript: an RxJS BehaviorSubject<ChatMessage[]>.

Because the source has a current value, every new subscription starts from the room’s history. Arc for Kotlin, Java and TypeScript also answers a one-off GET of the query from that value instead of reporting that no result is ready yet.

Send appends the message and publishes a copy of the complete list. Sending the whole list keeps the backend straightforward; Arc sends each subscriber only what changed (see Step 2). The C#, Kotlin and Java rooms serialize concurrent sends, so two messages sent at once both end up in the history. TypeScript runs one handler at a time on its event loop.

The query returns the room’s source:

• C# creates a relay BehaviorSubject for each subscriber, seeded with the room’s current history. Observable.Using subscribes the relay to the room when Arc subscribes to the query and disposes that subscription when the client leaves, so relays do not pile up on the room. With direct mode (queryDirectMode on <Arc>), when Arc for C# fails to deliver to a WebSocket subscriber, it calls OnError on the subject the query returned, so the relay keeps one subscriber’s failure away from the shared room. The default multiplexed connection reports a failed subscription to the client instead.
• Kotlin, Java and TypeScript return the room’s source directly. Arc subscribes to it and unsubscribes when the client leaves; it never publishes into it.

The query parameters are told apart the way each backend binds them. roomName is a query argument; ChatService comes from the application’s services. C# resolves a parameter from dependency injection when its type is a registered service, so ChatService must be registered. Kotlin and Java mark the service parameter @FromServices, and TypeScript lists argument('roomName', String) and service(ChatService) in @query.

The chat service must be a single instance, or each request would see its own rooms:

• C#: register it with builder.Services.AddSingleton<ChatService>(); in Program.cs.
• Kotlin and Java: @Component makes it a Spring singleton.
• TypeScript: @singleton() registers it when the application discovers the module.

ChatMessageId gives every message an Id. Arc uses a property named Id (in any case) to compute what changed between two emissions, as the next step explains.

The Kotlin and Java query and command are marked @AllowAnonymous because Arc on the JVM requires an authenticated caller by default; the C# and TypeScript versions have no authorization rules.

Build the backend to generate the TypeScript proxies before writing the frontend. For C#, dotnet build runs the Arc proxy generator; the other backends use their own proxy generation commands and paths. Each produces:

• ChatMessage.ts — the TypeScript model type
• ForRoom.ts — the observable query proxy with use() and when() hooks
• SendMessage.ts — the command proxy with a use() hook

The message id is a Guid in every backend’s proxy. The frontend below only turns it into a string, so it works with any backend.

Step 2 — Delta mode

Arc observable queries use delta mode by default. Knowing what it sends helps you reason about what crosses the network.

What happens on each emission:

Emission	What is sent
First	The complete collection — the room’s full history as the initial payload
Subsequent	A ChangeSet with only the added, replaced, and removed items

The use() hook applies each ChangeSet transparently. messagesResult.data always holds the full current collection — the React component never sees raw deltas.

How Arc computes the ChangeSet. The server compares each emission with the previous one it sent to that subscriber, matching items by their Id. An item with a new Id is added, a missing Id is removed, and an Id whose content changed is replaced. Because ChatMessage has an Id and messages never change, each new message arrives as one item in added, however long the history is.

The Id matters. Without an Id property, Arc for C# sends the full collection on every emission and logs a warning, and Arc for Kotlin, Java and TypeScript compare items by their JSON instead, which cannot report a replaced item.

The server still builds and compares the whole list on every message. That is cheap for a chat room; the Incremental Pushes guide shows a backend that emits only new messages.

Full mode

To send the complete collection on every emission (useful during debugging), set observableQueryTransferMode={ObservableQueryTransferMode.Full} on the <Arc> provider. Delta mode is the default and is recommended for production.

Step 3 — The React component

Chat/ChatRoomPage.tsx

import { useState } from 'react';
import { ForRoom } from './ForRoom';
import { SendMessage } from './SendMessage';
import type { ChatMessage } from './ChatMessage';

export const ChatRoomPage = () => {
    const [roomName, setRoomName] = useState('');
    const [joinedRoom, setJoinedRoom] = useState('');
    const [user, setUser] = useState('');
    const [messageText, setMessageText] = useState('');

    const [messagesResult] = ForRoom
        .when(joinedRoom.length > 0)
        .use({ roomName: joinedRoom });

    const [sendCommand, setSendValues] = SendMessage.use();

    const handleJoin = () => {
        if (roomName.trim() && user.trim()) {
            setJoinedRoom(roomName.trim());
        }
    };

    const handleSend = async () => {
        if (!messageText.trim()) return;
        setSendValues({ roomName: joinedRoom, user, message: messageText });
        await sendCommand.execute();
        setMessageText('');
    };

    if (!joinedRoom) {
        return (
            <div style={{ maxWidth: 400, margin: '80px auto', display: 'flex', flexDirection: 'column', gap: 12 }}>
                <h2>Join a Chat Room</h2>
                <input
                    placeholder="Room name"
                    value={roomName}
                    onChange={e => setRoomName(e.target.value)}
                />
                <input
                    placeholder="Your name"
                    value={user}
                    onChange={e => setUser(e.target.value)}
                />
                <button
                    onClick={handleJoin}
                    disabled={!roomName.trim() || !user.trim()}
                >
                    Join
                </button>
            </div>
        );
    }

    const messages: ChatMessage[] = messagesResult.data ?? [];

    return (
        <div style={{ maxWidth: 600, margin: '40px auto', display: 'flex', flexDirection: 'column', gap: 16 }}>
            <h2>{joinedRoom}</h2>
            <p style={{ color: '#888', margin: 0 }}>Chatting as <strong>{user}</strong></p>

            <div style={{
                border: '1px solid #e0e0e0',
                borderRadius: 8,
                height: 400,
                overflowY: 'auto',
                padding: 16,
                display: 'flex',
                flexDirection: 'column',
                gap: 8,
            }}>
                {messages.length === 0 && (
                    <p style={{ color: '#aaa', alignSelf: 'center', marginTop: 'auto', marginBottom: 'auto' }}>
                        No messages yet. Say hello!
                    </p>
                )}
                {messages.map(msg => (
                    <div
                        key={String(msg.id)}
                        style={{
                            background: msg.user === user ? '#e8f4fd' : '#f5f5f5',
                            borderRadius: 8,
                            padding: '8px 12px',
                            alignSelf: msg.user === user ? 'flex-end' : 'flex-start',
                            maxWidth: '75%',
                        }}
                    >
                        <div style={{ fontSize: 12, color: '#888', marginBottom: 2 }}>
                            <strong>{msg.user}</strong>
                            {' · '}
                            {new Date(msg.sentAt).toLocaleTimeString()}
                        </div>
                        <div>{msg.message}</div>
                    </div>
                ))}
            </div>

            <div style={{ display: 'flex', gap: 8 }}>
                <input
                    style={{ flex: 1 }}
                    placeholder="Type a message…"
                    value={messageText}
                    onChange={e => setMessageText(e.target.value)}
                    onKeyDown={e => { if (e.key === 'Enter') handleSend(); }}
                />
                <button
                    onClick={handleSend}
                    disabled={!messageText.trim()}
                >
                    Send
                </button>
            </div>
        </div>
    );
};

What is happening here?

ForRoom.when(joinedRoom.length > 0).use({ roomName: joinedRoom }) — .when(condition) prevents the subscription from opening until the user has joined a room. Once joinedRoom is set, Arc opens the subscription. The component immediately receives the room’s full history as the first push. Every subsequent SendMessage command — from any user in that room — triggers a new push, and messagesResult.data updates automatically.

setSendValues then execute() — setSendValues updates the command object with the room name, user name, and message text. sendCommand.execute() sends the HTTP POST. On the server, the command’s handler sends the message to the room, the room publishes the new history, and every subscriber receives it — including this browser.

Messages are keyed by id, which stays stable as messages arrive, so React keeps each rendered message in place. String(msg.id) works whether the proxy types the id as a Guid or a string.

Summary

Piece	What it does
ChatRoom	Holds the history in a source with a current value; Send publishes the full list
ChatService	One instance for the application; creates and tracks rooms by name
ChatMessage query ForRoom	Returns the room’s source (a per-subscriber relay in C#)
SendMessage command	Sends the message to the room, which pushes to every subscriber
ChatMessageId	Lets Arc match messages by Id when it computes deltas
Delta mode	First push = full history; later pushes = a ChangeSet with the new message
ForRoom.ts (generated)	TypeScript observable query proxy with use() and when() hooks
SendMessage.ts (generated)	TypeScript command proxy with a use() hook
ChatRoomPage.tsx	React component; subscribes on join, renders messagesResult.data

✓Recap

You built a chat room whose history lives in memory and reaches every subscriber through an observable query, with Arc sending only what changed. Next, load the history from storage and fan messages out over RabbitMQ.