refactoring of client side connection stuff

src/feedline/client/context/identity.tsx

···
-import {ParentProps, createContext, createMemo, createResource, useContext} from 'solid-js'
+import {ParentProps, createContext, createMemo, createResource, onCleanup, useContext} from 'solid-js'
 
-import {RealmIdentityStore} from '#realm/client/identity-store.js'
+import {RealmIdentityStore} from '#realm/client/identity-store'
 
 const makeIdentityContext = () => {
   const store = createMemo(() => new RealmIdentityStore())
   const [identity] = createResource(() => store().ensure())
-  return identity
+
+  onCleanup(() => {
+    identity()?.destroy()
+  })
+
+  return {
+    identity,
+    store,
+    register: (upstream: string) => identity()?.register(upstream),
+    exchange: (upstream: string, invitation: string) => identity()?.exchangeInvite(upstream, invitation),
+  }
 }
 
 export type IdentityContext = ReturnType<typeof makeIdentityContext>

src/feedline/client/index.tsx

···
 type MyAction = z.infer<typeof myActionSchema>
 
 function Name() {
+  const {identity, store} = useIdentityContext()
+
   createEffect(() => {
     console.log('identity:', identity.loading, identity.error, identity())
   })
 
-  const identity = useIdentityContext()
-  const addAction = async () => {
-    await identity()?.dispatchAction<MyAction>('my-action', identity()?.identid ?? IdentBrand.generate())
+  const addAction = () => {
+    identity()
+      ?.dispatchAction<MyAction>('my-action', identity()?.identid ?? IdentBrand.generate())
+      .catch((exc: unknown) => {
+        console.error('error dispatching action', exc)
+      })
+  }
+
+  const registerAction = () => {
+    identity()?.register('ws://localhost:4000/stream')
   }
 
   return (
···
       <h2>loading: {identity.loading}</h2>
       <h2>error: {identity.error}</h2>
       <button onClick={addAction}>Add</button>
+      <button onClick={registerAction}>Register</button>
     </>
   )
 }
···
     <IdentityProvider>
       <Layout>
         <h2>Hi</h2>
+        <Name />
       </Layout>
     </IdentityProvider>
   )

src/feedline/client/layout.tsx

···
 import {ParentProps} from 'solid-js'
 
