Stateful, transport-independent realtime rooms for Vix.cpp.

vix::realtime provides the runtime foundations required to build collaborative applications, multiplayer services, live dashboards, chat systems, shared workspaces, presence systems, and other stateful realtime services in modern C++.

The module separates authoritative room logic from networking:

• commands describe client intentions;
• room handlers validate commands and produce events;
• events are persisted before being applied;
• room state is reconstructed from snapshots and event replay;
• sessions survive temporary transport disconnections;
• presence tracks logical room membership;
• transports deliver protocol envelopes without owning application logic.

Current module version:

0.1.0

The API is under active development. Public headers outside internal/ are intended to form the stable module surface.

• C++20
• CMake
• Vix.cpp core
• Vix JSON
• Vix WebSocket for the optional WebSocket adapter
• PostgreSQL and libpq for the optional PostgreSQL stores

Include the complete public API with:

#include <vix/realtime.hpp>

Advanced users may include individual module headers:

#include <vix/realtime/room.hpp>
#include <vix/realtime/room_manager.hpp>
#include <vix/realtime/session_resume.hpp>

Headers under vix/realtime/internal/ are implementation details and are not part of the stable public API.

A Realtime room follows an event-driven execution model:

RoomCommand
    |
    v
RoomHandler
    |
    v
CommandResult
    |
    v
RoomEvent persistence
    |
    v
RoomState::apply()
    |
    v
Event dispatch

The authoritative state changes only through persisted room events.

This provides:

• deterministic state transitions;
• replayable room history;
• storage-independent room logic;
• recoverable state after process restarts;
• transport-independent command handling;
• explicit event audiences;
• optimistic room-version checks.

The module uses strongly typed identifiers:

vix::realtime::RoomId roomId{"workspace/main"};
vix::realtime::SessionId sessionId{"session-42"};
vix::realtime::NodeId nodeId{"node-1"};

vix::realtime::RoomVersion version{0};
vix::realtime::EventId eventId{0};

RoomVersion identifies the logical version of room state.

EventId identifies the position of an event in one room event stream.

A RoomCommand represents one client or server intention:

vix::realtime::JsonObject payload;
payload.set_string("message", "Hello");

vix::realtime::RoomCommand command{
    vix::realtime::RoomId{"chat/general"},
    vix::realtime::SessionId{"session-42"},
    "message.send",
    std::move(payload)};

command.set_request_id("request-1");
command.set_correlation_id("conversation-9");

Commands may include an expected room version:

command.set_expected_version(
    vix::realtime::RoomVersion{12});

The expected version can be used for optimistic concurrency validation.

A RoomEvent describes an authoritative state transition:

vix::realtime::JsonObject payload;
payload.set_string("message", "Hello");

vix::realtime::RoomEvent event{
    vix::realtime::RoomId{"chat/general"},
    "message.sent",
    std::move(payload),
    vix::realtime::EventAudience::Room};

Supported audiences are:

Session-targeted events require a target session:

event.set_target_session(
    vix::realtime::SessionId{"session-84"});

event.set_audience(
    vix::realtime::EventAudience::Session);

Room handlers return a CommandResult:

return vix::realtime::CommandResult::accepted(
    {std::move(event)});

A command may be:

• accepted;
• rejected;
• ignored.

Example rejection:

return vix::realtime::CommandResult::rejected(
    vix::realtime::ErrorCode::Unauthorized,
    "session cannot modify this room");

Application state derives from RoomState.

#include <memory>
#include <string>

#include <vix/realtime.hpp>

class ChatState final
    : public vix::realtime::RoomState
{
public:
  [[nodiscard]] vix::realtime::SchemaVersion
  schema_version() const noexcept override
  {
    return 1;
  }

  void apply(
      const vix::realtime::RoomEvent &event) override
  {
    if (event.type() == "message.sent")
    {
      ++messageCount_;
    }
  }

  [[nodiscard]] vix::realtime::JsonObject
  serialize() const override
  {
    vix::realtime::JsonObject state;

    state.set_i64(
        "message_count",
        messageCount_);

    return state;
  }

  void restore(
      const vix::realtime::JsonObject &state,
      vix::realtime::SchemaVersion schemaVersion) override
  {
    if (schemaVersion != 1)
    {
      throw vix::realtime::Error{
          vix::realtime::ErrorCode::CorruptedState,
          "unsupported chat state schema"};
    }

    /*
     * Read the application fields using the Vix JSON accessors used by the
     * surrounding application.
     */
  }

  [[nodiscard]] std::unique_ptr<vix::realtime::RoomState>
  clone() const override
  {
    return std::make_unique<ChatState>(*this);
  }

private:
  std::int64_t messageCount_{0};
};

