Livestreaming & Voice
Two real-time media systems. Livestreams come in two kinds that share one LiveStream table (kind: "mux" | "livekit"), one viewer screen, and one realtime room for chat, hearts and viewer count.
| OBS livestream | Phone livestream ("Go live") | Voice/video rooms | |
|---|---|---|---|
| Provider | Mux | LiveKit (self-hosted) | LiveKit (self-hosted) |
| Topology | One-to-many broadcast | One-to-many broadcast | Many-to-many conference |
| Protocol | RTMP in, HLS out | WebRTC in and out | WebRTC |
| Latency | Seconds | Sub-second | Sub-second |
LiveStream rows | One per user, reused (kind: "mux") | One per broadcast (kind: "livekit") | — |
| Used for | Creators streaming from OBS | Instagram-Live-style streams from the app camera | Channel rooms and 1-on-1 sessions |
Livestreaming from OBS (Mux)
Credentials are provisioned early
Every user gets a LiveStream record with a Mux stream key — it is created during registration (registerUser calls createLiveStream). Streaming is a matter of retrieving credentials rather than provisioning them at broadcast time:
GET /api/v1/livestreams/credentials → { streamKey, muxPlaybackId, status }
POST /api/v1/livestreams/credentials/regenerate → new stream keyGoing live
The User.isLive and User.activeLiveStreamId denormalised columns are what let the feed and discovery surfaces show a live badge without joining to LiveStream.
Webhook events handled
POST /api/v1/webhooks/mux (apps/api/src/router/webhookRoutes.ts):
| Event | Effect |
|---|---|
video.asset.ready | Media marked ready, dimensions and duration stored, post marked ready, realtime update emitted |
video.live_stream.active | Stream active, User.isLive = true, live:status sent to followers, followers notified (push + in-app, via the live-notify queue) |
video.live_stream.idle / .disconnected | Stream idle; User.isLive cleared only if activeLiveStreamId still points at this Mux row (the user may be live from their phone by now) |
.connected, .recording, .updated, .enabled, .disabled, .warning | Acknowledged, no action |
The Mux webhook is unauthenticated
MUX_WEBHOOK_SECRET is present in the environment and the deployment manifests, but the handler does not verify the signature. Anyone who can reach api.playpalz.gg/api/v1/webhooks/mux can post a crafted video.live_stream.active and mark an arbitrary creator live, or mark arbitrary media ready.
Fix by verifying Mux-Signature with mux.webhooks.verifySignature(rawBody, headers, secret) before the switch — which also requires capturing the raw body for that route, since express.json() has already consumed it.
Livestreaming from the phone (LiveKit)
Home + → Go live opens a full-screen camera preview (app/(app)/(modals)/live/go-live.tsx), then a 3-2-1 countdown, then the live view. The host publishes camera + mic to a LiveKit room named live_<livestreamId>; viewers join the same room hidden and subscribe-only.
| Status | Meaning |
|---|---|
starting | Created on "Go Live"; publishing during the countdown, visible to nobody |
live | On air; followers notified |
ended | Host ended it, or the host dropped (webhook / sweeper) |
cancelled | Aborted during the countdown |
abandoned | Never started (sweeper, after 3 minutes) |
Follower notifications (queue)
"X started a live video" goes out through the BullMQ queue live-notify (@playpals/queue), never on the request path — for phone lives (startPhoneLive) and for OBS streams (the Mux video.live_stream.active webhook). notifyFollowersLive checks a 15-minute per-host cooldown (which also absorbs OBS reconnects) and enqueues one job per broadcast session: a phone live's session is its row id; an OBS session is <muxRowId>-<activation time>, because the Mux row is reused and dedupe keyed on it would swallow the next broadcast's notifications. the worker (apps/api/src/jobs/liveNotify.ts) does the rest:
| Job | Does |
|---|---|
fanout | Reads one page of 1,000 followers, enqueues deliver batches of 50, and enqueues the next page as a new fanout job |
deliver | Calls notify() for each follower in the batch (which applies blocks and the liveStreams preference) |
Both jobs stop if the stream is no longer live, so a queue backlog never sends notifications for a stream that has already ended. Retries are safe: batch job ids are derived from their contents, and each recipient is claimed in the Redis set live:notify:done:<id> before being notified (released again if the push fails), so a retried batch skips anyone already reached.
The worker runs inside every API replica by default (BullMQ spreads jobs across them). To move it to a dedicated process, set LIVE_NOTIFY_WORKER=false on the API and call startLiveNotifyWorker() from the new process's entrypoint.
Ending is idempotent (conditional updateMany on status in (starting, live)), because three things race to end a stream: the host's End button, the LiveKit webhook and the sweeper.
POST /api/v1/livestreams/live → { livestream, token, livekitHost }
POST /api/v1/livestreams/:id/start → owner only; starting → live
POST /api/v1/livestreams/:id/end → owner only; → LiveSummary
GET /api/v1/livestreams/:id/token → viewer token (410 once ended, 403 if blocked)
GET /api/v1/livestreams/following-live → followed users live now (Home "Live now" rail)Ending a stream nobody ended
- LiveKit webhook —
POST /api/v1/webhooks/livekit, verified withWebhookReceiverover the rawapplication/webhook+jsonbody. Onlylive_*rooms are handled: the host'sparticipant_leftorroom_finishedends the stream. - Sweeper —
apps/api/src/jobs/liveSweeper.ts, every 60s on every replica behind a Redis lock: alivestream whose host is missing from the LiveKit room on two sweeps in a row is ended; astartingstream older than 3 minutes isabandoned. This is also what cleans up in local dev, where LiveKit can't reach the webhook without a tunnel. - Host backgrounding — the app mutes the camera (viewers see "will be right back"). Android ends the stream after 60s in the background; iOS suspends the app, so the webhook/sweeper ends it.
Chat, hearts and viewer count (both kinds)
Realtime lives in the esu socket room livestream:<id>, joined by the host and every viewer.
| Event | Direction | Notes |
|---|---|---|
livestream:join / :leave | client → esu | Authorised by GET /livestream/:id/authz (host always; others while on air and not blocked). Ack carries the current viewer count. |
livestream:heart { count } | client → esu | Batched taps (≤10 per batch, ≤1 batch / 250ms per socket). Relayed to the room and INCRBY in Redis. |
livestream:message:new | esu → room | Emitted by the API after a message is stored, minus users blocked with the sender |
livestream:viewers { count } | esu → room | Unique users in the room minus the host, debounced 1s, via fetchSockets() (all replicas) |
livestream:hearts { count } | esu → room | Anonymous |
livestream:started / :ended | esu → room | |
live:status | esu → followers' user rooms | Refreshes the Home "Live now" rail |
esu writes the current and peak viewer counts and the heart total to Redis (livestreamRedisKeys in @playpals/types); the API reads them into peakViewers / heartCount when the stream ends.
Chat is stored, not ephemeral — LiveStreamMessage rows:
GET /api/v1/livestreams/:id/messages → latest 50, oldest first, blocked authors excluded
POST /api/v1/livestreams/:id/messages → { body (1–300), clientId? }; only while on air; 1 per 700ms per userThe sender's clientId is echoed on the response and the socket event so the optimistic row is replaced in place (same pattern as DMs and room chat).
Live tips
Viewers tip the host with POST /tips { targetType: "livestream", targetId: <stream id> } while the stream is on air (live, starting or active). The tip and a LiveStreamMessage with kind: "tip" and tipId are written in one transaction; the line is then broadcast through the normal livestream:message:new path, so block exclusions and history work as for chat. The LiveMessage carries kind: "tip" and tip: { id, coins }; body is the tip's message (may be empty). Clients show it as a gold row and a "Riya tipped 500" banner.
The host can hide a tip's message (POST /tips/:id/hide-message). The API re-broadcasts the line with body: "" and tip.messageHidden: true, and clients replace the row they already show. Anyone watching except the sender can report it. The viewer screen is a full-screen modal, so it mounts its own tip sheet. See Monetization for the coin rules.
Voice rooms (LiveKit)
Why self-hosted
LiveKit runs on a dedicated DigitalOcean droplet rather than in the Kubernetes cluster. WebRTC needs a wide UDP port range and direct host networking, which is awkward to express through an ingress controller. The droplet is provisioned by Ansible — see Ansible & the LiveKit Node for provisioning, upgrades, and credentials.
Locally, LiveKit runs in Docker Compose with the dev key pair devkey / secret:
docker compose up -d livekit| Port | Purpose |
|---|---|
| 7880 | HTTP API |
| 7881 | RTC over TCP |
| 7882/udp | RTC over UDP |
Joining a room
The API is the only issuer of LiveKit tokens, so room access rules live in one place. Tokens are short-lived (1 hour) and scoped to a single room with explicit canPublish / canSubscribe grants. Empty rooms are torn down after emptyTimeout: 300 seconds.
Room access rules
apps/api/src/lib/roomAccess.ts — canAccessRoom(userId, roomId):
| Visibility | Who may enter |
|---|---|
public | Anyone |
members (default) | The channel owner, or a ChannelMember |
vip | The channel owner, or a channel member with an active CreatorSubscription to the channel owner |
private | Only the two parties of the Session attached to the room (the PlayPal and the purchaser) |
vip is the paywall: membership alone is not enough, the fan must hold a live subscription whose expiresAt is null or in the future.
private is how 1-on-1 booked sessions work — the room is bound to a Session row, and only the playpalId and purchaserId can get a token.
The same rule gates a room's text chat: GET /rooms/:roomId, GET /rooms/:roomId/messages and POST /rooms/:roomId/messages answer 403 when canAccessRoom is false, so a channel member without a subscription can't read or post in a vip room.
Client side
useLiveKitCall (apps/mobile/hooks/useLiveKitCall.ts) wraps @livekit/react-native. The native WebRTC module means voice rooms cannot be tested in Expo Go — you need a development build.
Configuration
| Variable | Service | Notes |
|---|---|---|
MUX_ACCESS_TOKEN, MUX_SECRET_KEY | api, ogun | Mux API credentials |
MUX_WEBHOOK_SECRET | api | Present but unused — see the warning above |
LIVEKIT_HOST | api | HTTP API base, e.g. https://ws.playpalz.gg |
LIVEKIT_WS_URL | api | Client WebSocket URL, e.g. wss://ws.playpalz.gg |
LIVEKIT_API_KEY, LIVEKIT_API_SECRET | api | Token signing and LiveKit webhook verification |
LIVE_NOTIFY_WORKER | api | false stops this replica consuming the live-notify queue (default: consumes) |
The LiveKit server must be configured to send webhooks to https://<api>/api/v1/webhooks/livekit (the webhook.urls block in livekit.yaml, signed with the same API key). Without it, the sweeper still ends abandoned phone lives, just more slowly.
In production these come from the mux-secrets and livekit-secrets sealed secrets. See Secrets Management.