-import {useIdentityContext} from './context/identity'
-
 export default function Layout(props: ParentProps) {
-  const identity = useIdentityContext()
   return (
     <>
       <header>
         <span>feedline</span>
-        <span>{identity()?.identid}</span>
       </header>
       <nav>sidebar nav in desktop</nav>
       <main>

src/lib/async/aborts.ts

···
   return {signal: controller.signal, cancel}
 }
 
+export function timeoutAbort(fn: () => void, ms: number, signal?: AbortSignal): {cancel: () => void} {
+  const timeout = setTimeout(fn, ms)
+  const cancel = () => {
+    clearTimeout(timeout)
+    signal?.removeEventListener('abort', cancel)
+  }
+
+  signal?.addEventListener('abort', cancel)
+  return {cancel}
+}
+
 /**
  * @param signals - the list of signals to combine
  * @returns a combined signal controller, which will abort when any given signal does, or be automatically aborted

src/lib/async/events.ts

···
+export function waitEvent(type: string, target: EventTarget, signal?: AbortSignal): Promise<Event> {
+  const {promise, resolve, reject} = Promise.withResolvers<Event>()
+
+  const handler = (event: Event) => {
+    signal?.throwIfAborted()
+    target.removeEventListener(type, handler)
+    resolve(event)
+  }
+
+  signal?.addEventListener('abort', reject)
+  target.addEventListener(type, handler)
+
+  return promise
+}

src/lib/async/fence.ts

···
+import {normalizeError} from '#lib/errors'
+
+/**
+ * Simple async fence, for blocking many async ops until a condition is met
+ */
+export class Fence {
+  #open: boolean
+  #resolvers: Array<() => void> = []
+
+  constructor(open = false) {
+    this.#open = open
+  }
+
+  get value() {
+    return this.#open
+  }
+
+  set value(v: boolean) {
+    if (v) {
+      this.open()
+    } else {
+      this.close()
+    }
+  }
+
+  enter(signal?: AbortSignal): Promise<boolean> {
+    if (this.#open) return Promise.resolve(true)
+
+    return new Promise((resolve, reject) => {
+      if (this.#open) {
+        resolve(true)
+      } else if (signal?.aborted) {
+        reject(normalizeError(signal.reason))
+      } else {
+        let cleanup: (() => void) | undefined
+        const resolver = () => {
+          cleanup?.()
+          resolve(true)
+        }
+
+        if (signal) {
+          const onabort = () => {
+            const idx = this.#resolvers.indexOf(resolver)
+            if (idx >= 0) {
+              this.#resolvers.splice(idx, 1)
+            }
+
+            reject(normalizeError(signal.reason))
+          }
+
+          signal.addEventListener('abort', onabort, {once: true})
+          cleanup = () => {
+            signal.removeEventListener('abort', onabort)
+          }
+        }
+
+        this.#resolvers.push(resolver)
+      }
+    })
+  }
+
+  open(): void {
+    this.#open = true
+
+    const waiters = [...this.#resolvers]
+    this.#resolvers = []
+
+    waiters.forEach(queueMicrotask)
+  }
+
+  close(): void {
+    this.#open = false
+  }
+}

src/lib/async/sleep.ts

···
   }
 }
 
-export function backoff(options?: {maxAttempts?: number; baseDelay?: number; maxDelay?: number}) {
+export function backoff(options?: {
+  maxAttempts?: number
+  baseDelay?: number
+  maxDelay?: number
+}): ((signal?: AbortSignal) => Promise<void>) & {attempts: number; delay: number} {
   const maxAttempts = options?.maxAttempts ?? 10
   const baseDelay = options?.baseDelay ?? 1000
   const maxDelay = options?.maxDelay ?? 30_000
 
   let attempts = 0
+  let currentDelay = 0
+
   const nextDelay = () => {
     return attempts === 0
       ? attempts++ // immediate at 0
       : Math.min(baseDelay * Math.pow(2, attempts++ - 1), maxDelay)
   }
 
-  return async (signal?: AbortSignal) => {
+  const delayFn = async (signal?: AbortSignal) => {
     signal?.throwIfAborted()
     if (attempts > maxAttempts) throw new Error('exceeded max attempts!')
 
-    await sleep(nextDelay(), signal)
+    currentDelay = nextDelay()
+    await sleep(currentDelay, signal)
   }
+
+  Object.defineProperty(delayFn, 'attempts', {
+    get() {
+      return attempts
+    },
+  })
+
+  Object.defineProperty(delayFn, 'delay', {
+    get() {
+      return currentDelay
+    },
+  })
+
+  return delayFn as ((signal?: AbortSignal) => Promise<void>) & {attempts: number; delay: number}
 }

src/lib/client/webrtc.ts

···
+import {match} from 'ts-pattern'
+import {z} from 'zod/v4'
+
+import {controllerWithSignals, timeoutAbort} from '#lib/async/aborts'
+import {BlockingQueue} from '#lib/async/blocking-queue'
+import {Fence} from '#lib/async/fence'
+import {normalizeError} from '#lib/errors'
+import {TypedEventTarget} from '#lib/events'
+
+export const rtcSessionDescriptionSchema = z.object({
+  sdp: z.string().optional(),
+  type: z.enum(['answer', 'offer', 'pranswer', 'rollback']),
+})
+
+export const rtcIceCandidateSchema = z.object({
+  candidate: z.string().optional(),
+  sdpMLineIndex: z.number().nullish(),
+  sdpMid: z.string().nullish(),
+  usernameFragment: z.string().nullish(),
+})
+
+export const rtcSignalDataSchema = z.discriminatedUnion('type', [
+  z.object({type: z.literal('offer'), sdp: rtcSessionDescriptionSchema}),
+  z.object({type: z.literal('answer'), sdp: rtcSessionDescriptionSchema}),
+  z.object({type: z.literal('candidate'), candidate: rtcIceCandidateSchema}),
+])
+
+export const rtcSignalPayloadSchema = z.object({
+  signal: rtcSignalDataSchema,
+  initiator: z.boolean(),
+})
+
+export const peerConnectionStatsSchema = z.object({
+  type: z.literal('candidate-pair'),
+  state: z.literal('succeeded'),
+  currentRoundTripTime: z.number().optional(),
+  bytesReceived: z.number().optional(),
+  bytesSent: z.number().optional(),
+})
+
+export type RTCSignalData = z.infer<typeof rtcSignalDataSchema>
+export type RTCSignalPayload = z.infer<typeof rtcSignalPayloadSchema>
+
+export type PeerConnectionStats = z.infer<typeof peerConnectionStatsSchema>
+export type DataChannelSendable = string | Blob | ArrayBuffer | ArrayBufferView<ArrayBuffer>
+
+export type DataChannelEventMap = {
+  signal: CustomEvent<RTCSignalData>
+  connect: CustomEvent<never>
+  close: CustomEvent<never>
+  error: CustomEvent<Error>
+  data: CustomEvent<string | ArrayBuffer>
+}
+
+const BUFFER_LOWWATER = 512 * 1024 // ~512KB
+
+export class DataChannelPeer<
+  T extends DataChannelEventMap = DataChannelEventMap,
+> extends TypedEventTarget<T> {
+  readonly initiator: boolean
+
+  #peer: RTCPeerConnection
+  #chan: RTCDataChannel | null = null
+  #candidates: RTCIceCandidateInit[] = []
+
+  #abort: AbortController
+  #queue = new BlockingQueue<DataChannelSendable>()
+
+  #polite: boolean
+  #pendingOffer: RTCSessionDescriptionInit | null = null
+  #ignoreOffer = false
+
+  constructor(initiator: boolean, config?: RTCConfiguration, signal?: AbortSignal) {
+    super()
+
+    this.initiator = initiator
+    this.#polite = !initiator
+
+    this.#abort = controllerWithSignals(signal)
+    const opts = {signal: this.#abort.signal}
+
+    this.#peer = new RTCPeerConnection(config)
+    this.#peer.addEventListener('icecandidate', this.#onPeerCandidate, opts)
+    this.#peer.addEventListener('connectionstatechange', this.#onPeerState, opts)
+
+    if (initiator) {
+      this.#chan = this.#peer.createDataChannel('data', {ordered: true, maxRetransmits: 10})
+      this.#setupDataChannel()
+      this.#advertiseDataChannel().catch((exc: unknown) => {
+        console.error('unexpected error in data channel offer', exc)
+        this.dispatchCustomEvent('error', normalizeError(exc))
+      })
+    } else {
+      this.#peer.addEventListener(
+        'datachannel',
+        (event) => {
+          this.#chan = event.channel
+          this.#setupDataChannel()
+        },
+        opts,
+      )
+    }
+  }
+
+  async connectionStats() {
+    const stats = await this.#peer.getStats()
+    for (const report of stats.values()) {
+      const parsed = peerConnectionStatsSchema.safeParse(report)
+      if (parsed.success) return parsed.data
+    }
+
+    return null
+  }
+
+  #onPeerCandidate = (event: RTCPeerConnectionIceEvent) => {
+    if (event.candidate) {
+      this.dispatchCustomEvent('signal', {type: 'candidate', candidate: event.candidate.toJSON()})
+    }
+  }
+
+  #onPeerState = () => {
+    const state = this.#peer.connectionState
+    if (state === 'failed') {
+      this.#restartIce().catch(() => {
+        this.dispatchCustomEvent('error', new Error('ice restart failed'))
+        this.dispatchCustomEvent('close')
+      })
+    } else if (state === 'disconnected') {
+      timeoutAbort(
+        () => {
+          if (this.#peer.connectionState === 'disconnected') {
+            this.dispatchCustomEvent('close')
+          }
+        },
+        5_000,
+        this.#abort.signal,
+      )
+    } else if (state === 'closed') {
+      this.dispatchCustomEvent('close')
+    }
+  }
+
+  async #restartIce() {
+    console.log('restarting ice negotiation for', this.#peer)
+
+    // nobody yet, we're starting over
+    this.#candidates = []
+
+    // trigger ice restart
+    const offer = await this.#peer.createOffer({iceRestart: true})
+    await this.#peer.setLocalDescription(offer)
+    if (this.#peer.localDescription) {
+      this.dispatchCustomEvent('signal', {type: 'offer', sdp: this.#peer.localDescription.toJSON()})
+    }
+  }
+
+  #setupDataChannel() {
+    if (!this.#chan) return
+
+    const opts = {signal: this.#abort.signal}
+
+    this.#chan.addEventListener(
+      'open',
+      () => {
+        this.dispatchCustomEvent('connect')
+      },
+      opts,
+    )
+
+    this.#chan.addEventListener('close', () => this.dispatchCustomEvent('close'), opts)
+    this.#chan.addEventListener('error', (e) => this.dispatchCustomEvent('error', normalizeError(e)), opts)
+    this.#chan.addEventListener(
+      'message',
+      (e: MessageEvent<string | ArrayBuffer>) => this.dispatchCustomEvent('data', e.data),
+      opts,
+    )
+
+    // drain loop blocks on lowwater, which it handles internally
+    void this.#drainLoop(this.#chan).catch((exc: unknown) => {
+      console.error('unexpected error in peer drain loop!', exc)
+      this.dispatchCustomEvent('error', normalizeError(exc))
+    })
+  }
+
+  async #advertiseDataChannel() {
+    try {
+      this.#pendingOffer = await this.#peer.createOffer()
+      await this.#peer.setLocalDescription(this.#pendingOffer)
+      if (this.#peer.localDescription) {
+        this.dispatchCustomEvent('signal', {type: 'offer', sdp: this.#peer.localDescription.toJSON()})
+      }
+    } finally {
+      this.#pendingOffer = null
+    }
+  }
+
+  signal(data: unknown) {
+    const parsed = rtcSignalDataSchema.safeParse(data)
+    if (!parsed.success) {
+      this.dispatchCustomEvent('error', parsed.error)
+      return
+    }
+
+    const promise = match(parsed.data)
+      .with({type: 'offer'}, (d) => this.#signalOffer(d.sdp))
+      .with({type: 'answer'}, (d) => this.#signalAnswer(d.sdp))
+      .with({type: 'candidate'}, (d) => this.#signalCandidate(d.candidate))
+      .exhaustive()
+
+    promise
+      .then(() => this.#flushCandidates())
+      .catch((exc: unknown) => this.dispatchCustomEvent('error', normalizeError(exc)))
+  }
+
+  #signalOffer = async (sdp: RTCSessionDescriptionInit) => {
+    const offerCollision =
+      sdp.type === 'offer' && (this.#pendingOffer !== null || this.#peer.signalingState !== 'stable')
+
+    this.#ignoreOffer = !this.#polite && offerCollision
+    if (this.#ignoreOffer) {
+      console.log('ignoring offer due to glare (impolite peer)')
+      return
+    }
+
+    if (offerCollision) {
+      console.log('offer collision detected, rolling back (polite peer)')
+      await this.#peer.setLocalDescription({type: 'rollback'})
+    }
+
+    await this.#peer.setRemoteDescription(new RTCSessionDescription(sdp))
+
+    const answer = await this.#peer.createAnswer()
+    await this.#peer.setLocalDescription(answer)
+    if (this.#peer.localDescription) {
+      this.dispatchCustomEvent('signal', {type: 'answer', sdp: this.#peer.localDescription.toJSON()})
+    }
+  }
+
+  #signalAnswer = async (sdp: RTCSessionDescriptionInit) => {
+    await this.#peer.setRemoteDescription(new RTCSessionDescription(sdp))
+  }
+
+  #signalCandidate = async (candidate: RTCIceCandidateInit) => {
+    if (this.#peer.remoteDescription) {
+      await this.#peer.addIceCandidate(new RTCIceCandidate(candidate))
+    } else {
+      this.#candidates.push(candidate)
+    }
+  }
+
+  async #flushCandidates() {
+    if (!this.#peer.remoteDescription) return
+
+    const toflush = this.#candidates.splice(0)
+    await Promise.all(
+      toflush.map(async (candidate) => {
+        try {
+          await this.#peer.addIceCandidate(candidate)
+        } catch (exc: unknown) {
+          this.dispatchCustomEvent('error', normalizeError(exc))
+        }
+      }),
+    )
+  }
+
+  send(data: DataChannelSendable): void {
+    this.#queue.enqueue(data)
+  }
+
+  async #drainLoop(chan: RTCDataChannel) {
+    const lowwater = new Fence(chan.bufferedAmount < BUFFER_LOWWATER)
+    chan.bufferedAmountLowThreshold = BUFFER_LOWWATER
+    chan.addEventListener(
+      'bufferedamountlow',
+      () => {
+        lowwater.open()
+      },
+      {signal: this.#abort.signal},
+    )
+
+    try {
+      let data: DataChannelSendable
+      while ((data = await this.#queue.dequeue(this.#abort.signal))) {
+        await lowwater.enter(this.#abort.signal) // wait for peer buffer capacity
+
+        // channel closed while waiting?
+        if (chan.readyState !== 'open') {
+          this.#queue.prequeue(data)
+          break
+        }
+
+        try {
+          // typescript gets hungup on the type check, because of how the dom types for data channel do overloads
+          // but we know that the data we have can go through, and at runtime it'll do whatever dispatch it was going to anyway
+          chan.send(data as string)
+        } catch (exc: unknown) {
+          this.#queue.prequeue(data)
+          this.dispatchCustomEvent('error', normalizeError(exc))
+          continue
+        }
+
+        lowwater.value = chan.bufferedAmount < BUFFER_LOWWATER
+      }
+    } catch (exc: unknown) {
+      if (this.#abort.signal.aborted) return
+      throw exc
+    }
+  }
+
+  destroy() {
+    this.#chan?.close()
+    this.#peer.close()
+
+    this.#abort.abort('shutting down')
+  }
+}

src/lib/events.ts

···
+export type TypedEventMap = Record<string, CustomEvent>
+
+export type TypedEventListener<M extends TypedEventMap, K extends keyof M> = (
+  evt: M[K],
+) => void | Promise<void>
+
+export interface TypedEventListenerObject<M extends TypedEventMap, K extends keyof M> {
+  handleEvent: (evt: M[K]) => void | Promise<void>
+}
+
+export type TypedEventListenerOrEventListenerObject<M extends TypedEventMap, K extends keyof M> =
+  | TypedEventListener<M, K>
+  | TypedEventListenerObject<M, K>
+
+/* eslint-disable @typescript-eslint/no-unsafe-declaration-merging */
+
+export interface TypedEventTarget<M extends TypedEventMap> {
+  addEventListener: <K extends keyof M & string>(
+    type: K,
+    listener: TypedEventListenerOrEventListenerObject<M, K> | null,
+    options?: boolean | AddEventListenerOptions,
+  ) => void
+
+  removeEventListener: <K extends keyof M & string>(
+    type: K,
+    callback: TypedEventListenerOrEventListenerObject<M, K> | null,
+    options?: EventListenerOptions | boolean,
+  ) => void
+
+  dispatchEvent: (event: Event) => boolean
+}
+
+export class TypedEventTarget<M extends TypedEventMap> extends EventTarget implements TypedEventTarget<M> {
+  dispatchCustomEvent<K extends keyof M>(type: K, detail?: M[K]['detail'], opts: Partial<EventInit> = {}) {
+    const event = new CustomEvent<M[K]['detail']>(type as string, {...opts, detail})
+    return super.dispatchEvent(event)
+  }
+}
+
+/* eslint-enable @typescript-eslint/no-unsafe-declaration-merging */

src/realm/client/action-store.ts

···
+import Dexie, {Collection, Table} from 'dexie'
+
+import {TypedEventTarget} from '#lib/events'
+
+import {LogicalClock} from '#realm/logical-clock'
+import {RealmAction} from '#realm/schema'
+import {IdentBrand, IdentID} from '#realm/schema/brands'
+import {PeerClocks, Timestamp, compare, explode, generate} from '#realm/schema/timestamp'
+
+export interface StoredAction {
+  identid: IdentID
+  action: RealmAction
+}
+
+export interface StoredClock {
+  identid: IdentID
+  clock: Timestamp
+}
+
+export type ActionEventMap = {
+  action: CustomEvent<RealmAction>
+}
+
+export class ActionStore extends Dexie implements TypedEventTarget<ActionEventMap> {
+  private actions!: Table<StoredAction, Timestamp>
+  private clocks!: Table<StoredClock, IdentID>
+
+  #events = new TypedEventTarget<ActionEventMap>()
+  addEventListener = this.#events.addEventListener.bind(this.#events)
+  dispatchEvent = this.#events.dispatchEvent.bind(this.#events)
+  dispatchCustomEvent = this.#events.dispatchCustomEvent.bind(this.#events)
+  removeEventListener = this.#events.removeEventListener.bind(this.#events)
+
+  #identid: IdentID
+  #clock: LogicalClock
+
+  constructor(identid: IdentID, clock: LogicalClock) {
+    super(`realm-actions-${identid}`)
+
+    this.#identid = identid
+    this.#clock = clock
+
+    this.version(1).stores({
+      actions: '&action.clk, [identid+action.clk]',
+      clocks: '&identid',
+    })
+  }
+
+  async recordActions(actions: RealmAction[]) {
+    const [actionrows, clocks] = actions.reduce(
+      ([actions, clocks], action) => {
+        const {identid} = explode(action.clk)
+        if (compare(clocks[identid], action.clk) < 0) {
+          clocks[identid] = action.clk
+        }
+
+        return [[...actions, {identid, action}], clocks]
+      },
+      [[] as StoredAction[], {} as PeerClocks],
+    )
+
+    let addedkeys: Timestamp[] | undefined
+    await this.transaction('rw', ['actions', 'clocks'], async () => {
+      addedkeys = await this.actions.bulkAdd(actionrows, {allKeys: true})
+
+      for (const [_identid, clock] of Object.entries(clocks)) {
+        if (clock === null) continue
+
+        const identid = IdentBrand.parse(_identid)
+        const extant = await this.clocks.get(identid)
+
+        if (!extant) {
+          await this.clocks.add({identid, clock})
+        } else if (compare(clock, extant.clock) > 0) {
+          await this.clocks.update(identid, {clock})
+        }
+      }
+    })
+
+    if (addedkeys) {
+      const keys = addedkeys
+      const added = actions.filter((a) => keys.includes(a.clk)).sort((a, b) => compare(a.clk, b.clk))
+
+      added.forEach((action) => {
+        this.#events.dispatchCustomEvent('action', action)
+      })
+    }
+  }
+
+  async actionsSince(timestamp: Timestamp): Promise<RealmAction[]> {
+    const rows = await this.actions.where('action.clk').above(timestamp).toArray()
+    return rows.map((a) => a.action)
+  }
+
+  async buildSyncState(): Promise<PeerClocks> {
+    const stored = await this.clocks.toArray()
+    return {
+      ...Object.fromEntries(stored.map((c) => [c.identid, c.clock])),
+      [this.#identid]: this.#clock.latest(),
+    }
+  }
+
+  async buildSyncDelta(clocks: PeerClocks): Promise<RealmAction[]> {
+    const state = await this.buildSyncState()
+    const results = await this.#buildSyncQuery(state, clocks).sortBy('action.clk')
+    return results.map((row) => row.action)
+  }
+
+  #buildSyncQuery(state: PeerClocks, clocks: PeerClocks): Collection<StoredAction, Timestamp> {
+    // in the initial case (they have no peers), just send everything
+    if (!Object.keys(clocks).length) {
+      return this.actions.toCollection()
+    }
+
+    // otherwise, build up a sync state from range queries
+    const peerids = Object.keys(state) as IdentID[]
+    const memoQuery = this.actions.where('action.clk').noneOf([])
+    return peerids.reduce((query, identid) => {
+      const lowerbound = clocks[identid] ?? generate(identid, 0, 0)
+      const upperbound = state[identid]
+      return query.or('[identid+action.clk]').between(
+        [identid, lowerbound],
+        [identid, upperbound],
+        clocks[identid] == null, // include the lower bound if they don't have it
+        true, // always include the upper bound
+      )
+    }, memoQuery)
+  }
+}

src/realm/client/connection.ts

···
-import {RealmID} from '#realm/schema/brands'
+import {type WebSocket as ISOWS} from 'isomorphic-ws'
+import {SignJWT} from 'jose'
+import {nanoid} from 'nanoid'
+import {z} from 'zod/v4'
+
+import {TimeoutSignal, combineSignals, controllerWithSignals, timeoutSignal} from '#lib/async/aborts'
+import {BlockingQueue} from '#lib/async/blocking-queue'
+import {Fence} from '#lib/async/fence'
+import {backoff, sleep} from '#lib/async/sleep'
+import {
+  DataChannelEventMap,
+  DataChannelPeer,
+  RTCSignalData,
+  RTCSignalPayload,
+  rtcSignalPayloadSchema,
+} from '#lib/client/webrtc'
+import {generateSignableJwt, jwkExport, jwkImport} from '#lib/crypto/jwks'
+import {jwtPayload, verifyJwtToken} from '#lib/crypto/jwts'
+import {ProtocolError, normalizeError, normalizeProtocolError} from '#lib/errors'
+import {TypedEventTarget} from '#lib/events'
+import {jsonCodec} from '#lib/schema'
+import {streamSocket, takeSocketSchema} from '#lib/socket'
+import {StrictMap} from '#lib/strict-map'
+
+import {
+  PreauthAuthnRequest,
+  PreauthExchangeInviteRequest,
+  PreauthRegisterRequest,
+  preauthResSchema,
+} from '#realm/protocol/messages-preauth'
+import {
+  RealmPingEvent,
+  realmPeerJoinedEventSchema,
+  realmPeerLeftEventSchema,
+  realmPingEventSchema,
+  realmSignalEventSchema,
+} from '#realm/protocol/messages-realm'
+import {RealmAction, RealmAddress} from '#realm/schema'
+import {IdentBrand, IdentID} from '#realm/schema/brands'
 
-import {RealmIdentity} from './identity'
+import {ActionStore} from './action-store'
 
-export class RealmConnection {
+const realmPeerMessageSchema = z.union([realmPingEventSchema])
+const realmSocketMessageSchema = z.union([
+  realmSignalEventSchema,
+  realmPingEventSchema,
+  realmPeerJoinedEventSchema,
+  realmPeerLeftEventSchema,
+])
+const jsonSocketMessageSchema = jsonCodec(realmSocketMessageSchema)
+
+type RealmSocketMessage = z.infer<typeof realmSocketMessageSchema>
+
+export type RealmRemoteEventMap = {
+  wsopen: CustomEvent<RealmAddress>
+  wsclose: CustomEvent<never>
+  wserror: CustomEvent<ProtocolError>
+  wsdata: CustomEvent<unknown>
+
+  peerjoined: CustomEvent<{identid: IdentID}>
+  peerleft: CustomEvent<{identid: IdentID}>
+}
+
+type RealmPeerEventMap = DataChannelEventMap & {
+  peeropen: CustomEvent<{identid: IdentID}>
+  peerclose: CustomEvent<{identid: IdentID}>
+  peererror: CustomEvent<{identid: IdentID; error: Error}>
+  peerdata: CustomEvent<{identid: IdentID; data: string | ArrayBuffer}>
+}
+
+const SOCKET_HIGHWATER = 1024 * 1024 // 1Mb
+
+/**
+ * a realm connection is the persistent connection to a signalling realm server and a collection of peers
+ *
+ * @summary
+ * * initial connection
+ *   - `new RealmConnection(address)` starts the re/connection loop
+ *   - `#connectLoop` manages the socket connection (with backoff on failures)
+ *     - `#connectedSocket` creates the websocket, and sets up event handlers, blocking until the socket closes
+ *     - `#socketOnOpen` fires, and starts authentication handshake
+ *     - authentication succeeds:
+ *       - peer connections started based on peers from authentication response
+ *         - socket carries signals via signed JWT
+ *         - `#peerMessageLoop` starts, handles peer messages, (dispatching upstream, except pings)
+ *         - `#peerPingLoop` starts, handles pings to the peer (every 30s)
+ *       - `#socketMessageLoop` starts, taking messages off the socket and into a queue
+ *         - processes `pings`, and dispatches others up to event listeners
+ *       - `#socketPingLoop` starts, pinging the server with sync requests every 30s
+ * * disconnect/reconnect
+ *   - socket closes (for whatever reason)
+ *     - `#connectedSocket` promise (see `#connectLoop`) resolves
+ *     - `#socketOnClose()`, closes the `#ready` gate
+ *       - socket loops exit (`while (#ready.value)`)
+ *     - `#connectLoop` does backoff delay
+ *     - loops back and create fresh socket and socket-bound abort signal
+ *
+ * peers should survive socket reconnects, since they're separate connections.
+ */
+export class RealmConnection extends TypedEventTarget<RealmRemoteEventMap & RealmPeerEventMap> {
+  #abort: AbortController
+  #ready = new Fence()
+
+  #address: RealmAddress
+  #identid: IdentID
+  #keypair: CryptoKeyPair
+  #actions: ActionStore
+
+  #identities = new StrictMap<IdentID, CryptoKey>()
+  #peers = new StrictMap<IdentID, RealmPeer>()
+  #peerConfig: Partial<RTCConfiguration> | undefined
+
+  #socket: WebSocket | null = null
+  #socketQueue = new BlockingQueue<string>()
+  #socketTimeout: TimeoutSignal | null = null
+  #socketAbort: AbortController | null = null
+
   constructor(
-    public readonly upstream: string,
-    public readonly realmid: RealmID,
-    public readonly identity: RealmIdentity,
-  ) {}
+    address: RealmAddress,
+    identid: IdentID,
+    keypair: CryptoKeyPair,
+    actions: ActionStore,
+    config?: Partial<RTCConfiguration>,
+    signal?: AbortSignal,
+  ) {
+    super()
+    this.#abort = controllerWithSignals(signal)
+
+    this.#address = address
+    this.#identid = identid
+    this.#keypair = keypair
+    this.#actions = actions
+
+    this.#peerConfig = config
+
+    this.#connectLoop().catch((exc: unknown) => {
+      console.error('fatal error in connection loop', exc)
+      this.dispatchCustomEvent('error', normalizeError(exc))
+    })
+  }
+
+  get address() {
+    return this.#address
+  }
+
+  get connected() {
+    return this.#ready.value
+  }
+
+  async #connectLoop() {
+    const delay = backoff({baseDelay: 1000, maxDelay: 30_000})
+    while (!this.#abort.signal.aborted) {
+      try {
+        await this.#connectedSocket(this.#abort.signal)
+
+        console.log('reconnecting', delay.attempts, 'after', delay.delay)
+        await delay(this.#abort.signal)
+      } catch (exc: unknown) {
+        console.error('unexpected error in connect loop, continuing...', exc)
+        this.dispatchCustomEvent('error', normalizeError(exc))
+        await delay(this.#abort.signal)
+      }
+    }
+  }
+
+  async #connectedSocket(signal?: AbortSignal) {
+    const {promise, resolve, reject} = Promise.withResolvers<void>()
+
+    this.#socketAbort?.abort('reconnecting')
+    this.#socketTimeout = timeoutSignal(60_000)
+    this.#socketAbort = controllerWithSignals(signal, this.#socketTimeout.signal)
+
+    try {
+      const opts = {signal: this.#socketAbort.signal, once: true}
+      this.#socket = new WebSocket(this.#address.upstream)
+
+      this.#socket.addEventListener(
+        'open',
+        () => {
+          this.#socketTimeout?.cancel()
+          this.#socketOnOpen().catch((exc: unknown) => {
+            reject(normalizeError(exc))
+          })
+        },
+        opts,
+      )
+
+      this.#socket.addEventListener(
+        'error',
+        () => {
+          this.#socketTimeout?.cancel()
+          reject(new Error('websocket error'))
+        },
+        opts,
+      )
+
+      this.#socket.addEventListener(
+        'close',
+        () => {
+          resolve()
+        },
+        opts,
+      )
+
+      this.#socketAbort.signal.addEventListener('abort', () => {
+        this.#socketTimeout?.cancel()
+        reject(normalizeError(this.#socketAbort?.signal.reason))
+      })
+
+      await promise
+      this.#socketOnClose()
+    } catch (exc: unknown) {
+      this.#socketTimeout.cancel()
+      this.#socketOnError()
+      throw exc
+    }
+  }
+
+  broadcast<T>(payload: T) {
+    if (!this.#ready.value) return
+
+    const json = JSON.stringify(payload)
+    const recipients = Array.of(...this.#peers.keys())
+
+    recipients.forEach((identid) => {
+      const peer = this.#peers.get(identid)
+      peer?.send(json)
+    })
+
+    // broadcast through to the server so that it can store actions
+    this.#socketQueue.enqueue(json)
+  }
+
+  destroy() {
+    // shut down any ongoing loops
+    if (!this.#abort.signal.aborted) {
+      this.#abort.abort('shutting down')
+    }
+
+    // close connections
+    for (const peer of this.#peers.values()) {
+      peer.destroy()
+    }
+
+    if (this.#socket?.readyState !== WebSocket.CLOSED) {
+      this.#socket?.close()
+    }
+
+    // empty state
+    this.#peers.clear()
+  }
+
+  chooseSyncPeer(): IdentID | null {
+    let max: IdentID | null = null
+    for (const key of this.#peers.keys()) {
+      if (key === this.#identid) continue
+      if (max === null || max.localeCompare(key) < 0) {
+        max = key
+      }
+    }
+
+    return max
+  }
+
+  ///
+
+  #socketOnOpen = async () => {
+    this.dispatchCustomEvent('wsopen', this.#address)
+
+    // drain loop is special, it doesn't wait for ready
+    this.#socketDrainLoop().catch((exc: unknown) => {
+      console.error('uncaught error in socket drain loop', exc)
+    })
+
+    try {
+      const authresp = await this.#socketAuthenticate()
+      if (authresp.typ === 'err') {
+        this.#ready.close()
+        throw new ProtocolError(authresp.err.detail ?? 'unknown error', authresp.err.code)
+      }
+
+      this.#address = {
+        realmid: this.#address.realmid,
+        upstream: this.#address.upstream,
+        // invitation/register intentionally omitted
+      }
+
+      // explicitly, we don't unlock authenticated until we've setup peers
+      // otherwise various loops will start before connections are ready
+
+      // store identities for message verification
+      for (const [identid_, pubkey_] of Object.entries(authresp.dat.identities)) {
+        try {
+          const identid = IdentBrand.parse(identid_)
+          const pubkey = await jwkImport.parseAsync(pubkey_)
+          this.#identities.set(identid, pubkey)
+        } catch (exc: unknown) {
+          console.warn(`couldn't import remote identity: ${identid_}`, exc)
+        }
+      }
+
+      // initiate outbound connections when starting up
+      for (const peerid of authresp.dat.peers) {
+        if (peerid === this.#identid) continue
+        this.#peerConnect(peerid, true)
+      }
+
+      // we're ready!
+      this.#ready.open()
+
+      // start the loops
+      this.#socketMessageLoop().catch((exc: unknown) => {
+        console.error('uncaught error in message loop', exc)
+      })
+
+      this.#socketPingLoop().catch((exc: unknown) => {
+        console.error('uncaught error in ping loop', exc)
+      })
+    } catch (exc: unknown) {
+      console.warn('unknown error in socket outhentication or setup', exc)
+      this.#socket?.close()
+      this.dispatchCustomEvent('error', normalizeError(exc))
+      throw exc
+    }
+  }
+
+  #socketOnClose = () => {
+    this.#ready.close()
+    this.dispatchCustomEvent('wsclose')
+  }
+
+  #socketOnError = () => {
+    this.dispatchCustomEvent('wserror', normalizeProtocolError('dunno', 400))
+  }
+
+  ///
+
+  #socketPingLoop = async () => {
+    await this.#ready.enter(this.#socketAbort?.signal)
+
+    while (this.#ready.value) {
+      await this.#socketPing()
+      await sleep(30_000, this.#socketAbort?.signal)
+    }
+  }
 