clone() is used to isolate command execution and replay work from the current authoritative state.

A handler validates commands and emits events.

class ChatHandler final
    : public vix::realtime::RoomHandler
{
public:
  [[nodiscard]] vix::realtime::CommandResult
  handle_command(
      const vix::realtime::RoomCommand &command,
      const vix::realtime::RoomState &state,
      const vix::realtime::RoomContext &context) override
  {
    static_cast<void>(state);
    static_cast<void>(context);

    if (command.type() != "message.send")
    {
      return vix::realtime::CommandResult::ignored();
    }

    vix::realtime::RoomEvent event{
        command.room_id(),
        "message.sent",
        command.payload(),
        vix::realtime::EventAudience::Room};

    event.set_source_session(
        command.session_id());

    event.set_request_id(
        command.request_id());

    event.set_correlation_id(
        command.correlation_id());

    return vix::realtime::CommandResult::accepted(
        {std::move(event)});
  }
};

Lifecycle hooks may also emit events:

vix::realtime::CommandResult on_open(
    const vix::realtime::RoomState &state,
    const vix::realtime::RoomContext &context) override;

vix::realtime::CommandResult on_join(
    const vix::realtime::SessionId &sessionId,
    const vix::realtime::RoomState &state,
    const vix::realtime::RoomContext &context) override;

vix::realtime::CommandResult on_leave(
    const vix::realtime::SessionId &sessionId,
    const vix::realtime::RoomState &state,
    const vix::realtime::RoomContext &context) override;

vix::realtime::CommandResult on_close(
    const vix::realtime::RoomState &state,
    const vix::realtime::RoomContext &context) override;

A RoomFactory creates a state and handler for one room type.

class ChatFactory final
    : public vix::realtime::RoomFactory
{
public:
  [[nodiscard]] std::string_view
  room_type() const noexcept override
  {
    return "chat";
  }

  [[nodiscard]] vix::realtime::RoomStatePtr
  create_state(
      const vix::realtime::RoomId &roomId) const override
  {
    static_cast<void>(roomId);

    return std::make_unique<ChatState>();
  }

  [[nodiscard]] vix::realtime::RoomHandlerPtr
  create_handler(
      const vix::realtime::RoomId &roomId) const override
  {
    static_cast<void>(roomId);

    return std::make_unique<ChatHandler>();
  }
};

Server is the transport-independent runtime facade.

#include <memory>

#include <vix/realtime.hpp>

int main()
{
  vix::realtime::Config config;

  config.maxActiveRooms = 1000;
  config.maxSessions = 10000;
  config.maxSessionsPerRoom = 256;
  config.enableSessionResume = true;
  config.enablePresence = true;

  auto server =
      std::make_shared<vix::realtime::Server>(
          vix::realtime::NodeId{"node-1"},
          config);

  server->register_factory(
      std::make_shared<ChatFactory>());

  server->start();

  auto room =
      server->open_room(
          vix::realtime::RoomId{"chat/general"},
          "chat");

  return 0;
}

Server does not open sockets. A transport adapter forwards connections and protocol envelopes to the runtime.

A session is a logical client identity that may survive a temporary transport disconnection.

auto session =
    server->create_session(
        vix::realtime::SessionId{"session-42"},
        "user-17");

A session may:

• attach to one active connection;
• join multiple rooms;
• detach without losing memberships;
• resume within a configured window;
• hold application metadata;
• be permanently closed.

auto result =
    server->join_room(
        vix::realtime::SessionId{"session-42"},
        vix::realtime::RoomId{"chat/general"});

