Function handler

packages/experimental/src/EventLogServer.ts:41–135

(socket: Socket.Socket)

  let chunkId = 0

  function* handler(socket: Socket.Socket) {
    const subscriptions = yield* FiberMap.make<string>()
    const writeRaw = yield* socket.writer
    const chunks = new Map<number, {
      readonly parts: Array<Uint8Array>
      count: number
      bytes: number
    }>()
    let latestSequence = -1

    function* writeGen(response: typeof ProtocolResponse.Type) {
      const data = encodeResponse(response)
      if (response._tag !== "Changes" || data.byteLength <= constChunkSize) {
        return yield* writeRaw(data)
      }
      const id = chunkId++
      for (const part of ChunkedMessage.split(id, data)) {
        yield* writeRaw(encodeResponse(part))
      }
    }
    const write = (response: typeof ProtocolResponse.Type) => Effect.gen(() => writeGen(response))

    yield* Effect.fork(write(new Hello({ remoteId })))

    function handleRequest(request: typeof ProtocolRequest.Type) {
      switch (request._tag) {
        case "Ping": {
          return write(new Pong({ id: request.id }))
        }
        case "WriteEntries": {
          if (request.encryptedEntries.length === 0) {
            return write(
              new Ack({
                id: request.id,
                sequenceNumbers: []
              })
            )
          }
          return Effect.gen(function*() {
            const entries = request.encryptedEntries.map(({ encryptedEntry, entryId }) =>
              new PersistedEntry({
                entryId,
                iv: request.iv,
                encryptedEntry
              })
            )
            const encrypted = yield* storage.write(request.publicKey, entries)
            latestSequence = encrypted[encrypted.length - 1].sequence
            return yield* write(
              new Ack({
                id: request.id,
                sequenceNumbers: encrypted.map((e) => e.sequence)
              })
            )
          })
        }
        case "RequestChanges": {
          return Effect.gen(function*() {