-  destroy() {}
+  #socketMessageLoop = async () => {
+    await this.#ready.enter(this.#socketAbort?.signal)
+
+    const stream = streamSocket(this.#socket as unknown as ISOWS, {signal: this.#socketAbort?.signal})
+    try {
+      for await (const raw of stream) {
+        await this.#socketOnData(raw)
+      }
+    } catch (exc: unknown) {
+      console.error('socket stream error', exc)
+      this.dispatchCustomEvent('error', normalizeError(exc))
+    }
+  }
+
+  #socketDrainLoop = async () => {
+    if (this.#socket == null) {
+      throw new Error('unexpectedly ready with no socket?')
+    }
+
+    try {
+      let data
+      while ((data = await this.#socketQueue.dequeue(this.#socketAbort?.signal))) {
+        while (this.#socket.bufferedAmount > SOCKET_HIGHWATER) {
+          await sleep(100, this.#socketAbort?.signal)
+        }
+
+        // channel closed while waiting?
+        if (this.#socket.readyState !== WebSocket.OPEN) {
+          this.#socketQueue.prequeue(data)
+          break
+        }
+
+        try {
+          this.#socket.send(data)
+        } catch (exc: unknown) {
+          this.#socketQueue.prequeue(data)
+          this.dispatchCustomEvent('error', normalizeError(exc))
+          continue
+        }
+      }
+    } catch (exc: unknown) {
+      if (this.#socketAbort?.signal.aborted) return
+      throw exc
+    }
+  }
+
+  #socketOnData = async (data: unknown) => {
+    const parsed = await jsonSocketMessageSchema.safeParseAsync(data)
+    if (parsed.success) {
+      await this.#socketOnMessage(parsed.data)
+    } else {
+      this.dispatchCustomEvent('wsdata', data)
+    }
+  }
+
+  ///
+
+  #socketAuthenticate = async () => {
+    let token: SignJWT
+
+    // send the appropriate type of request
+    if (this.#address.register) {
+      token = await this.#buildAuthenticateRegister()
+    } else if (this.#address.invitation) {
+      token = await this.#buildAuthenticateExchange(this.#address.invitation)
+    } else {
+      token = this.#buildAuthenticateAuthn()
+    }
+
+    const signed = await token.sign(this.#keypair.privateKey)
+    this.#socketQueue.enqueue(signed)
+
+    // now, we're expecting an authenticated responese
+    const timeout = timeoutSignal(5000)
+    const signal = combineSignals(this.#socketAbort?.signal, timeout.signal)
+    return await takeSocketSchema(this.#socket as unknown as ISOWS, jsonCodec(preauthResSchema), signal)
+  }
+
+  #buildAuthenticateAuthn = () => {
+    return generateSignableJwt<PreauthAuthnRequest>({
+      iss: this.#identid,
+      aud: this.#address.realmid,
+      typ: 'req',
+      msg: 'preauth.authn',
+      seq: nanoid(),
+    })
+  }
+
+  #buildAuthenticateRegister = async () => {
+    const pubkey = await jwkExport.parseAsync(this.#keypair.publicKey)
+    return generateSignableJwt<PreauthRegisterRequest>({
+      iss: this.#identid,
+      aud: this.#address.realmid,
+      typ: 'req',
+      msg: 'preauth.register',
+      seq: nanoid(),
+      dat: {pubkey},
+    })
+  }
+
+  #buildAuthenticateExchange = async (inviteJwt: string) => {
+    const pubkey = await jwkExport.parseAsync(this.#keypair.publicKey)
+    return generateSignableJwt<PreauthExchangeInviteRequest>({
+      iss: this.#identid,
+      aud: this.#address.realmid,
+      typ: 'req',
+      msg: 'preauth.exchange',
+      seq: nanoid(),
+      dat: {pubkey, inviteJwt},
+    })
+  }
+
+  #socketPing = async () => {
+    const clocks = await this.#actions.buildSyncState()
+    const ping: RealmPingEvent = {
+      typ: 'evt',
+      msg: 'realm.ping',
+      dat: {clocks, requestSync: true},
+    }
+
+    this.#socketQueue.enqueue(JSON.stringify(ping))
+  }
+
+  // anything we don't handle directly at the realm level gets handled by the server peer
+  #socketOnMessage = async (data: RealmSocketMessage) => {
+    switch (data.msg) {
+      case 'realm.peer-joined':
+        this.#identities.set(data.dat.identid, await jwkImport.parseAsync(data.dat.pubkey))
+        this.dispatchCustomEvent('peerjoined', {identid: data.dat.identid})
+        return
+
+      case 'realm.peer-left':
+        this.#peerDisconnect(data.dat.identid)
+        this.dispatchCustomEvent('peerleft', {identid: data.dat.identid})
+        return
+
+      case 'realm.ping': {
+        if (data.dat.requestSync) {
+          this.#socketSendActions(await this.#actions.buildSyncDelta(data.dat.clocks))
+        }
+
+        return
+      }
+
+      case 'realm.signal': {
+        // find the referenced peer and hand off the signal to it
+        if (data.dat.remoteid !== this.#identid) {
+          console.warn('got a socket signal for a different identity?')
+          return
+        }
+
+        const pubkey = this.#identities.get(data.dat.localid)
+        if (pubkey == null) {
+          console.warn('signal from an unknown identity (we cannot verify!)')
+          return
+        }
+
+        const token = await jwtPayload(rtcSignalPayloadSchema).parseAsync(data.dat.signed)
+        await verifyJwtToken(data.dat.signed, pubkey)
+
+        let peer = this.#peers.get(data.dat.localid)
+        if (!peer && token.payload.initiator) {
+          // false because _we_ are not the initiator, they are (they did initial outbound connections)
+          peer = this.#peerConnect(data.dat.localid, false)
+        }
+
+        // might not have a connection yet (if we're waiting for an answer)
+        peer?.signal(token.payload.signal)
+        return
+      }
+    }
+  }
+
+  #socketSendActions(actions: RealmAction[], batch_size = 100) {
+    for (let i = 0; i < actions.length; i += batch_size) {
+      const batch = actions.slice(i, i + batch_size)
+      this.#socketQueue.enqueue(JSON.stringify(batch))
+    }
+  }
+
+  ///
+
+  #peerConnect(remoteid: IdentID, initiator: boolean): RealmPeer {
+    return this.#peers.ensure(remoteid, () => {
+      // important: we use the _connection_ signal, not the socket signal
+      // the websocket dying and getting reconnected shouldn't effect webrtc peers
+      const timeout = timeoutSignal(60_000)
+      const signal = combineSignals(timeout.signal, this.#abort.signal)
+      const opts = {signal}
+
+      try {
+        const peer = new RealmPeer(this, this.#actions, remoteid, initiator, this.#peerConfig, signal)
+        this.#abort.signal.addEventListener('abort', () => {
+          timeout.cancel()
+        })
+
+        peer.addEventListener(
+          'signal',
+          (e) => {
+            this.#peerOnSignal(peer, e)
+          },
+          opts,
+        )
+
+        peer.addEventListener(
+          'connect',
+          () => {
+            timeout.cancel()
+            this.dispatchCustomEvent('peeropen', {identid: remoteid})
+          },
+          opts,
+        )
+
+        peer.addEventListener(
+          'peerdata',
+          (e) => {
+            this.dispatchCustomEvent('peerdata', e.detail)
+          },
+          opts,
+        )
+
+        peer.addEventListener(
+          'close',
+          () => {
+            timeout.cancel()
+            this.dispatchCustomEvent('peerclose', {identid: remoteid})
+          },
+          opts,
+        )
+
+        peer.addEventListener(
+          'error',
+          (e) => {
+            timeout.cancel()
+            this.dispatchCustomEvent('peererror', {
+              identid: remoteid,
+              error: normalizeProtocolError(e.detail),
+            })
+          },
+          opts,
+        )
+
+        return peer
+      } catch (exc) {
+        timeout.cancel()
+        throw exc
+      }
+    })
+  }
+
+  #peerDisconnect(remoteid: IdentID) {
+    const peer = this.#peers.remove(remoteid)
+    if (peer) {
+      peer.destroy()
+    }
+  }
+
+  #peerOnSignal = (peer: RealmPeer, event: CustomEvent<RTCSignalData>) => {
+    const go = async () => {
+      const token = generateSignableJwt<RTCSignalPayload>({
+        aud: peer.identid,
+        iss: this.#identid,
+        exp: Math.floor(Date.now() / 1000) + 60,
+        signal: event.detail,
+        initiator: peer.initiator,
+      })
+
+      const signed = await token.sign(this.#keypair.privateKey)
+      this.#socketQueue.enqueue(signed)
+    }
+
+    go().catch((exc: unknown) => {
+      console.error('error during peer signal send!', exc)
+      this.dispatchCustomEvent('error', normalizeError(exc))
+    })
+  }
+}
+
+export class RealmPeer extends DataChannelPeer<RealmPeerEventMap> {
+  #abort: AbortController
+  #connected: Fence
+  #queue: BlockingQueue<string>
+
+  constructor(
+    readonly realm: RealmConnection,
+    readonly actions: ActionStore,
+    readonly identid: IdentID,
+    readonly initiator: boolean,
+    config?: Partial<RTCConfiguration>,
+    signal?: AbortSignal,
+  ) {
+    super(initiator, {
+      iceServers: [{urls: 'stun:stun.l.google.com:19302'}, {urls: 'stun:stun1.l.google.com:19302'}],
+      iceCandidatePoolSize: 0, // no pre-gather, trickle ico
+      bundlePolicy: 'balanced',
+      iceTransportPolicy: 'all',
+      ...(config || {}),
+    })
+
+    this.#abort = controllerWithSignals(signal)
+    this.#connected = new Fence()
+    this.#queue = new BlockingQueue()
+
+    const opts = {signal: this.#abort.signal}
+    this.addEventListener('connect', this.#peerConnected, opts)
+    this.addEventListener('close', this.#peerClosed, opts)
+    this.addEventListener('data', this.#peerData, opts)
+
+    this.#pingLoop().catch((exc: unknown) => {
+      console.warn('unexpected error in ping loop', exc)
+    })
+
+    this.#receiveLoop().catch((exc: unknown) => {
+      console.warn('unexpected error in receive loop', exc)
+    })
+  }
+
+  destroy(error?: Error) {
+    this.#abort.abort(error)
+    super.destroy()
+  }
+
+  ///
+
+  #peerConnected = () => {
+    this.#connected.open()
+  }
+
+  #peerClosed = () => {
+    this.#connected.close()
+  }
+
+  #peerData = (event: CustomEvent<string | ArrayBuffer>) => {
+    const data = event.detail
+    const string = typeof data === 'string' ? data : new TextDecoder().decode(data)
+    this.#queue.enqueue(string)
+  }
+
+  ///
+
+  #receiveLoop = async () => {
+    this.#abort.signal.throwIfAborted()
+
+    while (await this.#connected.enter(this.#abort.signal)) {
+      const payload = await this.#queue.dequeue(this.#abort.signal)
+      if (!this.#connected.value) break
+
+      const parsed = await jsonCodec(realmPeerMessageSchema).safeParseAsync(payload)
+      if (parsed.success) {
+        await this.#receivePing(parsed.data)
+      } else {
+        this.dispatchCustomEvent('peerdata', {identid: this.identid, data: payload})
+      }
+    }
+  }
+
+  #pingLoop = async () => {
+    while (await this.#connected.enter(this.#abort.signal)) {
+      // wait a moment to let the connect settle before ping
+      await sleep(100, this.#abort.signal)
+      if (!this.#connected.value) {
+        break
+      }
+
+      await this.#sendPing()
+      await sleep(30_000, this.#abort.signal)
+    }
+  }
+
+  #sendPing = async () => {
+    if (!this.#connected.value) return
+
+    const clocks = await this.actions.buildSyncState()
+    const syncPeer = this.realm.chooseSyncPeer()
+
+    const message: RealmPingEvent = {
+      typ: 'evt',
+      msg: 'realm.ping',
+      dat: {clocks, requestSync: syncPeer === this.identid},
+    }
+
+    this.send(JSON.stringify(message))
+  }
+
+  #receivePing = async (ping: RealmPingEvent) => {
+    if (!ping.dat.requestSync) return
+
+    // TODO: record last received ping
+
+    this.#sendActions(await this.actions.buildSyncDelta(ping.dat.clocks))
+  }
+
+  ///
+
+  #sendActions(actions: RealmAction[], batch_size = 100) {
+    for (let i = 0; i < actions.length; i += batch_size) {
+      const batch = actions.slice(i, i + batch_size)
+      this.send(JSON.stringify(batch))
+    }
+  }
 }