if (result.is_rejected())
{
  // Handle the lifecycle rejection.
}

server->leave_room(
    vix::realtime::SessionId{"session-42"},
    vix::realtime::RoomId{"chat/general"});

Commands may be executed immediately:

vix::realtime::RoomCommand command{
    vix::realtime::RoomId{"chat/general"},
    vix::realtime::SessionId{"session-42"},
    "message.send",
    payload};

auto result =
    server->execute(command);

They may also be inserted into a room queue:

const auto queueStatus =
    server->enqueue(
        std::move(command));

if (queueStatus ==
    vix::realtime::internal::CommandQueueStatus::Success)
{
  server->process_next(
      vix::realtime::RoomId{"chat/general"});
}

Queued execution preserves command ordering inside one room.

EventStore is the authoritative room event persistence contract.

Provided implementations:

• MemoryEventStore
• PostgresEventStore

auto eventStore =
    std::make_shared<
        vix::realtime::MemoryEventStore>();

The in-memory store is useful for:

• tests;
• examples;
• local development;
• ephemeral applications.

It does not survive process restarts.

vix::realtime::PostgresEventStoreOptions options;

options.connectionString =
    "host=127.0.0.1 port=5432 dbname=vix user=vix password=secret";

options.schema = "public";
options.table = "vix_realtime_events";
options.createTableIfMissing = true;

auto eventStore =
    std::make_shared<
        vix::realtime::PostgresEventStore>(
            std::move(options));

The PostgreSQL event store:

• assigns event IDs transactionally per room;
• validates contiguous room versions;
• validates contiguous event IDs during replay;
• uses room-specific PostgreSQL advisory transaction locks;
• supports atomic event batches;
• stores payload and metadata as JSONB.

PostgreSQL support must be enabled when building the module.

Snapshots reduce the number of events required to restore a room.

Provided implementations:

• MemorySnapshotStore
• PostgresSnapshotStore

vix::realtime::Config config;

config.snapshotEveryEvents = 100;
config.snapshotsToKeep = 3;
config.snapshotOnRoomClose = true;
config.restoreRoomsOnOpen = true;

vix::realtime::PostgresSnapshotStoreOptions options;

options.connectionString =
    "host=127.0.0.1 port=5432 dbname=vix user=vix password=secret";

options.schema = "public";
options.table = "vix_realtime_snapshots";
options.createTableIfMissing = true;

auto snapshotStore =
    std::make_shared<
        vix::realtime::PostgresSnapshotStore>(
            std::move(options));

Snapshots are uniquely identified by:

room_id + room_version

Replacing an existing snapshot version requires the same last_event_id.

ReplayEngine reconstructs room state from a snapshot and subsequent events.

auto replayEngine =
    vix::realtime::internal::ReplayEngine::from_config(
        config,
        eventStore,
        snapshotStore);

auto replayResult =
    replayEngine.restore(
        roomId,
        roomState);

Replay validates:

• snapshot room identity;
• snapshot stream position;
• contiguous room versions;
• contiguous event IDs;
• event count limit;
• serialized byte limit;
• replay timeout;
• event application failures.

Default limits are configured through:

config.maxReplayEvents = 1000;
config.maxReplayBytes = 4U * 1024U * 1024U;
config.replayTimeout = std::chrono::milliseconds{5000};

Presence represents ephemeral logical membership in rooms.

auto presenceStore =
    std::make_shared<
        vix::realtime::LocalPresenceStore>();

A presence record contains:

• room ID;
• session ID;
• optional node ID;
• optional connection ID;
• identity;
• joined timestamp;
• latest activity timestamp;
• detached or left timestamp;
• metadata.

Presence may be:

• present;
• detached;
• left.

Presence is not authoritative room state and should not replace persisted events.

DistributedPresence defines the contract for a presence store shared by multiple runtime nodes.

A distributed implementation must provide:

• the full PresenceStore API;
• local node identity;
• node heartbeats;
• active-node queries;
• stale-node pruning;
• per-node presence cleanup;
• backend health reporting;
• backend connectivity checks.

The module does not impose a specific distributed backend.

Possible implementations include:

