@codefusion-cc/realtime

Accounts, devices and live channels for apps on Cloudflare Workers and Durable Objects. A device signs in once with an identity provider's ID token bound to its own non-extractable key; from then on the Worker trusts that key, not the token. Each account gets a gateway (a Durable Object) holding its sessions, devices and an event stream it pushes down one control socket per tab. Apps add their own commands to the gateway and their own Durable Objects beside it.

npm install @codefusion-cc/realtime

| Entry | Runs in | What it has | | --- | --- | --- | | @codefusion-cc/realtime | Worker | session and control routes, the gateway, the lobby channel, linked devices, identity certificates, ID token checks | | @codefusion-cc/realtime/browser | page | device identity, the control client, encrypted Durable Object channels, account keys and linked devices, signed messages, DM rings, the outbox | | @codefusion-cc/realtime/react | page | hooks for channels, conversations, history and the outbox | | @codefusion-cc/realtime/protocol | both | the control socket's wire contract and limits, for apps' own commands | | @codefusion-cc/realtime/preview-login | Worker | test accounts for preview deployments (never imported by production code) | | @codefusion-cc/realtime/testing | tests | a device that signs requests, a test sign-in provider, a control socket to a gateway, in-memory gateway storage |

The app's namespace

Every key and route the package derives is bound to the app: its id (APP_ID in the Worker, VITE_APP_ID in the page) names the gateways (<app>:<uid>), the tokens, and the device-proof headers (x-<app>-device-key, …), and the page passes it as namespace wherever a secret becomes a key (deriveChannelCapability, sealData, openDurableChannel, deviceKeyStore). Two apps on one account never share a key or an object. The id must never change: a new one makes every stored history, ring and channel unreadable, and signs every device in anew.

In the Worker

import { defineUserGateway, defineEphemeralChannel, handleRealtimeRoute, createIdentityRoutes } from '@codefusion-cc/realtime'

const config = {
  app: 'example',                                       // the Worker's APP_ID must be the same
  allowedOrigin: origin => origin === 'https://example.app',
  requirePublicUserId: true,
  identityProvider: (id, env) => id === 'google' ? googleVerifier(env) : null,
}

export const UserGateway = defineUserGateway({ commands, handlers, schema })   // the app's own commands
export const LobbyChannel = defineEphemeralChannel({ allowedEvents: ['ring'], audience, lastSeen: true })

export default {
  async fetch(request, env) {
    return await handleRealtimeRoute(request, env, config) ?? env.ASSETS.fetch(request)
  },
}