src/realm/client/context.tsx

···
-import {PropsWithChildren, createContext, useCallback, useContext, useEffect, useMemo, useState} from 'react'
-
-import {normalizeError} from '#lib/errors'
-
-import {RealmID} from '#realm/schema/brands'
-
-import {RealmConnection} from './connection'
-import {RealmIdentity} from './identity'
-import {RealmContextStorage} from './storage'
-
-// loading -> local
-//       \     v ^
-//         -> remote
-
-export type RealmState =
-  | {status: 'loading'}
-  | {status: 'local'; identity: RealmIdentity}
-  | {status: 'remote'; identity: RealmIdentity; remote: RealmRemote; connection?: RealmConnection}
-  | {status: 'error'; identity: RealmIdentity | null; error: Error}
-
-export type RealmRemote = {
-  realmid: RealmID
-  upstream: string
-  register?: boolean
-  invitation?: string
-}
-
-export const RealmContext = createContext<RealmContext | null>(null)
-export interface RealmContext {
-  state: RealmState
-
-  connect: (opts: RealmRemote) => Promise<void>
-  disconnect: () => Promise<void>
-}
-
-export type RealmContextProviderProps = {
-  url: string
-}
-
-export const RealmContextProvider = (props: PropsWithChildren<RealmContextProviderProps>) => {
-  const store = useMemo(() => new RealmContextStorage(), [])
-  const [state, setState] = useState<RealmState>({status: 'loading'})
-
-  const status = state.status
-  const identity = status !== 'error' && status !== 'loading' ? state.identity : null
-  const remote = status === 'remote' ? state.remote : null
-  const connection = status === 'remote' ? state.connection : null
-
-  const connect = useCallback(
-    async (remote: RealmRemote) => {
-      if (!identity) return
-
-      await store.connect(remote.upstream, remote.realmid)
-      setState({status: 'remote', identity, remote})
-    },
-    [identity, store],
-  )
-
-  const disconnect = useCallback(async () => {
-    if (!identity) return
-
-    await store.disconnect()
-    setState({status: 'local', identity})
-  }, [identity, store])
-
-  // on mount, pull our connection info from storage
-  useEffect(() => {
-    if (status !== 'loading') return
-
-    const go = async () => {
-      const row = await store.fetch()
-      const identity = new RealmIdentity(row.identid, row.keypair, row.clock)
-      const {realmid, upstream} = row
-
-      if (realmid && upstream) {
-        setState({status: 'remote', identity, remote: {realmid, upstream}})
-      } else {
-        setState({status: 'local', identity})
-      }
-    }
-
-    go().catch((exc: unknown) => {
-      console.error(exc)
-      setState({status: 'error', error: normalizeError(exc), identity: null})
-    })
-  }, [status, identity, store])
-
-  // whenever we're in connecting state, connect!
-  useEffect(() => {
-    if (status !== 'remote' || !identity || !remote) return
-    if (connection) return
-
-    const conn = new RealmConnection(remote.upstream, remote.realmid, identity)
-    setState({status: 'remote', identity, remote, connection: conn})
-
-    return () => {
-      conn.destroy()
-      setState({status: 'remote', identity, remote})
-      // TODO: reconnect counter, backoff, etc.
-    }
-  }, [status, identity, remote, connection])
-
-  return <RealmContext.Provider value={{state, connect, disconnect}}>{props.children}</RealmContext.Provider>
-}
-
-export function useRealmContext(): RealmContext {
-  const context = useContext(RealmContext)
-  if (context == null) throw new Error('expected to be under a realm context!')
-
-  return context
-}

