Realtime WebSocket Channels

Category: Backend & APIs

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Add realtime to your app the way you'd use Ably or Pusher — pub/sub channels, presence, history-rewind, retained state — but with two things they can't do: the message can run your pipeline, and auth can be Air Pipe's, yours, or fully open. No message broker, Redis or NATS to run; cross-node fan-out rides your private node mesh.

Great for live dashboards, chat, presence/"who's online", collaborative apps, notifications, order/match/status feeds, and public live pages.

How Air Pipe compares (WebSocket realtime)

Capability	Ably	Pusher	Air Pipe
Pub/sub channels + wildcards	✅	✅	✅
Presence (who's here, enter/leave)	✅	✅	✅
History / message rewind	✅	💲 add-on	✅
Retained "last value" on subscribe	✅	✅	✅
Public channels (no auth)	✅	✅	✅
Token / JWT end-user auth	✅	✅	✅
Issues the token for you (no IdP needed)	➖	➖	🏆 `add_jwt`
Run your logic on the message (validate/enrich/store)	❌	❌	🏆 built-in
Channel ACL = your own pipeline + claims	partial	partial	🏆
One vendor: transport + auth + compute + billing	❌	❌	🏆
Bring-your-own-infra / no external broker	❌	❌	🏆

Parity on every fundamental; ahead wherever "realtime + your backend logic + your auth" matters.

What's included

File	Purpose
`channels.yml`	Five interfaces: `public-feed` (open), `live` (authenticated pub/sub + presence + history + retain), `authorize-channel` (JWT-claim ACL), `login` (Air Pipe mints the token), `publish` (backend broadcast).

Endpoints

Interface	Connect / call	Auth
`public-feed`	`wss://<host>/ws/<ORG_UUID>/production/public-feed`	none (public)
`live`	`wss://<host>/ws/<ORG_UUID>/production/live?token=<JWT>`	end-user JWT
`login`	`POST https://<host>/login`	none → returns a JWT
`publish`	`POST https://<host>/publish`	your backend (protect it)

Pick your auth — three ways, one pack

Mode	How	Use when
Open	`public: true` on the interface — connect with no credential	Non-sensitive public feeds (scores, status, public presence)
Air Pipe issues it	Users `POST /login`; `add_jwt` mints an HS256 token Air Pipe also verifies	You have no identity provider — "just do auth for me"
Your IdP	End-users bring their JWT; `authorize-channel` verifies it — HS256 shared secret, or swap to `jwks_url` for RS256/ES256	You already have Auth0/Cognito/your own login

The org API key is a backend-only credential — never ship it to a browser. Browsers use ?token=<jwt> (a browser can't set WebSocket headers); native clients can use the airpipe-jwt header.

Setup

Managed variable	Value	Needed for
`JWT_SECRET`	any strong secret	`login` (minting) + `authorize-channel` (verifying) — the Air-Pipe-issued mode
`JWKS_URL`	your IdP's JWKS endpoint	only if you switch `authorize-channel` to the JWKS form

The open public-feed needs no variables — deploy and connect.

Control frames (client → server)

Frame	Effect
`{"ap_subscribe": "room:42"}` / `{"ap_subscribe": ["a","b"]}`	Join channel(s); reply `{"ap_subscribed":[…]}`. Wildcards: `room:+`, `room:#`
`{"ap_subscribe": ["room:42"], "history": 20}`	Join and replay the last 20 messages (each flagged `"ap_history": true`)
`{"ap_publish": {"channels":["room:42"], "data":{…}, "retain": true}}`	Publish; `retain` stores it as the last value for new subscribers
`{"ap_presence_enter": {"channel":"room:42","id":"alice","state":{…}}}`	Join presence; subscribers get a `$presence` enter event
`{"ap_presence_leave": {"channel":"room:42"}}`	Leave presence (also automatic on disconnect)
`{"ap_presence_get": {"channel":"room:42"}}`	`{"ap_presence":{"channel":"room:42","members":[…]}}` (cluster-wide)
`{"ap_unsubscribe": "room:42"}`	Leave channel(s)
any other JSON frame	Runs the interface (request/response) on the same socket

Pushes arrive as {"ap_channel":"room:42","data":{…}}.

Quick start

1) Open public feed — no auth (browser):

const ws = new WebSocket("wss://your-airpipe-host/ws/<ORG_UUID>/production/public-feed");
ws.onopen = () => ws.send(JSON.stringify({ ap_subscribe: "scores:live" }));
ws.onmessage = (e) => console.log(JSON.parse(e.data)); // {ap_channel:"scores:live", data:{…}}

2) Authenticated — Air Pipe issues the token:

