Realtime

The real-time engine allows you to subscribe to collection changes and receive instant updates via WebSocket connections.

Subscribing to Collections

Basic Subscription

import asyncio
from vski import VskiClient

async def main():
    async with VskiClient("http://127.0.0.1:3000") as client:
        res = await client.admins.auth_with_password(
            "admin@yourdomain.com", "your-password"
        )
        client.set_token(res["token"])

        # Subscribe to a collection (the callback may be sync or async)
        def on_event(event):
            print("Event received:", event)

        unsubscribe = client.collection("posts").subscribe(on_event)

        # Event structure:
        # {
        #     "action": "create" | "update" | "delete",
        #     "record": {"id": ..., "title": ...},  # the record that changed
        # }

asyncio.run(main())

Unsubscribing

# Stop receiving events
unsubscribe()

Event Types

VSKI emits three types of events:

Create Events

def on_event(event):
    if event["action"] == "create":
        print("New post created:", event["record"])

client.collection("posts").subscribe(on_event)

Update Events

def on_event(event):
    if event["action"] == "update":
        print("Post updated:", event["record"])

client.collection("posts").subscribe(on_event)

Delete Events

def on_event(event):
    if event["action"] == "delete":
        print("Post deleted:", event["record"]["id"])

client.collection("posts").subscribe(on_event)

Filtering Subscriptions

Subscribe only to specific records matching a filter:

# Only receive events for published posts
client.collection("posts").subscribe(
    lambda event: print("Published post changed:", event["record"]),
    {"filter": "published = true"},
)

Multiple Subscriptions

You can subscribe to multiple collections simultaneously:

unsub1 = client.collection("posts").subscribe(
    lambda event: print("Post event:", event)
)
unsub2 = client.collection("comments").subscribe(
    lambda event: print("Comment event:", event)
)

# Unsubscribe from all when done
unsub1()
unsub2()

Real-time Groups

Real-time groups allow you to distribute events among multiple consumers using groups:

Subscribing to Groups

When multiple clients subscribe to the same collection with a group, events are distributed among them:

# Client 1 subscribes to "worker-group"
client1.collection("posts").subscribe(
    lambda event: print("Client 1 received event:", event),
    {"group": "worker-group"},
)

# Client 2 also subscribes to "worker-group"
client2.collection("posts").subscribe(
    lambda event: print("Client 2 received event:", event),
    {"group": "worker-group"},
)

# Events are distributed between client 1 and client 2

Default Group (Fan-out)

If no group is specified, all clients receive all events:

# All clients receive every event
client.collection("posts").subscribe(lambda event: print("Received event:", event))

Acknowledging Events

When using groups, you can acknowledge events to receive the next one:

# When subscribing with a group, each event carries an ack ID on the wire.
# Process the event inside your callback; sending the ACK frame (and
# resuming from the last seen event after a reconnect) is handled
# automatically by the SDK.
client.collection("posts").subscribe(on_event, {"group": "worker-group"})

WebSocket Connection Management

The SDK manages WebSocket connections automatically. When you subscribe to a collection, the SDK:

1. Establishes a WebSocket connection if not already connected
2. Sends subscription messages to the server
3. Automatically reconnects if the connection is lost
4. Resends subscriptions after reconnection

Automatic Connection

WebSocket connections are established automatically when you subscribe:

# This automatically establishes the WebSocket connection if needed
unsubscribe = client.collection("posts").subscribe(
    lambda event: print("Event:", event)
)

Disconnection

Close the client to stop all WebSocket connections:

# Stop all subscriptions and close the WebSocket
# (leaving the `async with VskiClient(...)` block does the same)
await client.aclose()

Practical Examples

Chat Application

# Send a message
await client.collection("messages").create(
    {"roomId": "room-123", "userId": "user-123", "text": "Hello!"}
)

# Subscribe to new messages in the room
def on_message(event):
    if event["action"] == "create" and event["record"]["roomId"] == "room-123":
        display_message(event["record"])

client.collection("messages").subscribe(
    on_message, {"filter": "roomId = 'room-123'"}
)

Live Dashboard

# Subscribe to metrics collection
def on_metric(event):
    if event["action"] == "update":
        update_dashboard(event["record"])

client.collection("metrics").subscribe(on_metric)

# Update metrics
await client.collection("metrics").update(
    "metric-id", {"value": calculate_new_value()}
)

Collaborative Editing

# Document changes
def on_document(event):
    if event["action"] == "update" and event["record"]["id"] == current_doc_id:
        # Update local document with remote changes
        merge_changes(event["record"]["content"])

client.collection("documents").subscribe(on_document)

Notification System

# Subscribe to user notifications
def on_notification(event):
    if event["action"] == "create" and event["record"]["userId"] == current_user_id:
        show_notification(event["record"])

client.collection("notifications").subscribe(
    on_notification, {"filter": f"userId = '{current_user_id}'"}
)

Performance Considerations

Rate Limiting

Avoid overwhelming the client with rapid updates:

import time