src/realm/client/identity-store.ts

···
+import Dexie, {Table} from 'dexie'
+
+import {generateSigningJwkPair} from '#lib/crypto/jwks'
+
+import {RealmAddress} from '#realm/schema'
+import {IdentBrand, IdentID} from '#realm/schema/brands'
+import {Timestamp} from '#realm/schema/timestamp'
+
+import {RealmIdentity} from './identity'
+
+interface StoredIdentity {
+  identid: IdentID
+  keypair: CryptoKeyPair
+  clock?: Timestamp
+  remote?: RealmAddress
+  lastUsed?: number
+}
+
+export class RealmIdentityStore extends Dexie {
+  private identities!: Table<StoredIdentity>
+
+  constructor() {
+    super('realm-identities')
+
+    this.version(1).stores({
+      identities: '&identid, lastUsed',
+    })
+  }
+
+  async list(): Promise<IdentID[]> {
+    const keys = await this.identities.toCollection().keys()
+    return keys.map((k) => IdentBrand.parse(k))
+  }
+
+  async load(identid: IdentID): Promise<RealmIdentity | undefined> {
+    const row = await this.identities.get(identid)
+    return row ? this.#loadIdentity(row) : undefined
+  }
+
+  ensure(): Promise<RealmIdentity> {
+    return this.transaction('rw', [this.identities], async () => {
+      let row = await this.identities.orderBy('lastUsed').reverse().first()
+      if (!row) {
+        const identid = IdentBrand.generate()
+        const keypair = await generateSigningJwkPair()
+
+        row = {identid, keypair, lastUsed: Date.now()}
+        await this.identities.put(row)
+      } else {
+        await this.identities.update(row.identid, {lastUsed: Date.now()})
+      }
+
+      return this.#loadIdentity(row)
+    })
+  }
+
+  #loadIdentity(row: StoredIdentity) {
+    const abort = new AbortController()
+    const identity = new RealmIdentity(row.identid, row.keypair, row.clock)
+    identity.addEventListener('tick', this.#onTick.bind(this, identity), {signal: abort.signal})
+    identity.addEventListener('connected', this.#onConnected.bind(this, identity), {signal: abort.signal})
+
+    // this will abort when the identity destroys itself, clearing event listeners
+    identity.addEventListener('destroyed', abort.abort.bind(abort), {signal: abort.signal})
+
+    if (row.remote) {
+      identity.connect(row.remote)
+    }
+
+    return identity
+  }
+
+  #onConnected = (identity: RealmIdentity, event: CustomEvent<RealmAddress>) => {
+    this.identities
+      .update(identity.identid, {
+        'remote.realmid': event.detail.realmid,
+        'remote.upstream': event.detail.upstream,
+        lastUsed: Date.now(),
+      })
+      .catch((exc: unknown) => {
+        console.error('uncaught error in identity connected handler', exc)
+      })
+  }
+
+  #onTick = (identity: RealmIdentity, event: CustomEvent<Timestamp>) => {
+    this.identities.update(identity.identid, {clock: event.detail}).catch((exc: unknown) => {
+      console.error('uncaught error in identity tick handler', exc)
+    })
+  }
+}

