revamped socket protocol handling

eslint.config.js

···
       '@typescript-eslint/no-unused-vars': ['warn', {varsIgnorePattern: '(?:^_)'}],
       '@typescript-eslint/no-unnecessary-condition': 'off',
       '@typescript-eslint/restrict-template-expressions': 'off',
+      '@typescript-eslint/no-unnecessary-type-parameters': 'off',
+      '@typescript-eslint/no-unnecessary-type-constraint': 'off',
       'tsdoc/syntax': 'warn',
     },
   },

readme.org

···
 Run the pwa+server with:
 
 #+BEGIN_SRC bash
-  $ pnpm install
-  $ pnpm run dev # lots of stuff concurrently with wireit
+  $ npm install
+  $ npm run dev # lots of stuff concurrently with wireit
 #+END_SRC
 
 - Common

src/client/components/peer-list.tsx

···
 import {useEffect, useState} from 'preact/hooks'
 
-import {PeerState, RealmConnection} from '#client/realm/connection'
+import {RealmConnection} from '#client/realm/connection'
+import {PeerState} from '#client/realm/types'
 import {IdentID} from '#common/protocol'
 
 export const PeerList: preact.FunctionComponent<{webrtcManager: RealmConnection}> = (props) => {
···
     const updatePeers = () => {
       queueMicrotask(() => {
         const states = webrtcManager.getPeerStates()
-        console.log('updating peers', states)
+        console.debug('updating peers', states)
         setPeers(states)
       })
     }

src/client/realm/connection.ts

···
 import WebSocket from 'isomorphic-ws'
 import {nanoid} from 'nanoid'
-import SimplePeer from 'simple-peer'
+import SimplePeer, {SimplePeerData} from 'simple-peer'
 import {z} from 'zod/v4'
 
 import {generateSignableJwt, jwkExport} from '#common/crypto/jwks'
 import {normalizeError, normalizeProtocolError, ProtocolError} from '#common/errors'
 import {
   IdentID,
-  parseJson,
-  PreauthRegisterMessage,
-  RealmBroadcastMessage,
-  realmFromServerMessageSchema,
-  RealmID,
-  RealmRtcPeerWelcomeMessage,
-  realmRtcPeerWelcomeMessageSchema,
-  RealmRtcSignalMessage,
+  PreauthRegisterRequest,
+  preauthRespSchema,
+  realmRtcPeerJoinedEventSchema,
+  realmRtcPeerLeftEventSchema,
+  realmRtcPingRequestSchema,
+  RealmRtcPongResponse,
+  realmRtcPongResponseSchema,
+  RealmRtcSignalEvent,
+  realmRtcSignalEventSchema,
 } from '#common/protocol'
-import {sendSocket, streamSocketJson, takeSocketJson} from '#common/socket'
+import {streamSocketJson, takeSocketJson} from '#common/socket'
+import {RealmPeer} from './peer'
+import {ConnectionIdentity, PeerState} from './types'
 
-/** the state of a specific peer */
-export interface PeerState {
-  /** if the peer connection is active */
-  connected: boolean
-
-  /** if the peer connection has been destroyed */
-  destroyed: boolean
-
-  /** the peer's address (ip and port) */
-  address: ReturnType<SimplePeer.Instance['address']>
-}
-
-/** identity info for connecting to a realm */
-export interface RealmIdentity {
-  realmid: RealmID
-  identid: IdentID
-  keypair: CryptoKeyPair
-}
+const realmRtcMessagesSchema = z.union([
+  realmRtcSignalEventSchema,
+  realmRtcPeerJoinedEventSchema,
+  realmRtcPeerLeftEventSchema,
+  realmRtcPongResponseSchema,
+])
 
 /** manages websocket and webrtc connections for a realm */
 export class RealmConnection extends EventTarget {
   #url: string
-  #identity: RealmIdentity
+  #identity: ConnectionIdentity
 
   #socket: WebSocket
-  #peers: Map<IdentID, SimplePeer.Instance>
+  #peers: Map<IdentID, RealmPeer>
   #nonces: Map<IdentID, string>
 
-  constructor(url: string, identity: RealmIdentity) {
+  constructor(url: string, identity: ConnectionIdentity) {
     super()
 
     this.#url = url
···
     this.#socket.onerror = this.#handleSocketError
   }
 
-  #handleSocketOpen: WebSocket['onopen'] = async () => {
-    if (this.#socket == undefined) throw new Error('socket open handler called with no socket?')
-
-    try {
-      console.debug('realm connection, socket loop open')
-      this.#dispatchCustomEvent('wsopen')
-
-      // do the auth dance
-      // TODO: this should be a state machine
-
-      const pubkey = await jwkExport.parseAsync(this.#identity.keypair.publicKey)
-      this.#socket.send(
-        await this.#signJwt({msg: 'preauth.register', pubkey} as PreauthRegisterMessage),
-      )
-
-      // the next message should be a welcome message
-      // this will throw (authenticated) otherwise
-
-      let welcome: RealmRtcPeerWelcomeMessage
-      try {
-        welcome = await takeSocketJson(this.#socket, realmRtcPeerWelcomeMessageSchema)
-      } catch (exc) {
-        const err = normalizeError(exc)
-        throw new ProtocolError('failure on authentication', 401, {cause: err})
-      }
-
-      this.#dispatchCustomEvent('wsauth', {welcome})
-
-      // we initiate connections outbound when starting up
-      // we never initiate connections to other peers otherwise
-
-      for (const peerid of welcome.peers) {
-        if (peerid === this.#identity.identid) continue
-        this.connectToPeer(peerid, true)
-      }
-
-      // then continue looping over messages on the socket and handle them
-
-      for await (const data of streamSocketJson(this.#socket)) {
-        const parse = realmFromServerMessageSchema.safeParse(data)
-
-        // not a known rtc specific message, so we don't capture it
-        if (!parse.success) {
-          this.#dispatchCustomEvent('wsdata', {data})
-          continue
-        }
-
-        // otherwise, we handle it
-        switch (parse.data.msg) {
-          case 'realm.rtc.peer-joined':
-            // new peers connect to us, we don't connect to them
-            this.#dispatchCustomEvent('peerjoined', {identid: parse.data.identid})
-            continue
-
-          case 'realm.rtc.peer-left':
-            // peer is gone, disconnect
-            this.disconnectPeer(parse.data.identid)
-            this.#dispatchCustomEvent('peerleft', {identid: parse.data.identid})
-            continue
+  get connected() {
+    return this.#socket.readyState === this.#socket.OPEN
+  }
 
-          case 'realm.rtc.signal': {
-            // some other peer is trying to send us some rtc data
-            // only connect if _they_ are the initiator - our initiating conns were above
-            let peer = this.#peers.get(parse.data.sender)
-            if (!peer && parse.data.initiator) {
-              peer = this.connectToPeer(parse.data.sender, false)
-            }
+  send<T extends unknown>(identid: IdentID, data: T) {
+    const peer = this.#peers.get(identid)
+    if (!peer?.connected) throw new Error(`Not connected to peer: ${identid}`)
 
-            // may not have a connection yet if we're waiting for them to answer
-            if (peer) {
-              peer.signal(parse.data.payload)
-            }
+    peer.send(JSON.stringify(data))
+  }
 
-            continue
-          }
-        }
-      }
-    } catch (exc) {
-      const err = normalizeProtocolError(exc)
+  sendRaw(identid: IdentID, data: SimplePeerData) {
+    const peer = this.#peers.get(identid)
+    if (!peer?.connected) throw new Error(`Not connected to peer: ${identid}`)
 
-      console.error('realm connection, socket loop error', err)
-      this.#dispatchCustomEvent('wserror', {error: err})
-    } finally {
-      console.debug('realm connection, socket loop ended')
-      this.destroy()
-    }
+    peer.send(data)
   }
 
-  #handleSocketError: WebSocket['onerror'] = (exc) => {
-    this.#dispatchCustomEvent('wserror', {error: normalizeProtocolError(exc)})
-    this.destroy()
+  broadcast(data: unknown, self = false) {
+    this.#peers.forEach((peer, identid) => {
+      if (self || identid !== this.#identity.identid) peer.send(JSON.stringify(data))
+    })
   }
 
-  #handleSocketClose: WebSocket['onclose'] = () => {
-    this.#dispatchCustomEvent('wsclose')
-    this.destroy()
-  }
+  getPeerStates(): Record<IdentID, PeerState> {
+    const states: Record<IdentID, PeerState> = {}
+    for (const [identid, peer] of this.#peers) {
+      states[identid] = {
+        identid,
+        address: peer.address(),
+        connected: peer.connected,
+        destroyed: peer.destroyed,
+      }
+    }
 
-  get connected() {
-    return this.#socket.readyState === this.#socket.OPEN
+    return states
   }
 
   destroy() {
-    console.debug('realm connection #destroy!')
+    console.debug('realm connection destroy!')
 
     if (this.connected) this.#socket.close()
-
     for (const peer of this.#peers.values()) peer.destroy()
 
     this.#peers.clear()
     this.#nonces.clear()
   }
 
-  #dispatchCustomEvent(type: string, detail?: object) {
-    this.dispatchEvent(new CustomEvent(type, {detail}))
-  }
+  // do the auth dance
+  // TODO: this should be a state machine
+  #connectionAuthenticate = async () => {
+    try {
+      const pubkey = await jwkExport.parseAsync(this.#identity.keypair.publicKey)
+      await this.#socketSignedWrite<PreauthRegisterRequest>({
+        typ: 'req',
+        msg: 'preauth.register',
+        dat: {pubkey},
+      })
 
-  /** generates and signs a JWT scoped to this identity/realm containing the given payload */
-  async #signJwt(payload: object): Promise<string> {
-    return await generateSignableJwt({
-      aud: this.#identity.realmid,
-      iss: this.#identity.identid,
-      payload,
-    }).sign(this.#identity.keypair.privateKey)
+      const resp = await takeSocketJson(this.#socket, preauthRespSchema)
+      this.#dispatchCustomEvent('wsauth', resp)
+
+      // we initiate connections outbound when starting up
+      // we never initiate connections to other peers otherwise
+      for (const peerid of resp.dat.activePeers) {
+        if (peerid === this.#identity.identid) continue
+        this.#connectPeer(peerid, true)
+      }
+    } catch (exc) {
+      const err = normalizeError(exc)
+      throw new ProtocolError('failure on authentication', 401, {cause: err})
+    }
   }
 
-  connectToPeer(remoteid: IdentID, initiator: boolean): SimplePeer.Instance {
-    let peer = this.#peers.get(remoteid)
-    if (peer) {
-      console.log(`already connected to ${remoteid}`)
-      return peer
+  // handle some message coming in on the _websocket_
+  #connectionMessage = (data: unknown) => {
+    const parse = realmRtcMessagesSchema.safeParse(data)
+    if (!parse.success) {
+      // publish non-handled data to the listeners
+      this.#dispatchCustomEvent('wsdata', {data})
+      return
     }
 
-    peer = new RealmConnectionPeer(this, nanoid(), this.#identity.identid, remoteid, initiator)
+    switch (parse.data.msg) {
+      case 'realm.rtc.pong':
+        console.debug('got a pong response', parse)
+        return
 
-    peer.on('connect', () => {
-      console.log(`connected to ${remoteid}`)
+      case 'realm.rtc.signal': {
+        // only connect if _they_ are the initiator - our initiating conns were above
+        let peer = this.#peers.get(parse.data.dat.localid)
+        if (!peer && parse.data.dat.initiator) {
+          peer = this.#connectPeer(parse.data.dat.localid, false)
+        }
 
-      this.#dispatchCustomEvent('peeropen', {remoteid})
-    })
+        // may not have a connection yet if we're waiting for them to answer
+        peer?.signal(parse.data.dat.payload)
+        return
+      }
 
-    peer.on('close', () => {
-      console.log(`Disconnected from ${remoteid}`)
+      case 'realm.rtc.peer-joined':
+        // new peers connect to us, we don't connect to them
+        this.#dispatchCustomEvent('peerjoined', {identid: parse.data.dat.identid})
+        return
 
-      this.#peers.delete(remoteid)
-      this.#nonces.delete(remoteid)
-      this.#dispatchCustomEvent('peerclose', {remoteid})
-    })
+      case 'realm.rtc.peer-left':
+        // peer is gone, disconnect
+        this.#disconnectPeer(parse.data.dat.identid)
+        this.#dispatchCustomEvent('peerleft', {identid: parse.data.dat.identid})
+        return
+    }
+  }
 
-    peer.on('error', (err) => {
-      console.error(`Error with peer ${remoteid}:`, err)
+  #connectPeer(remoteid: IdentID, initiator: boolean): RealmPeer {
+    let peer = this.#peers.get(remoteid)
+    if (!peer) {
+      peer = new RealmPeer(remoteid, nanoid(), initiator)
+      peer.on('connect', this.#handlePeerConnect.bind(this, peer))
+      peer.on('close', this.#handlePeerClose.bind(this, peer))
+      peer.on('error', this.#handlePeerError.bind(this, peer))
+      peer.on('data', this.#handlePeerData.bind(this, peer))
+      peer.on('signal', this.#handlePeerSignal.bind(this, peer))
 
-      this.#dispatchCustomEvent('peererror', {remoteid, error: err})
-    })
+      this.#peers.set(remoteid, peer)
+    }
 
-    peer.on('message', (data: unknown) => {
-      this.#dispatchCustomEvent('peerdata', {remoteid, data})
-    })
-
-    this.#peers.set(remoteid, peer)
     return peer
   }
 
-  disconnectPeer(identid: IdentID) {
+  #disconnectPeer(identid: IdentID) {
     const peer = this.#peers.get(identid)
     if (peer) {
       peer.destroy()
···
     }
   }
 
-  sendToPeer(identid: IdentID, data: unknown) {
-    const peer = this.#peers.get(identid)
-    if (peer && peer.connected) {
-      peer.send(JSON.stringify(data))
-    } else {
-      throw new Error(`Not connected to peer: ${identid}`)
-    }
-  }
+  // handle socket open is the main loop for the sendSocket
+  #handleSocketOpen: WebSocket['onopen'] = async () => {
+    if (this.#socket == undefined) throw new Error('socket open handler called with no socket?')
 
-  sendToServer(data: unknown) {
-    sendSocket(this.#socket, data)
-  }
+    console.debug('realm connection, socket loop open')
+    this.#dispatchCustomEvent('wsopen')
 
-  broadcast(data: unknown) {
-    const message = JSON.stringify(data)
-    for (const [_, peer] of this.#peers) {
-      if (peer.connected) {
-        peer.send(message)
+    try {
+      await this.#connectionAuthenticate()
+      for await (const data of streamSocketJson(this.#socket)) {
+        this.#connectionMessage(data)
       }
+    } catch (exc) {
+      const err = normalizeProtocolError(exc)
+
+      console.error('realm connection, socket loop error', err)
+      this.#dispatchCustomEvent('wserror', {error: err})
+    } finally {
+      this.destroy()
     }
   }
 
-  broadcastViaServer(data: unknown) {
-    const resp: RealmBroadcastMessage = {
-      msg: 'realm.broadcast',
-      payload: data,
-      recipients: false,
-    }
+  #handleSocketError: WebSocket['onerror'] = (exc) => {
+    this.#dispatchCustomEvent('wserror', {error: normalizeProtocolError(exc)})
+    this.destroy()
+  }
 
-    sendSocket(this.#socket, resp)
+  #handleSocketClose: WebSocket['onclose'] = () => {
+    this.#dispatchCustomEvent('wsclose')
+    this.destroy()
   }
 
-  getPeerStates(): Record<IdentID, PeerState> {
-    const states: Record<IdentID, PeerState> = {}
-    for (const [peerId, peer] of this.#peers) {
-      states[peerId] = {
-        address: peer.address(),
-        connected: peer.connected,
-        destroyed: peer.destroyed,
-      }
+  #handlePeerSignal = (peer: RealmPeer, payload: SimplePeer.SignalData) => {
+    const msg: RealmRtcSignalEvent = {
+      typ: 'evt',
+      msg: 'realm.rtc.signal',
+      dat: {
+        initiator: peer.initiator,
+        payload: JSON.stringify(payload),
+        localid: this.#identity.identid,
+        remoteid: peer.identid,
+      },
     }
 
-    return states
+    this.#socket.send(JSON.stringify(msg))
   }
-}
 
-const peerPingSchema = z.object({type: z.literal('ping'), timestamp: z.number()})
+  #handlePeerConnect = (peer: RealmPeer) => {
+    console.debug(`connected to ${peer.identid}`)
+    this.#dispatchCustomEvent('peeropen', {identid: peer.identid})
+  }
 
-/** a single webrtc peer connection within a realm */
-export class RealmConnectionPeer extends SimplePeer {
-  #connection: RealmConnection
+  #handlePeerClose = (peer: RealmPeer) => {
+    console.debug(`disconnected from ${peer.identid}`)
 
-  initiator: boolean
-  localid: IdentID
-  remoteid: IdentID
-  nonce: string
+    this.#peers.delete(peer.identid)
+    this.#nonces.delete(peer.identid)
+    this.#dispatchCustomEvent('peerclose', {identid: peer.identid})
+  }
 
-  constructor(
-    connection: RealmConnection,
-    nonce: string,
-    localid: IdentID,
-    remoteid: IdentID,
-    initiator: boolean,
-  ) {
-    super({
-      initiator,
-      config: {
-        iceServers: [
-          {urls: 'stun:stun.l.google.com:19302'},
-          {urls: 'stun:stun1.l.google.com:19302'},
-        ],
-      },
-    })
+  #handlePeerError = (peer: RealmPeer, err: Error) => {
+    console.error(`Error with peer ${peer.identid}:`, err)
+    this.#dispatchCustomEvent('peererror', {identid: peer.identid, error: err})
+  }
 
-    this.initiator = initiator
-    this.localid = localid
-    this.remoteid = remoteid
-    this.nonce = nonce
+  #handlePeerData = (peer: RealmPeer, chunk: string | Uint8Array) => {
+    const data = typeof chunk === 'string' ? chunk : new TextDecoder().decode(chunk)
+    const parsed = realmRtcPingRequestSchema.safeParse(chunk)
+    if (parsed.success) {
+      // reply to pings with pongs
+      this.send<RealmRtcPongResponse>(peer.identid, {
+        typ: 'res',
+        msg: 'realm.rtc.pong',
+        seq: parsed.data.seq,
+      })
+    } else {
+      this.#dispatchCustomEvent('peerdata', {identid: peer.identid, data})
+    }
+  }
 
-    this.#connection = connection
+  // helpers that are just too damn long
 
-    this.on('signal', this.#handlePeerSignal)
-    this.on('data', this.#handlePeerData)
+  #dispatchCustomEvent(type: string, detail?: object) {
+    this.dispatchEvent(new CustomEvent(type, {detail}))
   }
 
-  #handlePeerSignal = (e: SimplePeer.SignalData) => {
-    this.#connection.sendToServer({
-      msg: 'realm.rtc.signal',
-      initiator: this.initiator,
-      sender: this.localid,
-      recipient: this.remoteid,
-      payload: JSON.stringify(e),
-    } satisfies RealmRtcSignalMessage)
-  }
+  // typed helpers
+  // @example
+  //   this.#socketWrite<YourType>({
+  //     ... // <- get type errors here
+  //   })
+
+  #socketSignedWrite = async <T extends object>(payload: T) => {
+    const token = await generateSignableJwt({
+      aud: this.#identity.realmid,
+      iss: this.#identity.identid,
+      payload,
+    }).sign(this.#identity.keypair.privateKey)
 
-  #handlePeerData = (chunk: string) => {
-    try {
-      const ping = parseJson.pipe(peerPingSchema).safeParse(chunk)
-      if (ping.success) {
-        this.send(JSON.stringify({type: 'pong', timestamp: ping.data.timestamp}))
-      } else {
-        this.emit('message', chunk)
-      }
-    } catch (err) {
-      console.error('Failed to parse message:', err)
-    }
+    this.#socket.send(token)
   }
 }

src/client/realm/context.tsx

···
 import {createContext} from 'preact'
 import {useCallback, useEffect, useState} from 'preact/hooks'
 
-import {RealmConnection, RealmIdentity} from '#client/realm/connection'
+import {RealmConnection} from '#client/realm/connection'
+import {ConnectionIdentity} from './types'
 
 interface RealmConnectionContext {
   realm?: RealmConnection
-  identity?: RealmIdentity
-  setIdentity: (ident: RealmIdentity) => void
+  identity?: ConnectionIdentity
+  setIdentity: (ident: ConnectionIdentity) => void
 }
 
 export const RealmConnectionContext = createContext<RealmConnectionContext | null>(null)
···
   url: string
   children: preact.ComponentChildren
 }> = (props) => {
-  const [identity$, setIdentity$] = useState<RealmIdentity>()
+  const [identity$, setIdentity$] = useState<ConnectionIdentity>()
   const [connection$, setConnection$] = useState<RealmConnection>()
 
   const connect = useCallback(() => {
···
     connection$?.destroy()
   }, [connection$])
 
+  // connect on mount, or identity change
   useEffect(() => {
-    console.log('use effect in provider', identity$)
-    // connect on mount, or identity change
     if (identity$) connect()
 
     // disconnect on unsubscribe

src/client/realm/peer.ts

···
+import {IdentID} from '#common/protocol.js'
+import SimplePeer from 'simple-peer'
+
+/** a single webrtc peer connection within a realm */
+export class RealmPeer extends SimplePeer {
+  identid: IdentID
+  nonce: string
+  initiator: boolean
+
+  constructor(identid: IdentID, nonce: string, initiator: boolean) {
+    super({
+      initiator,
+      config: {
+        iceServers: [
+          {urls: 'stun:stun.l.google.com:19302'},
+          {urls: 'stun:stun1.l.google.com:19302'},
+        ],
+      },
+    })
+
+    this.identid = identid
+    this.nonce = nonce
+    this.initiator = initiator
+  }
+}

src/client/realm/types.ts

···
+import {IdentID, RealmID} from '#common/protocol'
+import SimplePeer from 'simple-peer'
+
+/** identity info for connecting to a realm */
+export interface ConnectionIdentity {
+  realmid: RealmID
+  identid: IdentID
+  keypair: CryptoKeyPair
+}
+
+/** the state of a specific peer */
+export interface PeerState {
+  identid: IdentID
+  connected: boolean
+  destroyed: boolean
+  address: ReturnType<SimplePeer.Instance['address']>
+}

src/common/protocol.ts

···
 import {z} from 'zod/v4'
+import {ProtocolError} from './errors'
+import {errorMessageSchema} from './protocol/schema'
 
 export * from './protocol/brands'
 export * from './protocol/messages'
-export * from './protocol/messages-preauth'
-export * from './protocol/messages-realm'
 
 /** a zod transformer for parsing json */
 export const parseJson = z.transform<string, unknown>((input, ctx) => {
···
     return z.NEVER
   }
 })
+
+export function makeError(
+  error: ProtocolError,
+  detail: string,
+  seq?: number,
+): z.infer<typeof errorMessageSchema> {
+  return {
+    typ: 'err',
+    msg: error.message,
+    seq,
+    dat: {
+      code: error.status,
+      detail,
+    },
+  }
+}

src/common/protocol/messages-preauth.ts

···
-import {jwkSchema} from '#common/crypto/jwks'
-import {z} from 'zod/v4'
-
-/** zod schema for `preauth.authn` message */
-export const preauthRegisterMessageSchema = z.object({
-  msg: z.literal('preauth.register'),
-  pubkey: jwkSchema,
-})
-export type PreauthRegisterMessage = z.infer<typeof preauthRegisterMessageSchema>
-
-/** zod schema for `preauth.authn` message */
-export const preauthAuthnMessageSchema = z.object({
-  msg: z.literal('preauth.authn'),
-})
-export type PreauthAuthnMessage = z.infer<typeof preauthAuthnMessageSchema>
-
-/** zod schema for any `preauth` messages */
-export const preauthMessageSchema = z.discriminatedUnion('msg', [
-  preauthRegisterMessageSchema,
-  preauthAuthnMessageSchema,
-])
-export type PreauthMessage = z.infer<typeof preauthMessageSchema>

src/common/protocol/messages-realm.ts

···
-import {z} from 'zod/v4'
-
-import {IdentBrand} from './brands'
-import {responseOkSchema} from './messages'
-
-/**
- * zod schema for `realm.broadcast` message
- *
- * recipients = true, include self in broadcast
- * recipients = false, exclude self in broadcast (default)
- * recipients = [], use these exact recipients
- */
-export const realmBroadcastMessageSchema = z.object({
-  msg: z.literal('realm.broadcast'),
-  payload: z.any(),
-  recipients: z.union([z.boolean(), z.array(IdentBrand.schema)]).default(false),
-})
-
-export type RealmBroadcastMessage = z.infer<typeof realmBroadcastMessageSchema>
-
-// rtc messages
-
-export const realmRtcSignalMessageSchema = z.object({
-  msg: z.literal('realm.rtc.signal'),
-  payload: z.string(),
-  sender: IdentBrand.schema,
-  recipient: IdentBrand.schema,
-  initiator: z.boolean(),
-})
-
-export const realmRtcPeerWelcomeMessageSchema = responseOkSchema.extend({
-  msg: z.literal('realm.rtc.peer-welcome'),
-  peers: z.array(IdentBrand.schema),
-})
-
-export const realmRtcPeerJoinedMessageSchema = responseOkSchema.extend({
-  msg: z.literal('realm.rtc.peer-joined'),
-  identid: IdentBrand.schema,
-})
-
-export const realmRtcPeerLeftMessageSchema = responseOkSchema.extend({
-  msg: z.literal('realm.rtc.peer-left'),
-  identid: IdentBrand.schema,
-})
-
-export type RealmRtcSignalMessage = z.infer<typeof realmRtcSignalMessageSchema>
-export type RealmRtcPeerWelcomeMessage = z.infer<typeof realmRtcPeerWelcomeMessageSchema>
-export type RealmRtcPeerJoinedMessage = z.infer<typeof realmRtcPeerJoinedMessageSchema>
-export type RealmRtcPeerLeftMessage = z.infer<typeof realmRtcPeerLeftMessageSchema>
-
-/// useful unions
-
-export const realmToServerMessageSchema = z.discriminatedUnion('msg', [
-  realmBroadcastMessageSchema,
-  realmRtcSignalMessageSchema,
-])
-
-export const realmFromServerMessageSchema = z.discriminatedUnion('msg', [
-  realmRtcPeerJoinedMessageSchema,
-  realmRtcPeerLeftMessageSchema,
-  realmRtcSignalMessageSchema,
-])
-
-export type RealmToServerMessage = z.infer<typeof realmToServerMessageSchema>
-export type RealmFromServerMessage = z.infer<typeof realmFromServerMessageSchema>

src/common/protocol/messages.ts

···
+import {jwkSchema} from '#common/crypto/jwks'
 import {z} from 'zod/v4'
 
-/** zod schema for `ok` message */
-export const responseOkSchema = z.object({
-  ok: z.literal(true),
-})
-export type ResponseOk = z.infer<typeof responseOkSchema>
+import {IdentBrand} from './brands'
+import {
+  makeEmptyRequestSchema,
+  makeEmptyResponseSchema,
+  makeEventSchema,
+  makeRequestSchema,
+  makeResponseSchema,
+} from './schema'
+
+/// preauth
+
+export const preauthRegisterReqSchema = makeRequestSchema(
+  'preauth.register',
+  z.object({pubkey: jwkSchema}),
+)
+
+export const preauthAuthnReqSchema = makeEmptyRequestSchema('preauth.authn')
+
+export const preauthRespSchema = makeResponseSchema(
+  'preauth.authn',
+  z.object({
+    activePeers: z.array(IdentBrand.schema),
+  }),
+)
+
+export const preauthReqSchema = z.union([preauthAuthnReqSchema, preauthRegisterReqSchema])
+
+export type PreauthRegisterRequest = z.infer<typeof preauthRegisterReqSchema>
+export type PreauthAuthnRequest = z.infer<typeof preauthAuthnReqSchema>
+export type PreauthResponse = z.infer<typeof preauthRespSchema>
 
-/** zod schema for `error` message */
-export const responseErrorSchema = z.object({
-  ok: z.literal(false),
-  status: z.number(),
-  message: z.string(),
-})
-export type ErrorResponse = z.infer<typeof responseErrorSchema>
+/// realms
+
+export const realmBroadcastEventSchema = makeEventSchema(
+  'realm.broadcast',
+  z.object({
+    payload: z.unknown(),
+    recipients: z.union([z.boolean(), z.array(IdentBrand.schema)]).default(false),
+  }),
+)
+
+export const realmRtcPingRequestSchema = makeEmptyRequestSchema('realm.rtc.ping')
+export const realmRtcPongResponseSchema = makeEmptyResponseSchema('realm.rtc.pong')
+export const realmRtcPingPongMessageSchema = z.union([
+  realmRtcPingRequestSchema,
+  realmRtcPongResponseSchema,
+])
+
+export const realmRtcSignalEventSchema = makeEventSchema(
+  'realm.rtc.signal',
+  z.object({
+    payload: z.string(),
+    initiator: z.boolean(),
+    localid: IdentBrand.schema,
+    remoteid: IdentBrand.schema,
+  }),
+)
+
+export const realmRtcPeerJoinedEventSchema = makeEventSchema(
+  'realm.rtc.peer-joined',
+  z.object({identid: IdentBrand.schema}),
+)
+
+export const realmRtcPeerLeftEventSchema = makeEventSchema(
+  'realm.rtc.peer-left',
+  z.object({identid: IdentBrand.schema}),
+)
+
+export type RealmBroadcastEvent = z.infer<typeof realmBroadcastEventSchema>
+export type RealmRtcPingRequest = z.infer<typeof realmRtcPingRequestSchema>
+export type RealmRtcPongResponse = z.infer<typeof realmRtcPongResponseSchema>
+export type RealmRtcSignalEvent = z.infer<typeof realmRtcSignalEventSchema>
+export type RealmRtcPeerJoinedEvent = z.infer<typeof realmRtcPeerJoinedEventSchema>
+export type RealmRtcPeerLeftEvent = z.infer<typeof realmRtcPeerLeftEventSchema>

src/common/protocol/schema.ts

···
+import {z} from 'zod/v4'
+
+const errorSchema = z.object({code: z.number(), detail: z.string().optional()})
+
+export const eventMessageSchema = z.object({
+  typ: z.literal('evt'),
+  msg: z.string(),
+  dat: z.unknown(),
+})
+
+export const requestMessageSchema = z.object({
+  typ: z.literal('req'),
+  msg: z.string(),
+  seq: z.number().optional(),
+  dat: z.unknown(),
+})
+
+export const responseMessageSchema = z.object({
+  typ: z.literal('res'),
+  msg: z.string(),
+  seq: z.number().optional(),
+  dat: z.unknown(),
+})
+
+export const errorMessageSchema = z.object({
+  typ: z.literal('err'),
+  msg: z.string(),
+  seq: z.number().optional(),
+  dat: errorSchema,
+})
+
+export const messageSchema = z.discriminatedUnion('typ', [
+  eventMessageSchema,
+  requestMessageSchema,
+  responseMessageSchema,
+  errorMessageSchema,
+])
+
+export const makeEventSchema = <N extends string, Z extends z.ZodType>(name: N, schema: Z) => {
+  return eventMessageSchema.extend({
+    msg: z.literal(name),
+    dat: schema,
+  })
+}
+
+export const makeEmptyEventSchema = <N extends string>(name: N) => {
+  return eventMessageSchema.extend({msg: z.literal(name)}).omit({dat: true})
+}
+
+export const makeRequestSchema = <N extends string, Z extends z.ZodType>(name: N, schema: Z) => {
+  return requestMessageSchema.extend({
+    msg: z.literal(name),
+    dat: schema,
+  })
+}
+
+export const makeEmptyRequestSchema = <N extends string>(name: N) => {
+  return requestMessageSchema.extend({msg: z.literal(name)}).omit({dat: true})
+}
+
+export const makeResponseSchema = <N extends string, Z extends z.ZodType>(name: N, schema: Z) => {
+  return responseMessageSchema.extend({
+    msg: z.literal(name),
+    dat: schema,
+  })
+}
+
+export const makeEmptyResponseSchema = <N extends string>(name: N) => {
+  return responseMessageSchema.extend({msg: z.literal(name)}).omit({dat: true})
+}

src/common/socket.ts

···
 
 import {parseJson} from './protocol'
 
-/** send some data in json format down the wire */
-export function sendSocket(ws: WebSocket, data: unknown): void {
-  ws.send(JSON.stringify(data))
-}
-
 /**
  * given a websocket, wait and take a single message off and return it
  *

src/common/types.ts

···
+export type Replace<K extends keyof U, U, V> = Omit<U, K> & Record<K, V>

src/server/index.ts

···
   // WebSocket handling
   const wss = new WebSocketServer({server, path: '/stream'})
   wss.on('connection', (ws) => {
-    socketHandler(ws)
-      .catch((e: unknown) => {
-        console.error('uncaught error from websocket', e)
-      })
-      .finally(() => {
-        console.log('socket handler complete')
-      })
+    socketHandler(ws).catch((e: unknown) => {
+      console.error('uncaught error from websocket', e)
+    })
   })
 
   return server

src/server/routes-error.ts

···
 import * as express from 'express'
 
 export const notFoundHandler: express.RequestHandler = (req, res) => {
-  console.log(req.url)
   res.status(404).send('wut')
 }

src/server/routes-socket/handler-preauth.ts

···
 import {jwkImport} from '#common/crypto/jwks'
 import {jwtPayload, verifyJwtToken} from '#common/crypto/jwts'
 import {normalizeError, ProtocolError} from '#common/errors'
-import {IdentBrand, IdentID, preauthMessageSchema, RealmBrand, RealmID} from '#common/protocol'
+import {
+  IdentBrand,
+  IdentID,
+  preauthReqSchema,
+  PreauthResponse,
+  RealmBrand,
+  RealmID,
+} from '#common/protocol'
 import {takeSocket} from '#common/socket'
 
 import * as realms from './state'
···
 
   try {
     const data = await takeSocket(ws, combinedSignal)
+    const jwt = await jwtPayload(preauthReqSchema).parseAsync(data)
 
     // if any of the parsing fails, it'll throw a zod error
-    const jwt = await jwtPayload(preauthMessageSchema).parseAsync(data)
     const identid = IdentBrand.parse(jwt.claims.iss)
     const realmid = RealmBrand.parse(jwt.claims.aud)
 
     // if we're registering, make sure the realm exists
     if (jwt.payload.msg === 'preauth.register') {
-      const registrantkey = await jwkImport.parseAsync(jwt.payload.pubkey)
+      const registrantkey = await jwkImport.parseAsync(jwt.payload.dat.pubkey)
       realms.ensureRegisteredRealm(realmid, identid, registrantkey)
     }
 
-    return await authenticatePreauth(realmid, identid, jwt.token)
+    const auth = await authenticatePreauth(realmid, identid, jwt.token)
+    const msg = preauthResponse(auth, jwt.payload.seq)
+    ws.send(JSON.stringify(msg))
+
+    return auth
   } finally {
     timeout.cancel()
   }
···
     throw new ProtocolError('jwt verification failed', 401, {cause: err})
   }
 }
+
+function preauthResponse(auth: realms.AuthenticatedIdentity, seq?: number): PreauthResponse {
+  return {
+    typ: 'res',
+    msg: 'preauth.authn',
+    dat: {
+      activePeers: Array.from(auth.realm.sockets.keys()),
+    },
+    seq,
+  }
+}

src/server/routes-socket/handler-realm.ts

···
 import {WebSocket} from 'isomorphic-ws'
+import {z} from 'zod/v4'
 
 import {normalizeProtocolError, ProtocolError} from '#common/errors'
 import * as protocol from '#common/protocol'
-import {sendSocket, streamSocket} from '#common/socket'
+import {streamSocket} from '#common/socket'
 import * as realm from '#server/routes-socket/state'
 
+// what can the server handle?
+const incomingMessageSchema = z.union([
+  protocol.realmBroadcastEventSchema,
+  protocol.realmRtcSignalEventSchema,
+])
+
 /**
  * ance we've retrieved authentication details, we go into the main realm loop.
  * read messages as they come in and dispatch actions.
···
   signal?: AbortSignal,
 ) {
   realmBroadcast(auth, buildRtcPeerJoined(auth))
-  sendSocket(ws, buildRtcPeerWelcome(auth))
-
-  const parser = protocol.parseJson.pipe(protocol.realmToServerMessageSchema)
 
   try {
-    for await (const data of streamSocket(ws, {signal})) {
+    const incomingParser = protocol.parseJson.pipe(incomingMessageSchema)
+    for await (const msg of streamSocket(ws, {signal})) {
       try {
-        const msg = await parser.parseAsync(data)
-        switch (msg.msg) {
+        const data = await incomingParser.parseAsync(msg)
+        switch (data.msg) {
           case 'realm.broadcast':
-            realmBroadcast(auth, msg.payload, msg.recipients)
+            realmBroadcast(auth, data.dat, data.dat.recipients)
             continue
 
           case 'realm.rtc.signal':
-            realmBroadcast(auth, msg, [msg.recipient])
+            realmBroadcast(auth, data, [data.dat.remoteid])
             continue
 
           default:
···
         if (error.status >= 500) throw error
 
         if (ws.readyState === ws.OPEN) {
-          sendSocket(ws, buildRealmError(error))
+          const err = protocol.makeError(error, 'error in realm loop')
+          ws.send(JSON.stringify(err))
         }
       }
     }
   } finally {
-    console.log('client left!', auth)
     realmBroadcast(auth, buildRtcPeerLeft(auth))
   }
 }
 
-function buildRtcPeerWelcome(
-  auth: realm.AuthenticatedIdentity,
-): protocol.RealmRtcPeerWelcomeMessage {
+function buildRtcPeerJoined(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerJoinedEvent {
   return {
-    ok: true,
-    msg: 'realm.rtc.peer-welcome',
-    peers: Array.from(auth.realm.sockets.keys()),
-  }
-}
-
-function buildRtcPeerJoined(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerJoinedMessage {
-  return {
-    ok: true,
+    typ: 'evt',
     msg: 'realm.rtc.peer-joined',
-    identid: auth.identid,
+    dat: {identid: auth.identid},
   }
 }
 
-function buildRtcPeerLeft(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerLeftMessage {
+function buildRtcPeerLeft(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerLeftEvent {
   return {
-    ok: true,
+    typ: 'evt',
     msg: 'realm.rtc.peer-left',
-    identid: auth.identid,
-  }
-}
-
-function buildRealmError(error: ProtocolError): protocol.ErrorResponse {
-  return {
-    ok: false,
-    message: error.message,
-    status: error.status,
+    dat: {identid: auth.identid},
   }
 }
 
···
 ) {
   const echo = recipients === true || Array.isArray(recipients)
   const recips = Array.isArray(recipients) ? recipients : Array.from(auth.realm.identities.keys())
+  const json = JSON.stringify(payload)
 
   for (const recip of recips) {
     if (!echo && recip === auth.identid) continue
···
     const sockets = auth.realm.sockets.get(recip)
     if (sockets) {
       for (const socket of sockets) {
-        sendSocket(socket, payload)
+        socket.send(json)
       }
     }
   }

src/server/routes-socket/handler.ts

···
 
 /** when the socket connects, we drive our protocol through handlers */
 export async function socketHandler(ws: WebSocket) {
-  console.log('WebSocket connection established')
-
   try {
     const auth = await preauthHandler(ws)
     try {