y-crdt.js

import P2PMesh from "./p2p-mesh.js"

import uuidv4 from "uuid/v4"
import yArray from "y-array"
import yMap from "y-map"
import yUnion, { Union } from "./y-union.js"
import yMemory from "y-memory"
import Y from "yjs"

yMemory(Y)
Y.Struct.Union = Union
yUnion(Y)
yMap(Y)
yArray(Y)

function gc(obj) {
  const objs = new Set()

  const free = (obj) => {
    if (obj == null) return

    for (const key of Object.keys(obj)) {
      if (typeof obj[key] == "object") {
        if (!objs.has(obj[key])) {
          objs.add(obj[key])
          free(obj[key])
        }
      }

      delete obj[key]
    }
  }

  free(obj)

  objs.clear()
}

export default class YjsCRDTWrapper extends Y.AbstractConnector {
  constructor(y, options) {
    if (options === undefined) {
      throw new Error("Options must not be undefined!")
    }

    options.role = "slave"
    super(y, options)

    this.y = y
    this.room = null
    this.mesh = new P2PMesh(this, options)
  }

  _initialise(room) {
    this.room = room

    super.onUserEvent((event) => {
      if (event.action == "userJoined") {
        const { user: id } = event
        this.room.dispatchEvent(new CustomEvent("userJoin", { detail: id }))
      } else if (event.action == "userLeft") {
        const { user: id } = event
        this.room.dispatchEvent(new CustomEvent("userLeave", { detail: id }))
      }
    })

    const dispatchPathUpdateEvent = (lineEvent) => {
      const pathID = lineEvent.name
      const points = this.room.getPathPoints(pathID)

      const detail = { id: pathID, points }

      this.room.dispatchEvent(new CustomEvent("addOrUpdatePath", { detail }))
    }

    const dispatchRemovedIntervalsEvent = (lineEvent) => {
      const pathID = lineEvent.name
      const intervals = this.room.getErasureIntervals(pathID)

      const detail = { id: pathID, intervals }

      this.room.dispatchEvent(
        new CustomEvent("removedIntervalsChange", {
          detail,
        }),
      )
    }

    this.y.share.strokePoints.observe((lineEvent) => {
      if (lineEvent.type == "add") {
        dispatchPathUpdateEvent(lineEvent)

        lineEvent.value.observe((pointEvent) => {
          if (pointEvent.type == "insert") {
            dispatchPathUpdateEvent(lineEvent)
          }
        })
      }
    })

    this.y.share.eraseIntervals.observe((lineEvent) => {
      if (lineEvent.type == "add") {
        dispatchRemovedIntervalsEvent(lineEvent)

        lineEvent.value.observe(() => {
          dispatchRemovedIntervalsEvent(lineEvent)
        })
      }
    })
  }

  destroy() {
    // yjs connectors have an optional destroy() method that is called on y.destroy()
    if (this.mesh == null) return

    this.mesh.disconnect()
    this.mesh = null

    delete this.room

    gc(this)
  }

  static async initialise(room, options) {
    const y = await Y({
      db: {
        name: "memory",
      },
      connector: Object.assign({}, options, { name: "y-crdt" }),
      share: {
        strokePoints: "Map",
        eraseIntervals: "Map",
      },
    })

    y.connector._initialise(room)

    room.crdt = y.connector
  }

  getUserID() {
    return this.y.db.userId
  }

  setUserID(uid) {
    return super.setUserId(uid)
  }

  fetchDrawingEvents() {
    // NOOP: twiddle thumbs
  }

  addPath([x, y, w, colour]) {
    const id = uuidv4()

    this.y.share.strokePoints.set(id, Y.Array).push([[x, y, w, colour]])
    this.y.share.eraseIntervals.set(id, Y.Union)

    return id
  }

  extendPath(pathID, [x, y, w, colour]) {
    const path = this.y.share.strokePoints.get(pathID)

    path.push([[x, y, w, colour]])

    return path.length
  }

  endPath(/*pathID*/) {
    // NOOP: twiddle thumbs
  }

  extendErasureIntervals(pathID, newIntervals) {
    this.y.share.eraseIntervals.get(pathID).merge(newIntervals)
  }

  getPathIDs() {
    return this.y.share.strokePoints.keys()
  }

  getPathPoints(pathID) {
    const points = this.y.share.strokePoints.get(pathID)

    if (!points) return []

    return points.toArray()
  }

  getErasureIntervals(pathID) {
    const intervals = this.y.share.eraseIntervals.get(pathID)

    if (!intervals) return []

    return intervals.get()
  }

  userJoined(uid) {
    super.userJoined(uid, "master")
  }

  userLeft(uid) {
    super.userLeft(uid)
  }

  receiveMessage(uid, message) {
    super.receiveMessage(uid, message)

    this.room.dispatchEvent(
      new CustomEvent("peerSyncedWithUs", { detail: uid }),
    )
  }

  reportConnectionQuality(uid, quality) {
    this.room.dispatchEvent(
      new CustomEvent("userConnection", { detail: { id: uid, quality } }),
    )
  }

  disconnect() {
    super.disconnect()
  }

  reconnect() {
    throw "Unsupported operation reconnect()"
  }

  send(uid, message) {
    let compressed = true

    if (message.type === "sync step 1") {
      compressed = false

      this.room.dispatchEvent(
        new CustomEvent("waitingForSyncStep", { detail: uid }),
      )
    } else if (message.type === "sync done") {
      compressed = false

      this.room.dispatchEvent(
        new CustomEvent("weSyncedWithPeer", { detail: uid }),
      )
    } else if (message.type === "sync check") {
      compressed = false
    }

    this.mesh.send(uid, message, compressed)
  }

  broadcast(message) {
    this.mesh.broadcast(message, true)
  }

  isDisconnected() {
    return false
  }
}

Y.extend("y-crdt", YjsCRDTWrapper)

export const syncStep1 = {
  uuid: "6e20b20d-e1d8-405d-8a61-d56cb1c47a24",
  message: Uint8Array.of(
    133,
    164,
    116,
    121,
    112,
    101,
    171,
    115,
    121,
    110,
    99,
    32,
    115,
    116,
    101,
    112,
    32,
    49,
    168,
    115,
    116,
    97,
    116,
    101,
    83,
    101,
    116,
    128,
    169,
    100,
    101,
    108,
    101,
    116,
    101,
    83,
    101,
    116,
    128,
    175,
    112,
    114,
    111,
    116,
    111,
    99,
    111,
    108,
    86,
    101,
    114,
    115,
    105,
    111,
    110,
    11,
    164,
    97,
    117,
    116,
    104,
    192,
  ),
  slice: 0,
  length: 1,
  compressed: false,
}

export const syncDone = {
  message: Uint8Array.of(
    129,
    164,
    116,
    121,
    112,
    101,
    169,
    115,
    121,
    110,
    99,
    32,
    100,
    111,
    110,
    101,
  ),
  slice: 0,
  length: 1,
  compressed: false,
}