• PostgreSQL;
• Redis;
• a shared key-value store;
• a message broker;
• a dedicated coordination service.

RoomDirectory tracks the runtime node responsible for one room.

auto directory =
    std::make_shared<
        vix::realtime::RoomDirectory>();

Room ownership supports:

• monotonically increasing generations;
• optional leases;
• renewal;
• transfer;
• release;
• stale-owner pruning;
• local ownership checks.

A generation prevents an older owner from reclaiming authority after a newer ownership claim has been created.

RoomDirectory is process-local. Distributed deployments should provide shared routing or coordination around this contract.

SessionResume issues and validates opaque session credentials.

auto resume =
    std::make_shared<
        vix::realtime::SessionResume>(
            server->manager());

const auto token =
    resume->issue(
        vix::realtime::SessionId{"session-42"});

After a transport disconnection:

auto result =
    resume->resume(
        vix::realtime::SessionId{"session-42"},
        token,
        replacementConnection);

Successful resumption may rotate the token:

const auto nextToken =
    result.resumeToken;

Resume validation requires:

• session resumption enabled;
• an existing logical session;
• a matching opaque token;
• a detached session;
• a non-expired resume window;
• an open replacement connection.

Default resume configuration:

config.enableSessionResume = true;
config.sessionResumeWindow =
    std::chrono::seconds{120};

Tokens use URL-safe, unpadded Base64 encoding with configurable entropy.

The Realtime protocol uses structured envelopes.

Envelope kinds are:

• request;
• response;
• event;
• error;
• snapshot;
• control.

Serialization:

const std::string text =
    vix::realtime::protocol::serialize(
        envelope);

Parsing:

const auto envelope =
    vix::realtime::protocol::parse(
        text);

Current protocol version:

1.0

Protocol envelopes may contain:

• message kind;
• protocol version;
• request ID;
• correlation ID;
• room ID;
• session ID;
• command or event type;
• room version;
• event ID;
• schema version;
• payload;
• metadata;
• error information.

Transport converts transport-specific activity into Realtime connections and protocol envelopes.

vix::realtime::TransportHandlers handlers;

handlers.onOpen =
    [](vix::realtime::ConnectionPtr connection)
    {
      // Associate the connection with a logical session.
    };

handlers.onEnvelope =
    [](vix::realtime::ConnectionPtr connection,
       const vix::realtime::protocol::Envelope &envelope)
    {
      // Route the envelope to application runtime logic.
    };

handlers.onClose =
    [](vix::realtime::ConnectionPtr connection)
    {
      // Detach the logical session.
    };

handlers.onError =
    [](vix::realtime::ConnectionPtr connection,
       vix::realtime::ErrorCode code,
       const std::string &message)
    {
      // Record or report the transport error.
    };

The abstraction allows Realtime to support WebSocket, TCP, local IPC, tests, or custom transports without changing room logic.

WebSocketAdapter bridges Vix WebSocket sessions with the Realtime transport contract.

vix::websocket::Server websocketServer;

auto adapter =
    std::make_shared<
        vix::realtime::WebSocketAdapter>(
            websocketServer);

adapter->set_handlers(
    std::move(handlers));

adapter->attach();

The adapter:

• wraps WebSocket sessions as Realtime connections;
• generates process-local connection IDs;
• parses incoming protocol envelopes;
• serializes outgoing envelopes;
• enforces an optional message-size limit;
• reports protocol and transport errors;
• may close invalid connections.

The adapter does not start or stop the underlying WebSocket server.

Metrics provides thread-safe observational counters and gauges.

auto metrics =
    std::make_shared<
        vix::realtime::Metrics>();

metrics->record_room_opened();
metrics->record_session_created();
metrics->record_events_persisted(2);

const auto snapshot =
    metrics->snapshot();

Available metrics include:

• active rooms;
• active sessions;
• attached connections;
• queued commands;
• active presence;
• opened and closed rooms;
• created and closed sessions;
• processed, accepted, rejected, and ignored commands;
• persisted events;
• event dispatch recipients and failures;
• snapshots created and restored;
• replay operations, events, bytes, and duration;
• session resume attempts;
• transport messages and bytes;
• protocol errors;
• runtime errors.