class TokenBucket:
    """Allow at most `rate` updates per second, drop the rest."""

    def __init__(self, rate: float):
        self.rate = rate          # tokens (updates) per second
        self.tokens = 1.0
        self.updated = time.monotonic()

    def allow(self) -> bool:
        now = time.monotonic()
        self.tokens = min(1.0, self.tokens + (now - self.updated) * self.rate)
        self.updated = now
        if self.tokens >= 1.0:
            self.tokens -= 1.0
            return True
        return False

bucket = TokenBucket(rate=1.0)  # limit UI updates to once per second

def on_metric(event):
    if bucket.allow():
        update_ui(event["record"])

client.collection("metrics").subscribe(on_metric)

Batch Processing

Process multiple events in batches:

event_queue: asyncio.Queue = asyncio.Queue()

def on_event(event):
    event_queue.put_nowait(event)

client.collection("posts").subscribe(on_event)

async def process_batches(interval: float = 1.0):
    while True:
        await asyncio.sleep(interval)
        batch = []
        while not event_queue.empty():
            batch.append(event_queue.get_nowait())
        if batch:
            await process_batch(batch)

batcher = asyncio.create_task(process_batches())

# ... later, stop batching:
# batcher.cancel()

Selective Subscriptions

Only subscribe to collections you need:

# Unsubscribe when done
unsubscribe = client.collection("posts").subscribe(
    lambda event: print("Event:", event)
)

# Later, stop receiving events
unsubscribe()

Error Handling

def on_event(event):
    try:
        print("Event:", event)
    except Exception as error:
        # An exception raised in a callback is logged by the SDK and never
        # breaks the WebSocket connection
        print("Handler error:", error)

unsubscribe = client.collection("posts").subscribe(
    on_event, {"filter": "published = true"}
)

The Python SDK takes no per-subscription error callback: connection errors are logged internally and the socket reconnects automatically with all subscriptions resent.

Authentication

WebSocket connections use the same authentication as HTTP requests. The SDK automatically includes your authentication token in the WebSocket connection URL:

# Set authentication token
client.set_token("your-jwt-token")

# The WebSocket connection will include the token automatically
unsubscribe = client.collection("posts").subscribe(
    lambda event: print("Event:", event)
)

Testing Real-time Features

# In tests, you can simulate real-time events
import asyncio
from vski import VskiClient

async def test_realtime():
    async with VskiClient("http://127.0.0.1:3000") as client:
        res = await client.admins.auth_with_password(
            "admin@yourdomain.com", "your-password"
        )
        client.set_token(res["token"])

        received = {}

        def on_event(event):
            received["event"] = event

        unsubscribe = client.collection("posts").subscribe(on_event)
        await asyncio.sleep(1.0)  # allow WebSocket connect + SUBSCRIBE

        # Create a record
        await client.collection("posts").create({"title": "Test Post"})

        # Wait for the event
        for _ in range(15):
            if "event" in received:
                break
            await asyncio.sleep(0.2)

        unsubscribe()

        assert received["event"]["action"] == "create"
        assert received["event"]["record"]["title"] == "Test Post"

Security

Authentication Required

All real-time subscriptions require authentication:

# Unauthenticated connections are rejected by the server
client = VskiClient("http://127.0.0.1:3000")
# client.set_token("your-token")  # Must set token before subscribing

client.collection("posts").subscribe(lambda event: print(event))
# Without a token the server refuses the socket: no events arrive and the
# SDK keeps retrying (connection errors are logged, not raised)

Authorization

Real-time events respect collection rules, which are declared in the IaC YAML (see Rules and Access Control). A user who cannot see a record will not receive events for it:

vski/default/collections.yaml

collections:
  posts:
    type: base
    fields:
      - {name: title, type: text, required: true}
      - {name: userId, type: text, required: true}
    rules:
      list: "userId = @request.auth.id"
      view: "userId = @request.auth.id"

WebSocket Protocol

Connection URL

ws://localhost:3000/api/realtime?db=default&auth=your-token

The SDK automatically constructs this URL based on your base_url, db_name, and authentication token.

Subscription Messages

The SDK sends subscription messages to the server:

{
  "type": "SUBSCRIBE",
  "collection": "posts",
  "filter": "published = true",
  "group": "worker-group"
}

Event Messages

The server sends event messages to subscribed clients:

{
  "type": "EVENT",
  "collection": "posts",
  "events": [
    {
      "id": "event-id",
      "action": "create",
      "ackId": "ack-id",
      "data": {
        "id": "record-id",
        "title": "Hello",
        ...
      }
    }
  ]
}

ACK Messages

When using groups, clients can acknowledge events:

{
  "type": "ACK",
  "collection": "posts",
  "group": "worker-group",
  "id": "ack-id"
}

Best Practices

1. Always unsubscribe - Remove subscriptions when components unmount
2. Use filters - Reduce unnecessary event traffic
3. Handle errors - Implement proper error handling for connections
4. Limit subscriptions - Don’t subscribe to everything at once
5. Use presence wisely - Track only what you need
6. Optimize reconnection - Configure appropriate reconnection intervals
7. Rate limit updates - Prevent overwhelming clients with rapid changes