modules/state-crdts/hyper-p2p-crdt-rga-text/index.js

require('bare-process/global')
const EventEmitter = require('bare-events')
const { assertNonEmpty } = require('../../_shared/lib/errors.js')
const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')
const { crdtStats } = require('../../_shared/crdt-base.js')

const PROTOCOL = 'crdt-rga-text/v1'

class HyperP2PCrdtRgaText extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.topic = opts.topic || null
    this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
    this._nodes = new Map()
    this._order = []
    this._stats = { ops: 0, gossipIn: 0, gossipOut: 0 }
    this.swarm = null
    this._peerMsgs = null
  }

  insert (index, charId, char) {
    return this._applyInsert(index, charId, char, true)
  }

  delete (charId) {
    return this._applyDelete(charId, true)
  }

  _applyInsert (index, charId, char, gossip) {
    if (index < 0) throw new Error('index must be non-negative')
    assertNonEmpty(charId, 'charId')
    if (char == null || char === '') throw new Error('char required')
    const id = String(charId)
    if (!this._nodes.has(id)) {
      this._nodes.set(id, { char: String(char), deleted: false })
      if (!this._order.includes(id)) this._order.push(id)
    } else if (this._nodes.get(id).deleted) {
      this._nodes.get(id).deleted = false
      this._nodes.get(id).char = String(char)
    }
    const visible = this._visibleIds()
    const curIdx = visible.indexOf(id)
    if (curIdx >= 0) visible.splice(curIdx, 1)
    const at = Math.min(index, visible.length)
    visible.splice(at, 0, id)
    this._order = [...visible, ...this._order.filter((x) => !visible.includes(x))]
    this._stats.ops++
    if (gossip) {
      gossipSend(this, { type: 'crdt-rga-text-sync', op: 'insert', charId: id, char: String(char), index: at })
      this._stats.gossipOut++
    }
    return id
  }

  _applyDelete (charId, gossip) {
    assertNonEmpty(charId, 'charId')
    const id = String(charId)
    const node = this._nodes.get(id)
    if (!node || node.deleted) return false
    node.deleted = true
    this._stats.ops++
    if (gossip) {
      gossipSend(this, { type: 'crdt-rga-text-sync', op: 'delete', charId: id })
      this._stats.gossipOut++
    }
    return true
  }

  toString () {
    return this._visibleIds().map((id) => this._nodes.get(id).char).join('')
  }

  length () {
    return this.toString().length
  }

  charAt (index) {
    const ids = this._visibleIds()
    const id = ids[index]
    return id ? this._nodes.get(id).char : ''
  }

  visibleIds () {
    return this._visibleIds()
  }

  merge (remote) {
    const ops = Array.isArray(remote) ? remote : (remote?.ops || [])
    let n = 0
    for (const op of ops) {
      if (!op || !op.op) continue
      if (op.op === 'insert') {
        this._applyInsert(op.index || 0, op.charId, op.char, false)
        n++
      }
      if (op.op === 'delete' && this._applyDelete(op.charId, false)) n++
    }
    return n
  }

  snapshot () {
    return {
      nodes: Object.fromEntries([...this._nodes].map(([id, n]) => [id, { ...n }])),
      order: [...this._order]
    }
  }

  insertBatch (ops) {
    if (!Array.isArray(ops)) throw new Error('ops array required')
    return ops.map((op) => this.insert(op.index, op.charId, op.char))
  }

  clearAll () {
    const n = this._nodes.size
    this._nodes.clear()
    this._order = []
    return n
  }

  _visibleIds () {
    return this._order.filter((id) => {
      const n = this._nodes.get(id)
      return n && !n.deleted
    })
  }

  _onGossip (d) {
    if (!d || d.type !== 'crdt-rga-text-sync' || !d.op) return
    this._stats.gossipIn++
    if (d.op === 'insert') this._applyInsert(d.index || 0, d.charId, d.char, false)
    if (d.op === 'delete') this._applyDelete(d.charId, false)
  }

  getStats () {
    return crdtStats(this._stats, PROTOCOL, { length: this.length() })
  }

  async ready () {
    if (this.swarm || !this.topic) return this
    await initModuleSwarm(this, {
      keyPair: this.keyPair,
      topic: this.topic,
      protocol: PROTOCOL,
      onmessage: (d) => this._onGossip(d)
    })
    return this
  }

  async close () {
    if (this.swarm) await this.swarm.destroy().catch(() => {})
    this.swarm = null
    this._peerMsgs = null
  }
}

module.exports = { HyperP2PCrdtRgaText, PROTOCOL }