• The gateway: what an app passes defineUserGateway (schema, commands and handlers, RPC methods, alarm work) can come in parts, an app's own and those packages make, joined with combineGatewayApps(own, part, …) (@codefusion-cc/realtime-media's mediaGateway() is one). Handlers and RPC methods get the gateway's GatewayDeps: its SQL, identity(), activeSession({ sid, dk, uid }) and inChannel({ binding, objectId, dk, sid }). Two parts naming the same method throws.
• Bindings: USER_GATEWAYS (the gateway class), and the rate limiting bindings AUTH_RATE_LIMITER (per device, LIMITS.authRequestsPerMinute) and AUTH_IP_RATE_LIMITER (per network, …PerIp: an IPv4 address, or an IPv6 address's /64), both required by enroll and session. Add SESSION_ACCOUNT_RATE_LIMITER (60 a minute, LIMITS.sessionRequestsPerMinutePerAccount) so an account's verified session requests count on their own: without it they count on AUTH_RATE_LIMITER, and an account with four or more tabs or devices is refused for a minute when they all reconnect after a deploy.
• Rate limits: every route that counts requests treats its binding the same way. The sign-in routes' two, the identity routes' IDENTITY_RATE_LIMITER (or what their limiter(env) returns; certificates per account) and the channel files' uploads binding, once a route names one (uploads per account), are required: without it, the route throws an Error naming it on its first request, since that is a deployment mistake and not a request's. Only success: true lets a request through; anything else is 429 with Retry-After: 60 (a binding's longest period). A binding that fails at run time refuses the request with 503 and Retry-After: 30, and logs an error, rather than leaving the route uncapped while it fails. Tests that call the routes with their own limits can pass testRateLimiter from @codefusion-cc/workers-http/testing.
• Secrets: NETWORK_SESSION_SECRET (signs capabilities and cookies; current:previous while it rotates) and OPAQUE_USER_ID_SECRET (derives account ids; never change it).
• Sign-in: config.identityProvider(providerId, env) returns the verifier for a provider the app offers (@codefusion-cc/google-sign-in makes them for Google and any OIDC issuer), or null. Its verify(token, nonce, now) must check the token is bound to the device key (nonce). A keys-unavailable problem answers 503 with Retry-After, so a provider outage never signs anyone out; any other problem is 401.
• Routes: POST /api/network/enroll (ID token and device proof → capability), POST /api/network/session (capability and device proof → network cookie), GET /api/realtime/control (the control socket), POST /api/realtime/logout. An overloaded gateway answers 503 with Retry-After, never a bare 500.
• An app's own WebSocket routes call prepareAuthenticatedRealtimeUpgrade, which checks the cookie and stamps x-realtime-uid, -dk, -sid, -user and -member onto the request its object receives (inbound copies are removed).
• Member ids: when the provider vouches for an email (email with email_verified: true in the verifier's claims; Microsoft's xms_edov is mapped to it there), the session carries deriveMemberId(secret, app, email), the same for every provider and account with that address. requireMemberId refuses sign-ins without one.

Channels only some may join

defineAuthorizedChannel({ binding, allowedEvents, persistedEvents, stateEvents }) is a Durable Object for a workspace or a conversation. The app proves who belongs (a list its owner signed, a friendship) and installs it with authorize({ version, owner, fingerprint, members, history }); the channel then admits only listed member ids (or public user ids, admitBy: 'user') with live sessions, closes the sockets of members a newer version drops, and relays { type: 'event', event, messageId, data } to everyone, one account (target) or one connection (to, for WebRTC signalling). Data must be an encrypted envelope unless validData says otherwise. With history: true or { until } (a paid plan, to its end) persisted events are kept before they are relayed and replayed to joiners, and state events keep their latest revision; with it off, nothing is stored. An authority without history leaves the setting as it is, so members who do not decide it can present the list too. The gateway lists the binding in its channelBindings.

One account holds at most maxConnectionsPerAccount sockets here (8), its tabs and devices together; one more is refused with 429 rather than evicting the oldest. A joiner's snapshot is at most maxSnapshotBytes (4 MiB): the state events in full, then the newest kept events that fit, and truncated: true when older ones were left out.

A mailbox: kept until delivered

retention: { mode: 'delivered', maxAgeMs, maxBytes } keeps persisted events only until every device of every listed member has them, whatever history says: history lives on the devices, and the server holds each event (ciphertext) only while some device still needs it. Every kept event carries its seq, in the relay, in the sender's ack and in replays. A device confirms with { type: 'delivered', seq } once the events up to seq are in its own store, and the channel deletes everything up to the lowest confirmed one. With openDurableChannel the page confirms each message (room.confirmDelivered(meta.seq)) once it stored it; the client tells the channel, at most every 200 ms, how far every message it heard is confirmed, never past one still waiting for its action, being opened or not yet paged in. Cursors only move forward, and the sender's own device counts as having what it sent when it had everything before it.

• Who is waited on: every device that joined, and every listed member with no device yet, from when they were listed. A member's first device starts there. With backlogForNewMembers: true it gets what is still kept from before; that is off by default, since a new member could read what was said before they belonged.
• Later devices: a later device of a member starts at the tail. Older messages reach it from the member's other devices.
• Joining: the snapshot carries the device's cursor and the events after it, oldest first, within maxSnapshotBytes. With more: true, the page asks { type: 'history', after } for each next page (openDurableChannel pages by itself).
• Limits:
  • Nothing waits forever. An event older than maxAgeMs (30 days) goes undelivered, and a device or member unseen that long is no longer waited on.
  • Past maxBytes (32 MiB) the oldest events go first.
  • A device whose cursor fell behind what was kept gets gap: { after, before } (onGap).
• Membership: a member removed from the list stops holding deletion, and so does a device its account signs out.
• Duplicates: a send retried after its event was delivered and deleted is kept and relayed again under a new seq, and a message not yet confirmed comes again after a reconnect, so pages dedupe by messageId.
• When it stops working: a channel whose oldest event outlived maxAgeMs by a day logs realtime retention: … as an error, which CodeFusion Console reports as a failure.

Files kept with history

With cloudFiles: { bucket, maxBytes, maxFileBytes } a channel keeps its members' files in R2 while history is on. The page seals each file before it leaves the device (cloudFileKeys(namespace, channelSecret): an id the server cannot link to the content, and an AES-GCM key it never has) and moves the sealed bytes with putCloudFile and getCloudFile. The Worker answers PUT and GET at <prefix><routeId>/<fileId> with handleChannelFile; put the prefix under /api/realtime/ so the session cookie reaches it. The route's uploads names a rate limiting binding that counts each account's uploads (429 rate-limited past it; 503 unavailable when it fails); a route that names one throws without it. Only a session connected to the channel right now, as a listed member, may read; storing also needs history on and room in the quota. A channel takes one upload of a file at a time: another one of the same id meanwhile answers 409 busy with Retry-After (retry, and it answers stored: false once the first is in), so two uploads never share or overwrite each other's room or bytes. An upload holds its room for 15 minutes; one slower answers 408 expired, and the channel's alarm deletes the bytes of uploads that never said how they ended. When history turns off or runs out, every file goes with it. open rejects a file that does not open as sealed with a mismatch CloudFileError; a failure of the browser's own WebCrypto rejects as itself, so it never reads as a corrupt file.

Adding to a channel

extend: host => extension adds what an app's channel needs beyond events: a call (channelCalls from @codefusion-cc/realtime-media), or anything else with its own frames. The extension gets the channel's host (its sockets and members, fields kept on a socket, send, broadcast, scheduleAlarm) and may add tables (schema), take frames whose type is not event (frame), add fields to a joiner's snapshot (snapshot; when it throws, the joiner is turned away with 503), hear who joined (joined, in the background) and left (left), and add deadlines to the channel's alarm (deadlines, alarm).

Test accounts on previews

import { createPreviewLogin } from '@codefusion-cc/realtime/preview-login'

const previewLogin = createPreviewLogin({ app: 'example', allowedOrigin })
// In the Worker's fetch, before the app's routes:
const preview = await previewLogin.handle(request, env)
if (preview) return preview
// In the realtime route config:
identityProvider: (id, env) => id === 'test' ? previewLogin.identityProvider(env) : …

The route at /api/test-login is a 404 unless the preview's config sets TEST_LOGIN = "enabled" and a TEST_LOGIN_PASSPHRASE secret of at least 16 characters. It takes an email, a name and the passphrase from the page (requestPreviewLoginToken in the browser entry), counts each try per network on AUTH_IP_RATE_LIMITER, and mints an ID token bound to the device key, which signs in through the ordinary routes as provider test. The issuer is <app>-test; the same email is the same account on every device. Anyone with the passphrase signs in as any email, so switch it on only in a deployment with its own Durable Objects and secrets, never in a preview version of the production Worker.

Plans

membershipGateway({ plans: ['premium'] }), joined with combineGatewayApps, keeps an account's plan: staff set it (setMembership({ plan, until, by, note }), from CodeFusion Console), the app's Worker and objects ask membership(), and the page reads membership.read and hears account.membership when it changes or runs out.

With directory: 'ACCOUNT_DIRECTORY' the gateway also lists its account in the app's AccountDirectory (one per app, accountDirectoryName): when it was created and last seen (at most hourly) and its plan, so staff can find and page through accounts (list({ search, plan, sort, cursor }), an id or how it starts) without visiting every gateway.

In the page

import { DeviceIdentity, deviceKeyStore, ensureDurableObjectsSession, openDurableChannel, deriveChannelCapability } from '@codefusion-cc/realtime/browser'

const device = new DeviceIdentity(deviceKeyStore('example'))
const routeId = await deriveChannelCapability(roomSecret, 'example:room:v1', 'example')
const room = await openDurableChannel({ namespace: 'example', endpointPrefix: '/api/realtime/room/', authorize, encryptionSecret: roomSecret })

Channel content and state are encrypted in the browser with keys derived from the room secret; the server sees routing capabilities and ciphertext only.

browserTabs({ namespace }) is this page's view of the other tabs of the browser: claim(key) is true in exactly one tab per key, so a chime, a ringtone or a notification happens once per browser (preferVisible lets a visible tab win), announceHandled(key) tells the others a call was answered here (onHandled), and hasSibling(tabKey) asks whether another live tab holds a tab key. Tabs live as long as their Web Lock; a tab's own lock never holds @codefusion-cc/app-update's reload, a claim does.

createDmRing({ scheme, sealing: { namespace, purpose } }) signs rings to a friend and seals them for the relay: seal(secret, ring) and open(secret, wire, { fromUserId, toUserId }), which answers the ring only between those two accounts and with a valid signature. Whether the signing device may speak for the sender stays the app's check.

What opens or verifies data (openSealedData, verifyWithDeviceKeyId, the account keys' open, openBundle and openVault, verifyDeviceSignature in the Worker) answers null or false only for what the data holds: another key, a tampered value, a key that is no key. When the platform's WebCrypto itself fails (an algorithm it lacks, a broken binding), the call rejects with that error, so an outage never reads as "no keys", "tampered" or "forged" and nothing acts on it as if it were.

An account's device goes by the browser and system it reports (describeDevice: "Chrome · macOS"; createLinkedDevices sends the page's own unless deviceBrowser gives one) until a person names it with rename(userId, deviceKeyId, name); rename(…, null) takes the name away again. A listed device's name is what to call it (the name a person gave, else its browser), named says whether a person gave it, and browser is shown beside a given name. The devices handlers bring a gateway made before 1.2.0 up to date themselves. addColumnIfMissing adds a column to an app's own Durable Object tables in their migrate step.

From @codefusion-cc/realtime/react: useLinkedDevices(devices, userId) follows the account's devices and gives their actions (approvalToShow and canLinkThisDevice in the browser entry pick what to show), and useBusyAction() runs one action at a time with its pending state and error. The browser entry's formatByteSize(bytes, locale) and splitSafeLinks(text) (only http(s) links) help show files and messages.

Testing

import { openControlSocket, testDevice, testIdToken, PROOF_PURPOSES } from '@codefusion-cc/realtime/testing'

const device = await testDevice()
const headers = await device.proofHeaders(PROOF_PURPOSES.session, 'example')        // signed as a browser signs
const token = await testIdToken('secret', { nonce: device.deviceKeyId })            // for testIdentityProvider('secret')
const control = await openControlSocket(env.USER_GATEWAYS.getByName(`example:${uid}`), { uid, deviceKeyId: device.deviceKeyId, sid })
expect((await control.command('profile.read')).type).toBe('ack')