src/realm/client/identity.ts

···
+import {z} from 'zod/v4'
+
+import {controllerWithSignals} from '#lib/async/aborts'
+import {jwtSchema} from '#lib/crypto/jwts'
+import {TypedEventTarget} from '#lib/events'
+import {jsonCodec} from '#lib/schema'
+
 import {LogicalClock} from '#realm/logical-clock'
-import {IdentID} from '#realm/schema/brands'
+import {RealmAction, RealmAddress, actionSchema} from '#realm/schema'
+import {IdentID, RealmBrand} from '#realm/schema/brands'
 import {Timestamp} from '#realm/schema/timestamp'
 
-export class RealmIdentity {
+import {ActionStore} from './action-store'
+import {RealmConnection} from './connection'
+
+export type RealmIdentityEventMap = {
+  connected: CustomEvent<RealmAddress>
+  disconnected: CustomEvent<never>
+  destroyed: CustomEvent<never>
+
+  tick: CustomEvent<Timestamp>
+  action: CustomEvent<RealmAction>
+
+  peerjoined: CustomEvent<{identid: IdentID}>
+  peerleft: CustomEvent<{identid: IdentID}>
+}
+
+const realmIdentityMessageSchema = z.union([
+  z.array(actionSchema),
+  // other messages (for now) are handled by connection directly
+])
+
+export class RealmIdentity extends TypedEventTarget<RealmIdentityEventMap> {
   readonly identid: IdentID
   readonly keypair: CryptoKeyPair
   readonly clock: LogicalClock
+  readonly actions: ActionStore
+
+  #abort = new AbortController()
 
-  constructor(identid: IdentID, keypair: CryptoKeyPair, timestamp?: Timestamp) {
+  #connection?: RealmConnection
+  #connectionAbort?: AbortController
+
+  constructor(identid: IdentID, keypair: CryptoKeyPair, latest?: Timestamp) {
+    super()
+
     this.identid = identid
     this.keypair = keypair
-    this.clock = new LogicalClock(this.identid, timestamp ?? null)
+
+    this.clock = new LogicalClock(identid, latest)
+    this.clock.addEventListener('tick', this.#onTick, {signal: this.#abort.signal})
+
+    this.actions = new ActionStore(identid, this.clock)
+    this.actions.addEventListener('action', this.#onAction, {signal: this.#abort.signal})
+  }
+
+  async dispatchAction<
+    A extends RealmAction,
+    K extends A['msg'] = A['msg'],
+    T extends Extract<A, {msg: K}>['dat'] = A['dat'],
+  >(msg: K, dat: T) {
+    const action: RealmAction<T> = {typ: 'act', clk: this.clock.now(), msg, dat}
+
+    await this.actions.recordActions([action])
+    this.#connection?.broadcast([action])
+  }
+
+  #onTick = (e: CustomEvent<Timestamp>) => {
+    this.dispatchCustomEvent('tick', e.detail)
+  }
+
+  #onAction = (e: CustomEvent<RealmAction>) => {
+    this.dispatchCustomEvent('action', e.detail)
+  }
+
+  connect(remote: RealmAddress) {
+    this.disconnect()
+
+    this.#connectionAbort = controllerWithSignals(this.#abort.signal)
+    const opts = {signal: this.#connectionAbort.signal}
+    const dispatch = this.dispatchEvent.bind(this) as EventListener
+
+    this.#connection = new RealmConnection(
+      remote,
+      this.identid,
+      this.keypair,
+      this.actions,
+      {},
+      this.#abort.signal,
+    )
+    this.#connection.addEventListener('wsopen', this.#onWSOpen, opts)
+    this.#connection.addEventListener('wsclose', this.#onWSClose, opts)
+    this.#connection.addEventListener('peerjoined', dispatch, opts)
+    this.#connection.addEventListener('peerleft', dispatch, opts)
+    this.#connection.addEventListener('peerdata', this.#onPeerData, opts)
+    this.#connection.addEventListener('wsdata', this.#onWSData, opts)
+  }
+
+  register(upstream: string): void {
+    const realmid = RealmBrand.generate()
+    this.connect({upstream, realmid, register: true})
+  }
+
+  exchangeInvite(upstream: string, invitation: string) {
+    const token = jwtSchema.parse(invitation)
+    const realmid = RealmBrand.parse(token.claims.aud)
+
+    const now = Math.floor(Date.now() / 1000)
+    if (!token.claims.exp || token.claims.exp < now) throw new Error('invitation has expired')
+
+    this.connect({realmid, upstream, invitation})
+  }
+
+  get remote(): RealmAddress | undefined {
+    return this.#connection?.address
+  }
+
+  get connected(): boolean {
+    return this.#connection != null && this.#connection.connected
+  }
+
+  #onWSOpen = (e: CustomEvent<RealmAddress>) => {
+    this.dispatchCustomEvent('connected', e.detail)
+  }
+
+  #onWSClose = () => {
+    this.dispatchCustomEvent('disconnected')
+  }
+
+  #onWSData = (e: CustomEvent<unknown>) => {
+    this.#onIncomingData(e.detail)
+  }
+
+  #onPeerData = (e: CustomEvent<{identid: IdentID; data: unknown}>) => {
+    this.#onIncomingData(e.detail.data)
+  }
+
+  #onIncomingData = (data: unknown) => {
+    const parsed = jsonCodec(realmIdentityMessageSchema).safeParse(data)
+    if (parsed.success) {
+      this.actions.recordActions(parsed.data).catch((exc: unknown) => {
+        console.error('error on identity action recording', exc, parsed.data)
+      })
+    } else {
+      console.warn('unknown message in identity:', data, parsed.error)
+    }
+  }
+
+  disconnect() {
+    this.#connection?.destroy()
+    this.#connectionAbort?.abort('shutting down')
+
+    this.#connection = undefined
+    this.#connectionAbort = undefined
+  }
+
+  destroy() {
+    this.disconnect()
+    this.#abort.abort('shutting down')
   }
 }

