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)

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

    this.y.destroy()
    this.y = null

    this.room = null
  }

  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) {
    if (message.type === "sync step 1") {
      this.room.dispatchEvent(
        new CustomEvent("waitingForSyncStep", { detail: uid }),
      )
    } else if (message.type === "sync done") {
      this.room.dispatchEvent(
        new CustomEvent("weSyncedWithPeer", { detail: uid }),
      )
    }

    this.mesh.send(uid, message)
  }

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

  isDisconnected() {
    return false
  }
}

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

export const syncStep1 = {
  uuid: "6e20b20d-e1d8-405d-8a61-d56cb1c47a24",
  message: Uint8Array.of(
    120,
    156,
    107,
    93,
    82,
    82,
    89,
    144,
    186,
    186,
    184,
    50,
    47,
    89,
    161,
    184,
    36,
    181,
    64,
    193,
    112,
    69,
    113,
    73,
    98,
    73,
    106,
    112,
    106,
    73,
    195,
    202,
    148,
    212,
    156,
    84,
    8,
    115,
    125,
    65,
    81,
    126,
    73,
    126,
    114,
    126,
    78,
    88,
    106,
    81,
    113,
    102,
    126,
    30,
    247,
    146,
    196,
    210,
    146,
    140,
    3,
    0,
    80,
    113,
    26,
    230,
  ),
  slice: 0,
  length: 1,
  compressed: true,
}

export const syncDone = {
  message: Uint8Array.of(
    120,
    156,
    107,
    92,
    82,
    82,
    89,
    144,
    186,
    178,
    184,
    50,
    47,
    89,
    33,
    37,
    63,
    47,
    21,
    0,
    64,
    79,
    7,
    20,
  ),
  slice: 0,
  length: 1,
  compressed: true,
}