persist a loro-crdt doc into indexed db

loro-indexeddb.ts

// This file persists Loro doc to indexeddb
import type { Loro, LoroEvent } from 'loro-crdt'
import { Remitter, ReadonlyRemitter } from 'remitter'
import * as idb from 'lib0/indexeddb'

export interface PersistEventData {
  synced: IPersistProvider
}

export interface IPersistProvider {
  readonly db: IDBDatabase | null
  readonly synced: boolean
  readonly emitter: ReadonlyRemitter<PersistEventData>
  readonly whenSynced: Promise<IPersistProvider>
  destroy(): Promise<void>
}

const Updates = 'updates'
const PreferredTrimSize = 500

// Credit: https://github.com/yjs/y-indexeddb
class PersistProvider implements IPersistProvider {
  db: IDBDatabase | null = null
  synced = false
  emitter = new Remitter<PersistEventData>

  _dbref = 0
  _dbsize = 0
  _destroyed = false
  _db: Promise<IDBDatabase>

  _storeTimeout = 1000
  _storeTimeoutId = 0
  _storeUpdate: (event: LoroEvent) => void

  whenSynced: Promise<PersistProvider>
  destroy: () => Promise<void>

  constructor(readonly name: string, readonly doc: Loro) {
    this._db = idb.openDB(name, db => idb.createStores(db, [
      [Updates, { autoIncrement: true }],
    ]))

    this.whenSynced = new Promise(resolve => this.emitter.once('synced', () => resolve(this)))

    this._db.then(db => {
      this.db = db
      fetchUpdates(this,
        (updates$) => {
          idb.addAutoKey(updates$, this.doc.exportSnapshot())
        },
        () => {
          if (this._destroyed) return
          this.synced = true
          this.emitter.emit('synced', this)
        }
      )
    })

    let lastVersion = doc.version()
    this._storeUpdate = (event: LoroEvent) => {
      if (this.db && !event.fromCheckout) {
        const [updates$] = idb.transact(this.db, [Updates])
        idb.addAutoKey(updates$, doc.exportFrom(lastVersion))
        lastVersion = doc.version()
        if (++this._dbsize >= PreferredTrimSize) {
          clearTimeout(this._storeTimeoutId)
          this._storeTimeoutId = setTimeout(() => storeState(this), this._storeTimeout)
        }
      }
    }

    const subscribeId = doc.subscribe(this._storeUpdate)

    this.destroy = () => {
      clearTimeout(this._storeTimeoutId)
      this.doc.unsubscribe(subscribeId)
      return this._db.then(db => db.close())
    }
  }
}

function fetchUpdates(persist: PersistProvider,
                      beforeUpdate = (updates$: IDBObjectStore) => void 0,
                      afterUpdate = (updates$: IDBObjectStore) => void 0): Promise<IDBObjectStore> {
  const [updates$] = idb.transact(persist.db!, [Updates])
  return idb.getAll(updates$, idb.createIDBKeyRangeLowerBound(persist._dbref, false)).then(updates => {
    if (!persist._destroyed) {
      beforeUpdate(updates$)
      // persist.doc.importUpdateBatch(updates)
      updates.forEach(update => persist.doc.import(update))
      afterUpdate(updates$)
    }
  }).then(() => idb.getLastKey(updates$).then(lastKey => { persist._dbref = lastKey + 1 }))
    .then(() => idb.count(updates$).then(count => { persist._dbsize = count }))
    .then(() => updates$)
}

function storeState(persist: PersistProvider): Promise<void> {
  return fetchUpdates(persist).then(updates$ => {
    if (persist._dbsize >= PreferredTrimSize) {
      idb.addAutoKey(updates$, persist.doc.exportSnapshot())
        .then(() => idb.del(updates$, idb.createIDBKeyRangeUpperBound(persist._dbref, true)))
        .then(() => idb.count(updates$).then(cnt => { persist._dbsize = cnt }))
    }
  })
}

/**
 * ```js
 * const db = persist('doc', doc)
 * db.emitter.on('synced', () => {
 *   console.log('loaded content from indexeddb')
 * })
 * ```
 */
export function persist(name: string, doc: Loro): IPersistProvider {
  return new PersistProvider(name, doc)
}