src/realm/client/storage.ts

···
-import Dexie, {Table} from 'dexie'
-
-import {generateSigningJwkPair} from '#lib/crypto/jwks.js'
-
-import {IdentBrand, IdentID, RealmID} from '#realm/schema/brands'
-import {Timestamp} from '#realm/schema/timestamp'
-
-export interface RealmContextRow {
-  identid: IdentID
-  keypair: CryptoKeyPair
-  clock?: Timestamp
-  realmid?: RealmID
-  upstream?: string
-}
-
-class ContextDB extends Dexie {
-  context!: Table<RealmContextRow>
-
-  constructor() {
-    super('realm-context')
-    this.version(1).stores({
-      context: '&identid',
-    })
-  }
-}
-
-export class RealmContextStorage {
-  #db: ContextDB
-
-  constructor() {
-    this.#db = new ContextDB()
-  }
-
-  async fetch(): Promise<RealmContextRow> {
-    const extant = await this.#db.context.toCollection().first()
-    if (extant) return extant
-
-    const identid = IdentBrand.generate()
-    const keypair = await generateSigningJwkPair()
-
-    const row = {identid, keypair}
-    await this.#db.context.put(row)
-
-    return row
-  }
-
-  async connect(upstream: string, realmid: RealmID) {
-    const extant = await this.fetch() // make sure there's an existing row
-    await this.#db.context.update(extant.identid, {upstream, realmid})
-  }
-
-  async disconnect() {
-    const extant = await this.fetch() // make sure there's an existing row
-    await this.#db.context.update(extant.identid, {upstream: undefined, realmid: undefined})
-  }
-}

