add message framing to webrtc (max size is easy to hit)

src/feedline/client/action-dispatcher.ts

···
 import {FeedAddAction} from '#feedline/schema/actions'
 
 import {FeedlineDatabase} from './database'
-import {ReqEvent} from './worker-api'
 
 export class FeedlineActionDispatcher {
   #abort: AbortController
   #identity: RealmIdentity
-  #worker: Worker
 
   constructor(_database: FeedlineDatabase, identity: RealmIdentity, worker: Worker, signal?: AbortSignal) {
     this.#abort = controllerWithSignals(signal)
     this.#identity = identity
-    this.#worker = worker
   }
 
   destroy() {
···
         reason: 'feed-add',
       },
     })
-
-    this.#worker.postMessage({
-      typ: 'evt',
-      msg: 'feed:fetch',
-      dat: {urls: [url]},
-    } satisfies ReqEvent)
   }
 }

src/feedline/client/action-handler.ts

···
   async #feedAdd(action: FeedAddAction) {
     await this.#db.feeds.add({
       url: action.dat.url,
-      lock: action.dat.lock,
+      lock: action.dat.lock ?? null,
       subscription: {
         health: 100,
         interval: 'daily',
···
     await this.#db.feeds.update(action.dat.url, action.dat.payload)
   }
 
-  async #entryPatch(action: EntryPatchAction) {
+  #entryPatch(action: EntryPatchAction) {
     // this should check locks and clocks
-    const key: [string, string] = [action.dat.feedUrl, action.dat.guid]
-    console.log('entry patch:', key, action.dat.payload)
+    return this.#db
+      .transaction('rw', ['feeds', 'entries'], async (tx) => {
+        const feed = await tx.feeds.get(action.dat.feedurl)
+        if (!feed) throw new Error('fkey constraint error, expected feedurl to exist')
 
-    await this.#db.entries.upsert(key, {
-      tags: [],
-      links: [],
-      meta: {},
-      ...action.dat.payload,
-      guid: action.dat.guid,
-      feedUrl: action.dat.feedUrl,
-    })
+        const key: [string, string] = [action.dat.feedurl, action.dat.guid]
+        const extant = await tx.entries.get(key)
+        if (extant) {
+          await tx.entries.update(key, action.dat.payload)
+        } else {
+          await tx.entries.add({
+            tags: [],
+            links: [],
+            meta: {},
+            lock: null,
+            // defaults ^^
+            ...action.dat.payload,
+            // overrides vv
+            guid: action.dat.guid,
+            feedurl: action.dat.feedurl,
+          })
+        }
+      })
+      .catch((exc: unknown) => {
+        console.error('error entry patching?', exc)
+        throw exc
+      })
   }
 }

src/feedline/client/database.ts

···
     super(`feedline-${identid}`)
 
     this.version(1).stores({
-      // feeds indexed by URL (primary key), tags, lock state, status, medium
       feeds: '&url, *tags.tag, lock.at, lock.by, subscription.status, medium',
-
-      // entries with compound key [guid, feedUrl], indexed by feedUrl, tags, lock, timestamps
-      entries: '&[feedUrl+guid], feedUrl, *tags.tag, lock.at, lock.by, publishedAt, fetchedAt',
+      entries: '&[feedurl+guid], *tags.tag, lock.at, lock.by, publishedAt, fetchedAt',
     })
   }
 

src/feedline/client/feed-fetcher.ts

···
-import {IndexableType} from 'dexie'
+import {IndexableType, liveQuery, Subscription} from 'dexie'
 import {parseFeed as parseFeedsmith} from 'feedsmith'
 import {Atom, Rss} from 'feedsmith/types'
 
···
   #owner: IdentID
   #apihost: string
 