// a) log in → get a token (no IdP required)
const { data } = await fetch("https://your-airpipe-host/login", {
  method: "POST", headers: { "content-type": "application/json" },
  body: JSON.stringify({ username: "alice", password: "demo-password" }),
}).then(r => r.json());
const token = data.Mint.data.token;

// b) open the authenticated socket with it (browsers pass ?token=)
const ws = new WebSocket(`wss://your-airpipe-host/ws/<ORG_UUID>/production/live?token=${token}`);
ws.onopen = () => {
  ws.send(JSON.stringify({ ap_subscribe: ["room:42"], history: 20 }));         // subscribe + rewind
  ws.send(JSON.stringify({ ap_presence_enter: { channel: "room:42", id: "alice", state: { typing: false } } }));
};
ws.onmessage = (e) => console.log(JSON.parse(e.data));

3) Publish from your backend (broadcast, retained):

curl -sX POST https://your-airpipe-host/publish -H 'content-type: application/json' -d '{
  "channels": ["room:42"],
  "retain": true,
  "data": { "message": "hello everyone" }
}'
# → every socket subscribed to room:42 receives {"ap_channel":"room:42","data":{…}}

Customisation

Real login: replace the DEMO check in login's Validate with a database lookup + bcrypt compare (see rest-api-starter / mcp-postgres-starter), then mint claims off the real user.
Enterprise IdP: switch authorize-channel's assert to is_valid_jwt: { jwks_url: "a|ap_var::JWKS_URL|" } (RS256/ES256) — no other change.
Per-user channels: uncomment the Bind action in authorize-channel to allow only user:<sub>.
Publish from any pipeline: add a ws_publish action to your HTTP/MQTT/scheduled interfaces so, e.g., a DB insert also broadcasts the new row.
Protect /publish: add an is_valid_jwt / validate_key assert so only your backend can broadcast.

Notes

Delivery is ordered per-channel; retain recovers the last value and history:N replays missed messages — so a (re)connecting client isn't left blank.
Cross-node fan-out uses your private node mesh — no Redis/NATS/broker to operate.
Safe on shared infra: per-org limits bound publish rate, subscriptions, presence, retained and topic count, so one org (or an open public-feed) can't exhaust a node. Higher plans lift the ceilings.
History depth is a plan entitlement (max_ws_history); the open/free default may be 0 (off) — raise it on paid tiers.
Each delivered message and each request/response frame counts as one request against your plan.
Only production and staging environments are accepted in the URL.

Configuration

channels.yml

name: RealtimeWebSocket
description: >
  Realtime WebSocket channels — the drop-in alternative to Ably / Pusher, with the same pub/sub,
  presence, history-rewind and retained-state, PLUS the message can run your pipeline and the auth
  can be Air Pipe's, yours, or fully open. No message broker, Redis or NATS to run.

docs: true

# ── How it fits together ────────────────────────────────────────────────────────
# Clients open ONE WebSocket and speak simple JSON control frames:
#
#   {"ap_subscribe": "room:42"}                 → join a channel (or ["a","b"], + wildcards a/#)
#   {"ap_subscribe": ["room:42"], "history": 20}→ join AND replay the last 20 messages (rewind)
#   {"ap_publish": {"channels":["room:42"], "data":{...}, "retain": true}}  → publish (retain=last value)
#   {"ap_presence_enter": {"channel":"room:42","id":"alice","state":{"typing":false}}}
#   {"ap_presence_get":   {"channel":"room:42"}}   → {"ap_presence":{"channel","members":[...]}}
#   {"ap_unsubscribe": "room:42"}
#
# Pushes arrive as {"ap_channel":"room:42","data":{...}}. Presence enter/leave events arrive on the
# channel as {"ap_channel":"room:42","data":{"$presence":{"action":"enter","id":"alice","state":{}}}}.
#
# THREE ways to authenticate an end-user — pick per interface (this pack shows all three):
#   1. OPEN         — `public: true`, no credential at all (non-sensitive public feeds).
#   2. AIR PIPE     — users log in through the `login` interface; Air Pipe mints + verifies the JWT.
#   3. YOUR IdP     — end-users bring their own JWT; verified by shared-secret (HS256) or JWKS.