src/realm/logical-clock.ts

···
+import {TypedEventTarget} from '#lib/events'
+
 import {IdentID} from '#realm/schema/brands'
 import {Timestamp, compare, generate} from '#realm/schema/timestamp'
+
+export type LogicalClockEventMap = {
+  tick: CustomEvent<Timestamp>
+}
 
 /**
  * A logical clock that generates monotonically increasing timestamps, with identity as tiebreaking.
···
  * clock.tick(receivedTimestamp)
  * ```
  */
-export class LogicalClock extends EventTarget {
+export class LogicalClock extends TypedEventTarget<LogicalClockEventMap> {
   #identid: IdentID
   #counter = 0
   #seconds = 0
···
   tick(timestamp: Timestamp) {
     if (!this.#latest || compare(this.#latest, timestamp) < 0) {
       this.#latest = timestamp
-      this.dispatchEvent(new TickEvent(LogicalClock.TICK, timestamp))
+      this.dispatchCustomEvent('tick', timestamp)
     }
 
     return this.#latest
   }
 }
-
-/**
- * An event emmitted when the logical clock advances.
- * @see {@link LogicalClock.tick}
- * @event
- */
-export class TickEvent extends CustomEvent<Timestamp> {
-  constructor(name: string, timestamp: Timestamp) {
-    super(name, {detail: timestamp})
-  }
-}

src/realm/protocol/messages-preauth.ts

···
 import {jwkSchema} from '#lib/crypto/jwks'
 import {makeEmptyRequestSchema, makeErrorSchema, makeRequestSchema, makeResponseSchema} from '#lib/schema'
 
-import {peerIdSchema} from '#realm/schema/brands'
+import {IdentBrand} from '#realm/schema/brands'
 
 /** preauth message: given a public key, request a new realm */
 export const preauthRegisterReqSchema = makeRequestSchema('preauth.register', z.object({pubkey: jwkSchema}))
···
 export const preauthSuccessSchema = makeResponseSchema(
   'preauth.authn',
   z.object({
-    peers: z.array(peerIdSchema),
+    peers: z.array(IdentBrand.schema),
     identities: z.record(z.string(), jwkSchema),
   }),
 )

src/realm/protocol/messages-realm.ts

···
   }),
 )
 
-export const realmSignalEventPayloadSchema = z.object({
-  signal: z.string(),
-  initiator: z.boolean(),
-})
-
 export const realmPeerJoinedEventSchema = makeEventSchema(
   'realm.peer-joined',
   z.object({
···
 )
 
 export type RealmSignalEvent = z.infer<typeof realmSignalEventSchema>
-export type RealmSignalEventPayload = z.infer<typeof realmSignalEventPayloadSchema>
-
 export type RealmBroadcastEvent = z.infer<typeof realmBroadcastEventSchema>
 export type RealmPeerJoinedEvent = z.infer<typeof realmPeerJoinedEventSchema>
 export type RealmPeerLeftEvent = z.infer<typeof realmPeerLeftEventSchema>

src/realm/protocol/messages-rtc.ts

···
+import {z} from 'zod/v4'
+
+export const rtcSignalDataSchema = z.discriminatedUnion('type', [
+  z.object({type: z.literal('offer'), sdp: z.any()}),
+  z.object({type: z.literal('answer'), sdp: z.any()}),
+  z.object({type: z.literal('candidate'), candidate: z.any()}),
+])
+
+export const rtcSignalPayloadSchema = z.object({
+  signal: rtcSignalDataSchema,
+  initiator: z.boolean(),
+})
+
+export type RTCSignalData = z.infer<typeof rtcSignalDataSchema>
+export type RTCSignalPayload = z.infer<typeof rtcSignalPayloadSchema>

src/realm/schema/brands.ts

···
-import {z} from 'zod/v4'
-
 import {Brand} from '#lib/schema/brand'
 
 const ident = Symbol('ident')
···
 const realm = Symbol('realm')
 export const RealmBrand = new Brand<typeof realm>(realm, 'rlm')
 export type RealmID = ReturnType<typeof RealmBrand.generate>
-
-export const serverPeerIdSchema = z.literal('server')
-export type ServerPeerID = z.infer<typeof serverPeerIdSchema>
-
-export const peerIdSchema = z.union([IdentBrand.schema, serverPeerIdSchema])
-export type PeerID = z.infer<typeof peerIdSchema>

src/realm/schema/index.ts

···
 import {z} from 'zod/v4'
 
+import {RealmBrand} from './brands'
 import {logicalClockSchema} from './timestamp'
 
 export const actionSchema = z.object({
···
     dat: schema,
   })
 }
+
+export const realmAddressSchema = z.object({
+  realmid: RealmBrand.schema,
+  upstream: z.url(),
+  invitation: z.string().optional(),
+  register: z.boolean().optional(),
+})
+
+export type RealmAddress = z.infer<typeof realmAddressSchema>

src/realm/schema/timestamp.ts

···
  * compare(ts1, ts2) // Returns negative (ts1 happened before ts2)
  * ```
  */
-export function compare(a?: Timestamp, b?: Timestamp): number {
-  if (a === undefined && b == undefined) return 0
-  if (a === undefined) return 1
-  if (b === undefined) return -1
+export function compare(a?: Timestamp | null, b?: Timestamp | null): number {
+  if (a == null && b == null) return 0
+  if (a == null) return 1
+  if (b == null) return -1
 
   const [_a, asec, acounter, aident] = a.split(':')
   const [_b, bsec, bcounter, bident] = b.split(':')

src/realm/server/driver-preauth.spec.ts

···
         typ: 'res',
         msg: 'preauth.authn',
         dat: {
-          peers: ['server'],
+          peers: [],
           identities: {
             [identid]: pubkeyJwk,
           },
···
         typ: 'res',
         msg: 'preauth.authn',
         dat: expect.objectContaining({
-          peers: expect.arrayContaining(['server']),
+          peers: expect.arrayContaining([]),
           identities: expect.objectContaining({
             [identid]: pubkeyJwk,
           }),
···
         typ: 'res',
         msg: 'preauth.authn',
         dat: expect.objectContaining({
-          peers: expect.arrayContaining(['server']),
+          peers: expect.arrayContaining([]),
           identities: expect.objectContaining({
             [inviterIdentid]: expect.anything(),
           }),

src/realm/server/driver-preauth.ts

···
         msg: 'preauth.authn',
         seq: jwt.payload.seq,
         dat: {
-          peers: [...realm.peers, 'server'],
+          peers: realm.peers,
           identities: realm.knownIdentities(),
         },
       })

src/realm/server/driver-realm.ts

···
 
   // and also send sync actions (server is always a sync partner)
   if (req.dat.requestSync) {
-    const actions = realm.buildSyncDelta(clocks)
+    const actions = realm.buildSyncDelta(req.dat.clocks)
     if (actions.length) {
-        putSocket(ws, actions)
+      putSocket(ws, actions)
     }
   }
 }

src/realm/server/realm.ts

···
    * @returns array of actions the peer has not seen
    */
   buildSyncDelta(clocks: PeerClocks): RealmAction[] {
-    const entries = Object.entries(clocks)
     const schema = jsonCodec(actionSchema)
 
+    // a null clock is the same as no clock
+    const entries = Object.entries(clocks).filter(([_, v]) => !!v)
     if (entries.length === 0) {
       const stmt = this.#database.prepare(`SELECT action FROM action ORDER BY clock ASC`)
       const rows = stmt.all()