+  #subscription: Subscription
+
   constructor(identid: IdentID, apihost: string, signal?: AbortSignal) {
     super()
 
···
     this.#owner = identid
     this.#apihost = apihost
 
-    this.#refreshLoop().catch((exc: unknown) => {
-      if (this.#abort.signal.aborted) return
+    const query = liveQuery(async () => await this.#db.feeds.where('subscription.status').equals('pending').primaryKeys())
+    this.#subscription = query.subscribe((urls) => {
+      console.log('pendingFeeds', urls)
+
+      const feeds = this.#db.feeds.where('url').anyOf(urls)
+      this.#db.withLock('feeds', feeds, this.#clock, this.#owner, this.#processLocked).catch((exc: unknown) => {
+        if (this.#abort.signal.aborted) return
 
-      const error = normalizeError(exc)
-      console.error('unexpected error out of refresh loop', error.name, error.message)
-      this.dispatchCustomEvent('error', error)
+        const error = normalizeError(exc)
+        console.error('unexpected error out of refresh loop', error.name, error.message)
+        this.dispatchCustomEvent('error', error)
+      })
     })
   }
 
   destroy() {
     this.#abort.abort(new Error('fetcher shutting down'))
-  }
-
-  async #refreshLoop() {
-    do {
-      try {
-        await this.#processPending()
-        await sleep(10_000, this.#abort.signal)
-      } catch (exc: unknown) {
-        if (this.#abort.signal.aborted) return
-
-        const error = normalizeError(exc)
-        console.error('unexpected error out of refresh loop', error.name, error.message)
-        this.dispatchCustomEvent('error', error)
-      }
-    } while (!this.#abort.signal.aborted)
+    this.#subscription.unsubscribe()
   }
 
   async processUrls(urls: string[]) {
···
     return this.#db.withLock('feeds', processFeeds, this.#clock, this.#owner, this.#processLocked)
   }
 
-  #processPending() {
-    const pendingFeeds = this.#db.feeds.where('subscription.status').equals('pending')
-    console.log('pendingFeeds', pendingFeeds)
-    return this.#db.withLock('feeds', pendingFeeds, this.#clock, this.#owner, this.#processLocked)
-  }
-
   async #fetchFeedProxy(feed: Feed): Promise<[Response, boolean]> {
     const proxyUrl = `${this.#apihost}/api/cors?url=${encodeURI(feed.url)}`
     return [await fetch(proxyUrl), true]
···
           const [parsedFeed, parsedEntries] = parseFeed(feed.url, content)
 
           // diff against existing entries - only sync new or updated entries
-          const existingEntries = await this.#db.entries.where('feedUrl').equals(feed.url).toArray()
+          const existingEntries = await this.#db.entries.where('feedurl').equals(feed.url).toArray()
           const existingMap = new Map(existingEntries.map((e) => [e.guid, e]))
 
           const newOrUpdatedEntries = parsedEntries.filter((parsed) => {
···
   }
 
   #postFeedPatch(
-    feedUrl: string,
+    feedurl: string,
     changes: Partial<Feed>,
     entries: Entry[],
     subscription: NonNullable<Feed['subscription']>,
···
       clk: this.#clock.now(),
       msg: 'feed:patch',
       dat: {
-        url: feedUrl,
+        url: feedurl,
         payload: {
           ...changes,
           lock: null,
···
       clk: this.#clock.now(),
       msg: 'entry:patch',
       dat: {
-        feedUrl,
+        feedurl,
         guid: entry.guid,
         payload: entry,
       },
···
   }
 }
 
-function parseFeed(feedUrl: string, content: string): [Partial<Feed>, Entry[]] {
+function parseFeed(feedurl: string, content: string): [Partial<Feed>, Entry[]] {
   const result = parseFeedsmith(content)
   switch (result.format) {
     case 'rss':
-      return mapRssToSchema(feedUrl, result.feed as Rss.Feed<string>)
+      return mapRssToSchema(feedurl, result.feed as Rss.Feed<string>)
     case 'atom':
-      return mapAtomToSchema(feedUrl, result.feed as Atom.Feed<string>)
+      return mapAtomToSchema(feedurl, result.feed as Atom.Feed<string>)
     default:
       throw new Error('unsupported feed type')
   }
···
   }
 }
 
-function mapRssToEntry(feedUrl: string, item: Rss.Item<string>, fetchedAt: number): Entry {
+function mapRssToEntry(feedurl: string, item: Rss.Item<string>, fetchedAt: number): Entry {
   const entryTags: Tag[] = []
   const entryLinks: Link[] = []
 
···
     : undefined
 
   return {
-    feedUrl,
-    guid: item.guid?.value ?? item.link ?? `${feedUrl}#${item.title}`,
+    feedurl,
+    guid: item.guid?.value ?? item.link ?? `${feedurl}#${item.title}`,
     lock: null,
     tags: entryTags,
     title: item.title ?? null,
···
   }
 }
 
-function mapRssToSchema(feedUrl: string, rss: Rss.Feed<string>): [Partial<Feed>, Entry[]] {
+function mapRssToSchema(feedurl: string, rss: Rss.Feed<string>): [Partial<Feed>, Entry[]] {
   const feed = mapRssToFeed(rss)
   const now = Date.now()
-  const entries = rss.items == null ? [] : rss.items.map((item) => mapRssToEntry(feedUrl, item, now))
+  const entries = rss.items == null ? [] : rss.items.map((item) => mapRssToEntry(feedurl, item, now))
 
   return [feed, entries]
 }
···
   }
 }
 
-function mapAtomToEntry(feedUrl: string, entry: Atom.Entry<string>, fetchedAt: number): Entry {
+function mapAtomToEntry(feedurl: string, entry: Atom.Entry<string>, fetchedAt: number): Entry {
   const entryTags: Tag[] = []
   const entryLinks: Link[] = []
 
···
     : undefined
 
   return {
-    feedUrl,
+    feedurl,
     guid: entry.id,
     lock: null,
     tags: entryTags,
···
   }
 }
 
-function mapAtomToSchema(feedUrl: string, atom: Atom.Feed<string>): [Partial<Feed>, Entry[]] {
+function mapAtomToSchema(feedurl: string, atom: Atom.Feed<string>): [Partial<Feed>, Entry[]] {
   const feed = mapAtomToFeed(atom)
   const now = Date.now()
-  const entries = atom.entries?.map((entry) => mapAtomToEntry(feedUrl, entry, now)) ?? []
+  const entries = atom.entries?.map((entry) => mapAtomToEntry(feedurl, entry, now)) ?? []
 
   return [feed, entries]
 }

src/feedline/schema/actions-entry.ts

···
 export const entryPatchSchema = makeActionSchema(
   'entry:patch',
   z.object({
-    guid: z.string(), // Entry GUID from feed
-    feedUrl: z.url(), // Which feed this entry belongs to
+    guid: z.string(), // entry GUID from feed
+    feedurl: z.url(), // which feed this entry belongs to
     payload: entrySchema.partial(),
   }),
 )

src/feedline/schema/entry.ts

···
 import {lockSchema} from './lock'
 
 export const entrySchema = z.object({
-  feedUrl: z.url(),
+  feedurl: z.url(),
   guid: z.string(),
   lock: lockSchema.nullable(),
 

src/lib/client/webrtc.ts

···
 import {Fence} from '#lib/async/fence'
 import {normalizeError} from '#lib/errors'
 import {TypedEventTarget} from '#lib/events'
+import { sleep } from '#lib/async/sleep.js'
 
 export const rtcSessionDescriptionSchema = z.object({
   sdp: z.string().optional(),
···
 export type RTCSignalPayload = z.infer<typeof rtcSignalPayloadSchema>
 
 export type PeerConnectionStats = z.infer<typeof peerConnectionStatsSchema>
-export type DataChannelSendable = string | Blob | ArrayBuffer | ArrayBufferView<ArrayBuffer>
+export type DataChannelSendable = string
 
 export type DataChannelEventMap = {
   signal: CustomEvent<RTCSignalData>
   connect: CustomEvent<never>
   close: CustomEvent<never>
   error: CustomEvent<Error>
-  data: CustomEvent<string | ArrayBuffer>
+  data: CustomEvent<string>
 }
 
+type DataChunk = { buffer: Uint8Array<ArrayBuffer>, compressed: boolean }
+
 const BUFFER_LOWWATER = 512 * 1024 // ~512KB
 
+const CHUNK_SIZE = 32 * 1024
+const COMPRESSION_THRESHOLD = 8 * 1024
+const MESSAGE_MAX_SIZE = 100 * 1024 * 1024 // 100MB
+
+const ChunkFlags = {
+  START: 0x01,
+  COMPRESSED: 0x02,
+  FINAL: 0x04,
+}
+
 export class DataChannelPeer<
   T extends DataChannelEventMap = DataChannelEventMap,
 > extends TypedEventTarget<T> {
   readonly initiator: boolean
+  #abort: AbortController
 
   #peer: RTCPeerConnection
+  #polite: boolean
+  #candidates: RTCIceCandidateInit[] = []
   #chan: RTCDataChannel | null = null
-  #candidates: RTCIceCandidateInit[] = []
+  #queue = new BlockingQueue<DataChannelSendable>()
 
-  #abort: AbortController
-  #queue = new BlockingQueue<DataChannelSendable>()
-  #polite: boolean
+  #chunkIncoming: (DataChunk & { offset: number }) | null = null
+  #chunkBuffer = new BlockingQueue<DataChunk>()
 
   constructor(initiator: boolean, config?: RTCConfiguration, signal?: AbortSignal) {
     super()
···
   #setupDataChannel() {
     if (!this.#chan) return
 
+    this.#chan.binaryType = "arraybuffer"
+
     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,
-    )
+    this.#chan.addEventListener('message', (e: MessageEvent<ArrayBuffer>) => this.#incomingChunk(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))
     })
+
+    // incoming loop blocks on buffer
+    void this.#incomingLoop().catch((exc: unknown) => {
+      console.error('unexpected error in peer incoming loop!', exc)
+      this.dispatchCustomEvent('error', normalizeError(exc))
+    })
   }
+
+  ///
 
   signal(data: unknown) {
     const parsed = rtcSignalDataSchema.safeParse(data)
···
     }
   }
 
+  ///
+
   send(data: DataChannelSendable): void {
     this.#queue.enqueue(data)
   }
···
 
     try {
       let data: DataChannelSendable
-      while ((data = await this.#queue.dequeue(this.#abort.signal))) {
-        await lowwater.enter(this.#abort.signal) // wait for peer buffer capacity
+      send: while ((data = await this.#queue.dequeue(this.#abort.signal))) {
+        let index = 0
+        const chunks = this.#chunkMessage(data as string)
+        for await (const chunk of chunks) {
+          await lowwater.enter(this.#abort.signal) // wait for peer buffer capacity
+          console.log('sending chunk', index++)
 
-        // channel closed while waiting?
-        if (chan.readyState !== 'open') {
-          this.#queue.prequeue(data)
-          break
+          // channel closed while waiting?
+          if (chan.readyState !== 'open') {
+            // TODO: the whole chunk is messed up, what do do here?
+            this.#queue.prequeue(data)
+            break send
+          }
+
+          try {
+            chan.send(chunk)
+          } catch (exc: unknown) {
+            // TODO: the whole chunk is messed up, what do do here?
+            this.#queue.prequeue(data)
+            this.dispatchCustomEvent('error', normalizeError(exc))
+            await sleep(100) // don't spin hard
+            break // break the chunk loop to restart the message
+          } finally {
+            console.log('chan bufferedAmount:', chan.bufferedAmount)
+            lowwater.value = chan.bufferedAmount <= BUFFER_LOWWATER
+          }
         }
+      }
+    } catch (exc: unknown) {
+      if (this.#abort.signal.aborted) return
+      throw exc
+    }
+  }
+
+  async *#chunkMessage(message: string): AsyncGenerator<ArrayBuffer> {
+    const encoded = new TextEncoder().encode(message)
+
+    let payload = encoded
+    let isCompressed = false
+
+    // maybe compress
+    if (encoded.length > COMPRESSION_THRESHOLD) {
+      const stream     = new Blob([encoded]).stream().pipeThrough(new CompressionStream("gzip"))
+      const compressed = new Uint8Array(await new Response(stream).arrayBuffer())
+
+      if (compressed.length < encoded.length) {
+        isCompressed = true
+        payload = compressed
+      }
+    }
+
+    // chunk it up
+    let offset = 0
+    let isFirst = true
+
+    const totalLength = payload.length
+    while (offset < totalLength) {
+      const headerSize = isFirst ? 5 : 1
+      const maxPayload = CHUNK_SIZE - headerSize
+      const chunkSize  = Math.min(maxPayload, totalLength - offset)
+      const isFinal    = offset + chunkSize >= totalLength
+
+      const chunk = new ArrayBuffer(headerSize + chunkSize)
+      const bytes = new Uint8Array(chunk)
+      const view  = new DataView(chunk)
+
+      // byte 0 = flags
+      view.setUint8(
+        0,
+        (isFinal ? ChunkFlags.FINAL : 0) |
+          (isCompressed ? ChunkFlags.COMPRESSED : 0) |
+          (isFirst ? ChunkFlags.START : 0),
+      )
+
+      // bytes 1-4 = length (first only)
+      if (isFirst) {
+        view.setUint32(1, totalLength)
+      }
+
+      // rest of a slice of the payload
+      bytes.set(payload.subarray(offset, offset + chunkSize), headerSize)
+
+      // then yield this slice and we'll start over
+      yield chunk
+
+      offset += chunkSize
+      isFirst = false
+    }
+  }
+
+  ///
+
+  #incomingChunk(data: ArrayBuffer) {
+    const chunk = new Uint8Array(data)
+    const view  = new DataView(chunk.buffer)
+
+    const flags = chunk[0]
+    const isFinal = (flags & ChunkFlags.FINAL) !== 0
+    const isCompressed = (flags & ChunkFlags.COMPRESSED) !== 0
+    const isStart = (flags & ChunkFlags.START) !== 0
+
+    console.log('received chunk', this.#chunkIncoming?.offset || 0, isFinal, isCompressed)
+
+    if (isStart) {
+      if (this.#chunkIncoming) {
+        console.warn('received start chunk while read in progress, discarding previous')
+        this.#chunkIncoming = null
+      }
+
+      // first chunk, includes total length
+      const totalLength = view.getUint32(1)
+      if (totalLength > MESSAGE_MAX_SIZE) {
+        console.error('message too large', totalLength)
+        return
+      }
+
+      this.#chunkIncoming = {
+        buffer: new Uint8Array(totalLength),
+        offset: 0,
+        compressed: isCompressed
+      }
+
+      this.#chunkIncoming.buffer.set(chunk.subarray(5), 0)
+      this.#chunkIncoming.offset = chunk.length - 5 // header length
+    }
+    else if (this.#chunkIncoming) {
+      // continuation chunk
+      this.#chunkIncoming.buffer.set(chunk.subarray(1), this.#chunkIncoming.offset)
+      this.#chunkIncoming.offset += chunk.length - 1 // header length
+    }
+    else {
+      console.warn('received orphan chunk, ignoring')
+      return
+    }
+
+    if (isFinal && this.#chunkIncoming) {
+      this.#chunkBuffer.enqueue(this.#chunkIncoming)
+      this.#chunkIncoming = null
+    }
+  }
+
+  async #incomingLoop() {
+    try {
+      let chunk: DataChunk
+      while ((chunk = await this.#chunkBuffer.dequeue(this.#abort.signal))) {
+        let payload = chunk.buffer
 
         try {
-          chan.send(data as string)
-          // 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
+          if (chunk.compressed) {
+            const stream = new Blob([payload]).stream().pipeThrough(new DecompressionStream("gzip"))
+            payload = new Uint8Array(await new Response(stream).arrayBuffer())
+          }
+
+          const message = new TextDecoder().decode(payload)
+
+          console.log('finished chunk', chunk.compressed, chunk.buffer.length, message.length)
+          this.dispatchCustomEvent('data', message)
         } catch (exc: unknown) {
-          this.#queue.prequeue(data)
+          console.error('dropping incoming message due to error:', exc)
           this.dispatchCustomEvent('error', normalizeError(exc))
           continue
         }
-
-        lowwater.value = chan.bufferedAmount <= BUFFER_LOWWATER
       }
     } catch (exc: unknown) {
       if (this.#abort.signal.aborted) return
       throw exc
     }
   }
+
+  ///
 
   destroy(error?: Error) {
     this.#chan?.close()

src/realm/client/connection.ts

···
     }
   }
 
+  batchBroadcast<T>(payload: T[], per = 10) {
+    for (let i = 0; i < payload.length; i += per) {
+      this.broadcast(payload.slice(i, i + per))
+    }
+  }
+
   destroy(error?: Error) {
     error = error || new Error('shutting down')
 
···
     }
   }
 
-  #socketSendActions(actions: RealmAction[], batch_size = 100) {
+  #socketSendActions(actions: RealmAction[], batch_size = 10) {
     for (let i = 0; i < actions.length; i += batch_size) {
       const batch = actions.slice(i, i + batch_size)
       this.#socketQueue.enqueue(JSON.stringify(batch))

src/realm/client/identity.ts

···
     const actions = Array.isArray(action) ? action : [action]
 
     await this.actions.recordActions(actions)
-    this.#connection?.broadcast(actions)
+    this.#connection?.batchBroadcast(actions)
   }
 
   #onTick = (e: CustomEvent<Timestamp>) => {