interfaces:

  # ── 1) OPEN public channel (Ably/Pusher "public channel" parity) ────────────────
  # `public: true` lets a browser connect with NO credential — for non-sensitive feeds
  # (live scores, status pages, public activity). The per-org abuse floors (publish rate,
  # subscriptions, topics, retained) still bound it, so open access is safe on shared infra.
  public-feed:
    ws: "on"                       # realtime — no `output` needed
    public: true
    # output: http                 # optional: also serve this interface over HTTP (dual interface)
    summary: Open public realtime feed (no auth)
    description: Connect with no credential and subscribe to public channels.
    tags: [websocket, realtime, public]
    actions:
      - name: Ping
        input: a|body|

  # ── 2) AUTHENTICATED channels (presence + history + retained) ───────────────────
  # End-users connect with a JWT (minted by `login` below, or your own IdP's token) passed as
  # `?token=<jwt>` (browsers) or the `airpipe-jwt` header (native clients). `subscribe_authorizer`
  # gates which channels each token may join. Presence, history-rewind and retained all work here
  # with no extra config — they are engine control frames (see the header comment).
  live:
    ws: "on"                       # realtime — no `output` needed
    subscribe_authorizer: authorize-channel
    # output: http                 # optional: also serve this interface over HTTP (dual interface)
    summary: Authenticated realtime channels (presence + history + retain)
    description: Subscribe, publish, presence and history over one socket, gated per JWT claim.
    tags: [websocket, realtime, presence, history, pubsub]
    actions:
      - name: Ping
        input: a|body|

  # Channel ACL — runs on every `ap_subscribe`/`ap_presence_enter` with {"channel":"<name>"} as
  # a|body| and the socket's token as a|headers::airpipe-jwt|. Allowed iff it returns 2xx.
  # Here: require a valid token (HS256 shared secret). For an enterprise IdP, swap the assert for
  # the JWKS form: `is_valid_jwt: { jwks_url: "a|ap_var::JWKS_URL|" }` (RS256/ES256).
  authorize-channel:               # internal authorizer — run by the socket, not an HTTP route
    actions:
      - name: Verify
        input: a|headers|
        assert:
          http_code_on_error: 403
          error_message: "not authorized for this channel"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
      # Optional: bind a channel to a claim so a user only joins their own room. Uncomment +
      # adapt — e.g. only allow `user:<sub>` (Verify exposes the decoded claims).
      # - name: Bind
      #   run_when_succeeded:
      #     actions: [Verify]
      #     http_code_on_error: 403
      #   input: a|body|
      #   assert:
      #     error_message: "channel not allowed for this user"
      #     tests:
      #       - value: channel
      #         is_equal_to: "user:a|Verify::sub|"

  # ── 3) LOGIN — Air Pipe issues the realtime token ("do auth for me") ────────────
  # A customer with no IdP can log their users in HERE: validate the credential, then Air Pipe
  # mints an HS256 JWT with `add_jwt`. The browser uses that token on the `live` socket, and
  # `authorize-channel` verifies it with the SAME secret — a closed Air-Pipe-only loop.
  # DEMO validates presence of fields only; replace with a real check (a database: lookup +
  # bcrypt compare — see the `rest-api-starter` / `mcp-postgres-starter` packs).
  login:
    output: http
    method: POST
    summary: Issue an end-user realtime token (Air Pipe signs it)
    description: Exchange a login for a short-lived JWT to open the realtime socket with.
    tags: [auth, websocket, jwt]
    request_example:
      username: alice
      password: "demo-password"
    actions:
      - name: Validate
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "username and password required"
          tests:
            - value: username
              is_not_null: true
              is_not_empty: true
            - value: password
              is_not_null: true
              is_not_empty: true
      - name: Mint
        run_when_succeeded:
          actions: [Validate]
          http_code_on_error: 401
        input: a|Validate|
        post_transforms:
          - add_jwt:
              key: token
              secret: a|ap_var::JWT_SECRET|
              exp: 24h
              data: [username]

  # ── Backend publish surface ─────────────────────────────────────────────────────
  # Your trusted backend POSTs here to fan a payload out to every subscriber (all nodes).
  # `retain: true` stores it as the channel's last value so late subscribers get it immediately.
  # PROTECT this route before shipping (add an is_valid_jwt / validate_key assert) so only your
  # backend can broadcast. Any pipeline (HTTP/MQTT/scheduled) can also push via a `ws_publish` action.
  publish:
    output: http
    method: POST
    summary: Publish to realtime channels (backend)
    description: Fan a payload out to all subscribers of the given channel(s), optionally retained.
    tags: [websocket, realtime, push]
    request_example:
      channels: ["room:42"]
      retain: true
      data:
        message: "hello everyone"
    actions:
      - name: Broadcast
        ws_publish:
          channels: a|body::channels|
          data: a|body::data|
          retain: a|body::retain|

Realtime WebSocket Channels

How Air Pipe compares (WebSocket realtime)​

What's included​

Endpoints​

Pick your auth — three ways, one pack​

Setup​

Control frames (client → server)​

Quick start​

Customisation​

Notes​

Configuration​

channels.yml​