mirror of
https://github.com/warmbly/warmbly.git
synced 2026-08-18 16:01:18 +00:00
feat: document the resumable gateway (resume/replay, seq, resume_failed) in realtime.mdx and publish a machine-readable AsyncAPI 3.1 spec at docs/asyncapi.json describing the org channel, join/resume, HELLO, events, intents, and presence
This commit is contained in:
Matthew Meszaros
@@ -20,6 +20,8 @@ Everything in this reference is generated from the same surface the spec describ
|
||||
|
||||
There are no official SDKs yet, so generating a client from this spec (or calling the API over plain HTTP) is the supported path today.
|
||||
|
||||
The realtime WebSocket gateway has its own machine-readable description in [AsyncAPI 3.1](https://docs.warmbly.com/asyncapi.json); see the [realtime guide](/api/realtime/).
|
||||
|
||||
## Conventions baked into the spec
|
||||
|
||||
- The single server is `https://api.warmbly.com/v1`. Every path in the spec is relative to that versioned base. See [versioning](/api/).
|
||||
|
||||
@@ -4,7 +4,13 @@ description: Subscribe to live Warmbly events over a WebSocket connection.
|
||||
icon: Radio
|
||||
---
|
||||
|
||||
Warmbly pushes dashboard events (campaign sends, opens, clicks, replies, inbox arrivals, audit entries, and more) over a WebSocket. The same socket that powers the live dashboard is available to developers.
|
||||
Warmbly pushes dashboard events (campaign sends, opens, clicks, replies, inbox arrivals, audit entries, and more) over a WebSocket. The same socket that powers the live dashboard is available to developers. Events are ordered and carry a monotonic `seq`, and a reconnecting client can resume and replay what it missed.
|
||||
|
||||
The gateway has a machine-readable AsyncAPI 3.1 description you can generate clients or docs from:
|
||||
|
||||
```
|
||||
https://docs.warmbly.com/asyncapi.json
|
||||
```
|
||||
|
||||
## Connecting
|
||||
|
||||
@@ -49,14 +55,42 @@ Each token is matched as a case-insensitive substring of the event type, so `CAM
|
||||
|
||||
## Heartbeats
|
||||
|
||||
The `org:<org_id>` join reply doubles as a HELLO: it returns the cadence the server expects, so a client library does not hardcode it.
|
||||
The `org:<org_id>` join reply doubles as a HELLO: it returns the cadence the server expects (so a client library does not hardcode it) and the current stream `seq`.
|
||||
|
||||
```json
|
||||
{ "org_id": "...", "role": "owner", "heartbeat_interval_ms": 25000, "server_timeout_ms": 60000 }
|
||||
{ "org_id": "...", "role": "owner", "heartbeat_interval_ms": 25000, "server_timeout_ms": 60000, "seq": 4821, "resume_supported": true }
|
||||
```
|
||||
|
||||
Heartbeats are client-initiated (standard Phoenix): send a `heartbeat` event on the `phoenix` topic every `heartbeat_interval_ms`. If the server receives nothing for `server_timeout_ms` it closes the socket. The reference client also arms a short watchdog after each heartbeat and force-reconnects if the reply does not arrive, which detects a silently dead connection faster than the timeout.
|
||||
|
||||
## Resuming after a disconnect
|
||||
|
||||
Every event carries a monotonic per-organization sequence number in its `seq` field, delivered in order. Track the highest `seq` you have processed; the HELLO also returns the current `seq`.
|
||||
|
||||
To resume after a reconnect, rejoin the `org:<org_id>` channel with a resume token instead of starting fresh:
|
||||
|
||||
```json
|
||||
{ "resume": { "last_seq": 4821 } }
|
||||
```
|
||||
|
||||
The server then either replays what you missed or tells you to resync:
|
||||
|
||||
- **Replay.** The events with `seq` greater than `last_seq` are pushed as normal channel messages (same shape as live, filtered by your permissions and intents exactly like live delivery), followed by a `resumed` marker:
|
||||
|
||||
```json
|
||||
{ "from": 4821, "current_seq": 4895, "replayed": 12 }
|
||||
```
|
||||
|
||||
- **Resync.** If your position is no longer in the buffer (you were disconnected longer than the buffer window) or the token is malformed, the server pushes `resume_failed` and no events:
|
||||
|
||||
```json
|
||||
{ "reason": "buffer_evicted", "current_seq": 5300 }
|
||||
```
|
||||
|
||||
On `resume_failed`, refetch the affected resources over the REST API, then continue live from `current_seq`.
|
||||
|
||||
Resume is **at-least-once**: a replay may re-deliver an event you already handled, so dedupe by `seq`. The buffer holds roughly the most recent events per organization, bounded by both size and time, so resume covers short disconnects (deploys, network blips, tab sleep) but not arbitrarily long ones.
|
||||
|
||||
## Presence
|
||||
|
||||
The org channel carries team presence for the collaboration features in the dashboard (who is online, who is viewing or replying to a record). Standard Phoenix presence events are used:
|
||||
@@ -100,11 +134,14 @@ Because the code is not delivered as a WebSocket close frame today, a client can
|
||||
|
||||
## Delivery guarantees
|
||||
|
||||
The stream is best-effort and at-most-once. There are no sequence numbers, no acknowledgements, and no server-side replay or resume, so events emitted while a client is disconnected are not buffered and are lost. On reconnect, resynchronize by refetching the affected resources over the REST API rather than assuming the stream is gap-free. Do not use the WebSocket as a system of record or an audit log: use it to reduce polling on top of REST, and use [webhooks](/api/endpoints/) where guaranteed delivery matters.
|
||||
Every event has a monotonic per-organization `seq` and is delivered in order. Within the buffer window, a reconnecting client can [resume](#resuming-after-a-disconnect) and replay exactly what it missed; across a resume, delivery is at-least-once (dedupe by `seq`).
|
||||
|
||||
The buffer is not an infinite log. It holds roughly the most recent events per organization, bounded by size and time, so a client disconnected longer than that window gets a `resume_failed` and must do a full resync over the REST API. For delivery that must survive arbitrary downtime, use [webhooks](/api/endpoints/), which are persisted and retried; the WebSocket is the low-latency path, webhooks are the durable one.
|
||||
|
||||
## Good citizenship
|
||||
|
||||
- Reuse one connection per process and multiplex channels over it instead of opening one socket per topic.
|
||||
- Pass `intents` so you only receive (and pay the message-rate budget for) the events you act on.
|
||||
- Reconnect with exponential backoff; the limits above treat reconnect storms the same as connection spam.
|
||||
- Treat events as invalidation signals and refetch via the REST API, rather than assuming every payload carries full state.
|
||||
- After a reconnect, resume with your last `seq` instead of refetching everything; only fall back to a full REST resync on `resume_failed`.
|
||||
- Treat event payloads as invalidation signals (they carry ids, not full state) and refetch the resource over the REST API when you need its current contents.
|
||||
|
||||
@@ -0,0 +1,230 @@
|
||||
{
|
||||
"asyncapi": "3.1.0",
|
||||
"info": {
|
||||
"title": "Warmbly Realtime Gateway",
|
||||
"version": "1.0.0",
|
||||
"description": "Resumable WebSocket gateway for live organization events. The socket speaks the Phoenix channel protocol (serializer 1.0.0). Authenticate with an API key (REALTIME_SUBSCRIBE permission) or a short-lived JWT passed as the `token` query parameter. Events carry a monotonic per-organization `seq`; reconnecting clients resume by replaying the gap. See the human guide at https://docs.warmbly.com/api/realtime/.",
|
||||
"contact": { "name": "Warmbly", "url": "https://docs.warmbly.com" },
|
||||
"license": { "name": "Proprietary", "url": "https://warmbly.com" }
|
||||
},
|
||||
"defaultContentType": "application/json",
|
||||
"servers": {
|
||||
"production": {
|
||||
"host": "realtime.warmbly.com",
|
||||
"pathname": "/socket/websocket",
|
||||
"protocol": "wss",
|
||||
"description": "Production gateway. Connect with ?vsn=1.0.0&token=<TOKEN>.",
|
||||
"security": [{ "$ref": "#/components/securitySchemes/token" }]
|
||||
}
|
||||
},
|
||||
"channels": {
|
||||
"org": {
|
||||
"address": "org:{org_id}",
|
||||
"title": "Organization channel",
|
||||
"description": "Per-organization event stream + team presence. A subscriber receives every event the member/key is permitted to see, optionally narrowed by intents. Messages are Phoenix channel frames [join_ref, ref, topic, event, payload].",
|
||||
"parameters": {
|
||||
"org_id": { "description": "The organization id (UUID) to subscribe to." }
|
||||
},
|
||||
"messages": {
|
||||
"join": { "$ref": "#/components/messages/Join" },
|
||||
"hello": { "$ref": "#/components/messages/Hello" },
|
||||
"event": { "$ref": "#/components/messages/Event" },
|
||||
"resumed": { "$ref": "#/components/messages/Resumed" },
|
||||
"resumeFailed": { "$ref": "#/components/messages/ResumeFailed" },
|
||||
"rateLimited": { "$ref": "#/components/messages/RateLimited" },
|
||||
"presenceState": { "$ref": "#/components/messages/PresenceState" },
|
||||
"presenceDiff": { "$ref": "#/components/messages/PresenceDiff" },
|
||||
"presenceUpdate": { "$ref": "#/components/messages/PresenceUpdate" }
|
||||
}
|
||||
}
|
||||
},
|
||||
"operations": {
|
||||
"joinOrg": {
|
||||
"action": "send",
|
||||
"channel": { "$ref": "#/channels/org" },
|
||||
"summary": "Join the org channel (optionally with intents and a resume token).",
|
||||
"messages": [{ "$ref": "#/channels/org/messages/join" }],
|
||||
"reply": {
|
||||
"channel": { "$ref": "#/channels/org" },
|
||||
"messages": [{ "$ref": "#/channels/org/messages/hello" }]
|
||||
}
|
||||
},
|
||||
"receiveOrgStream": {
|
||||
"action": "receive",
|
||||
"channel": { "$ref": "#/channels/org" },
|
||||
"summary": "Receive live events, resume markers, presence, and rate-limit notices.",
|
||||
"messages": [
|
||||
{ "$ref": "#/channels/org/messages/event" },
|
||||
{ "$ref": "#/channels/org/messages/resumed" },
|
||||
{ "$ref": "#/channels/org/messages/resumeFailed" },
|
||||
{ "$ref": "#/channels/org/messages/rateLimited" },
|
||||
{ "$ref": "#/channels/org/messages/presenceState" },
|
||||
{ "$ref": "#/channels/org/messages/presenceDiff" }
|
||||
]
|
||||
},
|
||||
"updatePresence": {
|
||||
"action": "send",
|
||||
"channel": { "$ref": "#/channels/org" },
|
||||
"summary": "Update your own presence activity (JWT members only).",
|
||||
"messages": [{ "$ref": "#/channels/org/messages/presenceUpdate" }]
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"securitySchemes": {
|
||||
"token": {
|
||||
"type": "httpApiKey",
|
||||
"in": "query",
|
||||
"name": "token",
|
||||
"description": "An API key with the REALTIME_SUBSCRIBE permission, or a short-lived connection JWT, passed as the `token` query parameter on the socket URL."
|
||||
}
|
||||
},
|
||||
"messages": {
|
||||
"Join": {
|
||||
"name": "phx_join",
|
||||
"title": "Join (phx_join)",
|
||||
"summary": "Subscribe to org:{org_id}. Optionally declare intents and a resume token.",
|
||||
"payload": { "$ref": "#/components/schemas/JoinPayload" }
|
||||
},
|
||||
"Hello": {
|
||||
"name": "phx_reply",
|
||||
"title": "HELLO (join reply)",
|
||||
"summary": "Heartbeat cadence and the current stream sequence.",
|
||||
"payload": { "$ref": "#/components/schemas/Hello" }
|
||||
},
|
||||
"Event": {
|
||||
"name": "event",
|
||||
"title": "Domain event",
|
||||
"summary": "A live or replayed event. The frame event name is the event_type (e.g. CAMPAIGN_UPDATED); the payload carries the body plus a monotonic seq.",
|
||||
"payload": { "$ref": "#/components/schemas/Event" }
|
||||
},
|
||||
"Resumed": {
|
||||
"name": "resumed",
|
||||
"title": "Resume complete",
|
||||
"summary": "Sent after a successful resume replay; the events preceded it.",
|
||||
"payload": { "$ref": "#/components/schemas/Resumed" }
|
||||
},
|
||||
"ResumeFailed": {
|
||||
"name": "resume_failed",
|
||||
"title": "Resume failed",
|
||||
"summary": "The buffer no longer covers your position (or the token was malformed). Do a full REST resync, then continue from current_seq.",
|
||||
"payload": { "$ref": "#/components/schemas/ResumeFailed" }
|
||||
},
|
||||
"RateLimited": {
|
||||
"name": "rate_limited",
|
||||
"title": "Outbound rate limited",
|
||||
"payload": { "$ref": "#/components/schemas/RateLimited" }
|
||||
},
|
||||
"PresenceState": {
|
||||
"name": "presence_state",
|
||||
"title": "Presence snapshot",
|
||||
"payload": { "$ref": "#/components/schemas/PresenceState" }
|
||||
},
|
||||
"PresenceDiff": {
|
||||
"name": "presence_diff",
|
||||
"title": "Presence diff",
|
||||
"payload": { "$ref": "#/components/schemas/PresenceDiff" }
|
||||
},
|
||||
"PresenceUpdate": {
|
||||
"name": "presence:update",
|
||||
"title": "Update own presence",
|
||||
"payload": { "$ref": "#/components/schemas/PresenceUpdate" }
|
||||
}
|
||||
},
|
||||
"schemas": {
|
||||
"JoinPayload": {
|
||||
"type": "object",
|
||||
"description": "Payload of the phx_join frame for org:{org_id}.",
|
||||
"properties": {
|
||||
"intents": {
|
||||
"type": "array",
|
||||
"items": { "type": "string" },
|
||||
"description": "Optional event-family tokens (case-insensitive substring of the event type, e.g. CAMPAIGN, EMAIL, AUDIT). Absent/empty = the full permitted stream. Not a security boundary."
|
||||
},
|
||||
"resume": {
|
||||
"type": "object",
|
||||
"description": "Resume token from a reconnecting client.",
|
||||
"properties": {
|
||||
"last_seq": {
|
||||
"type": "integer",
|
||||
"minimum": 0,
|
||||
"description": "The highest seq the client has processed. The server replays events with seq greater than this."
|
||||
}
|
||||
},
|
||||
"required": ["last_seq"]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Hello": {
|
||||
"type": "object",
|
||||
"required": ["org_id", "heartbeat_interval_ms", "server_timeout_ms", "seq"],
|
||||
"properties": {
|
||||
"org_id": { "type": "string", "format": "uuid" },
|
||||
"role": { "type": "string", "description": "The member's role in the org (JWT sockets)." },
|
||||
"heartbeat_interval_ms": { "type": "integer", "description": "Send a heartbeat on the phoenix topic at least this often." },
|
||||
"server_timeout_ms": { "type": "integer", "description": "The server closes the socket after this long with no heartbeat." },
|
||||
"seq": { "type": "integer", "description": "Current stream sequence. A fresh client should start tracking from here." },
|
||||
"resume_supported": { "type": "boolean" }
|
||||
}
|
||||
},
|
||||
"Event": {
|
||||
"type": "object",
|
||||
"required": ["event_type", "seq"],
|
||||
"description": "Invalidation-oriented: the body carries ids, not full resource state. Refetch over REST for current contents. Permission- and intent-filtered identically for live and replayed events.",
|
||||
"properties": {
|
||||
"event_type": { "type": "string", "description": "e.g. AUDIT_CREATED, CAMPAIGN_UPDATED, EMAIL_SENT, EMAIL_RECEIVED. Also the frame event name." },
|
||||
"seq": { "type": "integer", "description": "Monotonic per-organization sequence. Track the highest seen; dedupe replays by it." },
|
||||
"org_id": { "type": "string", "format": "uuid" },
|
||||
"user_id": { "type": "string", "format": "uuid" },
|
||||
"campaign_id": { "type": "string", "format": "uuid" },
|
||||
"email_account_id": { "type": "string", "format": "uuid" }
|
||||
},
|
||||
"additionalProperties": true
|
||||
},
|
||||
"Resumed": {
|
||||
"type": "object",
|
||||
"required": ["from", "current_seq", "replayed"],
|
||||
"properties": {
|
||||
"from": { "type": "integer", "description": "The last_seq the client resumed from." },
|
||||
"current_seq": { "type": "integer", "description": "The stream sequence after replay; the client is now caught up to here." },
|
||||
"replayed": { "type": "integer", "description": "Number of events delivered during replay (after permission + intent filtering)." }
|
||||
}
|
||||
},
|
||||
"ResumeFailed": {
|
||||
"type": "object",
|
||||
"required": ["reason", "current_seq"],
|
||||
"properties": {
|
||||
"reason": { "type": "string", "enum": ["buffer_evicted", "invalid_resume"] },
|
||||
"current_seq": { "type": "integer", "description": "Resync over REST, then continue live from here." }
|
||||
}
|
||||
},
|
||||
"RateLimited": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"category": { "type": "string", "example": "ws_message" },
|
||||
"retry_after_ms": { "type": "integer" }
|
||||
}
|
||||
},
|
||||
"PresenceState": {
|
||||
"type": "object",
|
||||
"description": "Phoenix presence snapshot: a map of user_id -> { metas: [...] }.",
|
||||
"additionalProperties": true
|
||||
},
|
||||
"PresenceDiff": {
|
||||
"type": "object",
|
||||
"description": "Phoenix presence diff with joins and leaves.",
|
||||
"properties": {
|
||||
"joins": { "type": "object", "additionalProperties": true },
|
||||
"leaves": { "type": "object", "additionalProperties": true }
|
||||
}
|
||||
},
|
||||
"PresenceUpdate": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"page": { "type": "string" },
|
||||
"resource": { "type": "string", "description": "e.g. thread:<id>, campaign:<id>, contact:<id>." },
|
||||
"action": { "type": "string", "enum": ["viewing", "editing", "replying", "idle"] }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user