Real-Time Synchronisation

The platform WebSocket gateway pushes live UI updates to all connected browsers. When data changes (tasks created, apps installed, members added), every user viewing affected content sees the update instantly — no page refresh needed.

Architecture

Browser tab ──ws──▶ api.saaslivery.com/ws (platform WebSocket)
                         │
                    _LOCAL_CONNS (per-worker in-memory registry)
                         │
                    PG LISTEN/NOTIFY on 'ws_push' channel
                         │
                    All workers deliver to their local connections

• One connection per browser tab — base.html opens the socket for authenticated users
• Cross-worker fan-out via PostgreSQL LISTEN/NOTIFY — zero new infrastructure
• One dedicated PG connection per worker (outside the pool) for LISTEN

Event format

Events sent to browsers use the same Amebo event name as kind:

{"kind": "core.tasks.task.created", "data": {"board_id": "brd_123"}}

• kind — the canonical event name (same as events.yml)
• data — minimal payload for the browser to react (just enough to know what to refresh, not the full resource)

Server-side: pushing events

Import from core.realtime and call one of three functions after a mutation:

Push to one member

from core.realtime import push

# After assigning a task, notify the assignee
await push(workspace_id, assignee_mrn, {
    "kind": "core.tasks.task.assigned",
    "data": {"board_id": board_id, "task_id": task_id},
})

Push to entire workspace

from core.realtime import push_workspace

# After admin installs an app, everyone should see it
await push_workspace(workspace_id, {
    "kind": "core.account.application.installed",
    "data": {"slug": "wiki"},
})

Push to specific members

from core.realtime import push_many

# After adding members to a chat, notify just those members
await push_many(workspace_id, [mrn_1, mrn_2, mrn_3], {
    "kind": "core.chat.participant.added",
    "data": {"chat_id": chat_id},
})

Excluding the actor

All push functions accept exclude_mrn to avoid sending the event back to the user who triggered it (their UI already updated via the API response):

await push_workspace(workspace_id, {
    "kind": "core.account.member.removed",
    "data": {"member_id": removed_id},
}, exclude_mrn=actor_mrn)

Where to call push

Call push() in your API handler after the mutation succeeds, alongside ctx_emit():

async def api_create_task(req: Request, res: Response, ctx: Context):
    # ... create task in database ...

    # Emit Amebo event (server-to-server, cross-app)
    ctx_emit(ctx, "core.tasks.task.created",
        task_id=new_id, board_id=board_id, title=title,
    )

    # Push to browsers (server-to-client, real-time UI)
    await push_workspace(ctx.workspace_id, {
        "kind": "core.tasks.task.created",
        "data": {"board_id": board_id},
    }, exclude_mrn=f"mrn:people:member:{ctx.member_id}")

    return res.out(HTTPStatus.CREATED, one(task))

ctx_emit and push serve different purposes:

• ctx_emit → Amebo → other apps' server-side event handlers (reliable, retried)
• push → PG NOTIFY → browsers (instant, fire-and-forget, ephemeral)

Client-side: listening for events

JavaScript listeners

PlatformSocket (loaded in base.html) dispatches CustomEvent on document using kind as the event name:

document.addEventListener('core.tasks.task.created', function(e) {
    // e.detail contains the data payload
    console.log('New task on board:', e.detail.board_id);

    // Option A: re-fetch an HTMX partial
    htmx.ajax('GET', '/partials/board/' + e.detail.board_id, {
        target: '#board-content',
        swap: 'innerHTML'
    });

    // Option B: update Alpine state directly
    // (if you have a reference to the Alpine component)
});

Alpine.js listeners

Use Alpine's @event.document syntax to listen on document:

<div x-data="{ tasks: [] }"
     @core.tasks.task.created.document="refreshBoard($event.detail)">
    <!-- board content -->
</div>

Common patterns

Re-fetch a partial (simplest — let the server render fresh HTML):

document.addEventListener('core.account.application.installed', function() {
    htmx.ajax('GET', '/partials/apps', {target: '#app-grid', swap: 'innerHTML'});
});

Update a counter (lightweight — no server round-trip):

document.addEventListener('core.chat.message.sent', function(e) {
    var badge = document.querySelector('#unread-badge');
    if (badge) badge.textContent = parseInt(badge.textContent || 0) + 1;
});

Remove an element (for deletes):

document.addEventListener('core.account.member.removed', function(e) {
    var el = document.querySelector('#member-' + e.detail.member_id);
    if (el) el.remove();
});

Bridging Amebo events to browsers

When an Amebo event handler receives a cross-app event, it can forward to browsers:

# In apps/chat/events.py — when a call starts, push to chat participants
async def on_call_started(payload, workspace_id, app):
    from core.realtime import push_many

    chat_id = payload.get("chat_id")
    participant_mrns = payload.get("participant_mrns", [])

    await push_many(workspace_id, participant_mrns, {
        "kind": "core.calls.call.started",
        "data": {"chat_id": chat_id},
    })

Payload size

PostgreSQL NOTIFY has an 8KB payload limit. Keep data minimal — IDs and flags, not full resources. The browser should re-fetch the full data via HTMX partial or API call if needed.

Good:

{"kind": "core.tasks.task.created", "data": {"board_id": "brd_123"}}

Bad:

{"kind": "core.tasks.task.created", "data": {"task": {"id": "...", "title": "...", "description": "... long text ...", "comments": [...]}}}

Connection lifecycle

• Socket connects when base.html loads (authenticated users only)
• Auto-reconnects with exponential backoff (1s → 30s max)
• On reconnect, the browser should assume it may have missed events — a full partial re-fetch on reconnect is acceptable
• window.platformSocket is the global instance

Files