Metrics use relaxed atomics and do not participate in authoritative runtime behavior.

HealthMonitor creates a point-in-time runtime report.

auto monitor =
    std::make_shared<
        vix::realtime::HealthMonitor>(
            server,
            metrics);

const auto report =
    monitor->check();

Health states are:

• Healthy
• Degraded
• Unhealthy
• Stopped

The report inspects:

• server lifecycle;
• event-store availability;
• snapshot-store availability;
• presence-store availability;
• room-directory availability;
• room lifecycle states;
• local room ownership;
• command queue depth;
• connected and detached sessions;
• retained closed sessions;
• presence count;
• runtime and protocol error counters.

Example:

if (!report.operational())
{
  for (const auto &issue : report.issues)
  {
    // Report the issue.
  }
}

Default runtime configuration:

Validate configuration before constructing custom runtime components:

vix::realtime::Config config;
config.validate();

Realtime operations use vix::realtime::Error.

try
{
  server->open_room(
      vix::realtime::RoomId{"chat/general"},
      "chat");
}
catch (const vix::realtime::Error &error)
{
  const auto code =
      error.code();

  const auto name =
      vix::realtime::to_string(code);
}

Important error categories include:

• invalid configuration;
• missing dependencies;
• room lifecycle failures;
• session expiration;
• invalid resume tokens;
• command rejection;
• queue saturation;
• payload limits;
• event-store failures;
• snapshot-store failures;
• replay failures;
• transport failures;
• protocol errors;
• timeouts;
• corrupted state.

Commands processed by one room are serialized through the room command queue and room execution lock.

Produced events are persisted before they are applied to authoritative room state.

Events are dispatched only after successful persistence and state application.

A command producing multiple events commits them as one store batch where the selected event-store implementation supports atomic batches.

State restoration rejects gaps or inconsistencies in room versions and event identifiers.

Logical sessions may remain alive after a transport connection disappears.

Application state and handlers do not depend on WebSocket types.

A single-process deployment may use:

RoomManager
MemoryEventStore
MemorySnapshotStore
LocalPresenceStore
RoomDirectory
WebSocketAdapter

A durable single-node deployment may use:

RoomManager
PostgresEventStore
PostgresSnapshotStore
LocalPresenceStore
RoomDirectory
WebSocketAdapter

A multi-node deployment additionally requires shared coordination for:

• room ownership;
• command routing;
• distributed presence;
• connection-to-session routing;
• event delivery across nodes.

DistributedPresence defines the shared-presence contract, while RoomOwner and RoomDirectory define the ownership model used by future distributed coordination implementations.

HTTP / WebSocket runtime
          |
          v
   WebSocketAdapter
          |
          v
        Server
          |
          v
     RoomManager
      /   |    \
     /    |     \
 Rooms  Sessions  Presence
   |        |
   v        v
Handlers  Connections
   |
   v
EventStore + SnapshotStore

Application code should place business rules in:

• RoomState;
• RoomHandler;
• RoomFactory.

Transport callbacks should remain focused on:

• authentication;
• session creation;
• session resumption;
• envelope routing;
• connection detachment.

The module is designed so most application logic can be tested without sockets.

Recommended test layers:

1. Test RoomState::apply() with explicit events.
2. Test RoomHandler::handle_command() with explicit state and context.
3. Test Room with memory stores.
4. Test RoomManager session and membership behavior.
5. Test protocol serialization and parsing.
6. Test transport adapters with connection doubles.
7. Run PostgreSQL integration tests separately.

Example memory-backed setup:

auto eventStore =
    std::make_shared<
        vix::realtime::MemoryEventStore>();

auto snapshotStore =
    std::make_shared<
        vix::realtime::MemorySnapshotStore>();

auto presenceStore =
    std::make_shared<
        vix::realtime::LocalPresenceStore>();

auto directory =
    std::make_shared<
        vix::realtime::RoomDirectory>();

Vix Realtime is distributed under the MIT License.

Copyright 2026, Gaspard Kirira.

Vix.cpp https://github.com/vixcpp/vix