Update rave-level to be compatible with abstract-level 2.0 API

get-sync-worker.js

@@ -0,0 +1,45 @@
+'use strict'
+
+const { workerData, parentPort } = require('worker_threads')
+const net = require('net')
+const { ManyLevelGuest } = require('many-level')
+
+const port = workerData.port
+
+parentPort.on('message', async function ({ payload, semaphore }) {
+  try {
+    const value = await get(payload.socketPath, payload.key)
+    port.postMessage({ value })
+  } catch (err) {
+    port.postMessage({
+      error: {
+        code: err && err.code,
+        message: err && err.message,
+        stack: err && err.stack
+      }
+    })
+  } finally {
+    Atomics.store(semaphore, 0, 1)
+    Atomics.notify(semaphore, 0, 1)
+  }
+})
+
+async function get (socketPath, key) {
+  const db = new ManyLevelGuest({
+    keyEncoding: 'buffer',
+    valueEncoding: 'buffer',
+    retry: false,
+    _remote: () => net.connect(socketPath)
+  })
+
+  await db.open()
+
+  try {
+    return await db.get(Buffer.from(key), {
+      keyEncoding: 'buffer',
+      valueEncoding: 'buffer'
+    })
+  } finally {
+    await db.close()
+  }
+}

get-sync.js

@@ -0,0 +1,86 @@
+'use strict'
+
+const ModuleError = require('module-error')
+const { Worker, MessageChannel, receiveMessageOnPort } = require('worker_threads')
+const path = require('path')
+
+const leaders = new Map()
+
+// HERE BE DRAGONS
+//
+// getSync() must be synchronous from abstract-level's perspective, but a follower can only reach the leader through async socket IO.
+// We do that async work in a worker thread, then block this thread with Atomics.wait() until the worker replies.
+// This exists because real LevelDB getSync() blocks too, and some callers may require the sync API surface.
+
+// If you are wondering why God has forsaken us and given us JavaScript: same.
+// This is the pinnacle of cursed code and war crimes. A monstrocity that should never have been born,
+// let alone be considered "useful" for any situation other than inflicting psychological damage.
+exports.createGetSync = function createGetSync (socketPath) {
+  let syncRead
+
+  return function getSync (key, options) {
+    if (options.snapshot !== undefined) {
+      throw new ModuleError('Snapshots are not supported by rave-level getSync() followers', {
+        code: 'LEVEL_NOT_SUPPORTED'
+      })
+    }
+
+    const leader = leaders.get(socketPath)
+
+    if (leader !== undefined) {
+      return leader._getSync(key, options)
+    }
+
+    if (syncRead === undefined) {
+      syncRead = createSyncReadWorker()
+    }
+
+    return syncRead({ socketPath, key })
+  }
+}
+
+exports.registerLeader = function registerLeader (socketPath, db) {
+  leaders.set(socketPath, db)
+
+  return function unregisterLeader () {
+    if (leaders.get(socketPath) === db) {
+      leaders.delete(socketPath)
+    }
+  }
+}
+
+function createSyncReadWorker () {
+  const { port1, port2 } = new MessageChannel()
+  const worker = new Worker(path.join(__dirname, 'get-sync-worker.js'), {
+    workerData: { port: port2 },
+    transferList: [port2]
+  })
+
+  worker.unref()
+  port1.unref()
+
+  return function syncRead (payload) {
+    const semaphore = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT))
+
+    worker.postMessage({ payload, semaphore })
+
+    Atomics.wait(semaphore, 0, 0)
+
+    const message = receiveMessageOnPort(port1).message
+
+    if (message.error) {
+      throw remoteError(message.error)
+    }
+
+    return message.value === undefined ? undefined : Buffer.from(message.value)
+  }
+}
+
+function remoteError (error) {
+  const err = new ModuleError(error.message || 'Could not get value', {
+    code: error.code || 'LEVEL_REMOTE_ERROR'
+  })
+
+  if (error.stack) err.stack = error.stack
+  return err
+}

index.d.ts

@@ -1,4 +1,25 @@
-import { AbstractLevel, AbstractDatabaseOptions } from 'abstract-level'
+import {
+  AbstractLevel,
+  AbstractDatabaseOptions,
+  AbstractOpenOptions,
+  AbstractGetOptions,
+  AbstractGetManyOptions,
+  AbstractHasOptions,
+  AbstractHasManyOptions,
+  AbstractPutOptions,
+  AbstractDelOptions,
+  AbstractBatchOperation,
+  AbstractBatchOptions,
+  AbstractChainedBatch,
+  AbstractChainedBatchWriteOptions,
+  AbstractIteratorOptions,
+  AbstractIterator,
+  AbstractKeyIterator,
+  AbstractKeyIteratorOptions,
+  AbstractValueIterator,
+  AbstractValueIteratorOptions,
+  AbstractSnapshot
+} from 'abstract-level'
 
 /**
  * Use a [LevelDB](https://github.com/google/leveldb) database from multiple processes
@@ -8,7 +29,7 @@ import { AbstractLevel, AbstractDatabaseOptions } from 'abstract-level'
  * @template VDefault The default type of values if not overridden on operations.
  */
 export class RaveLevel<KDefault = string, VDefault = string>
-  extends AbstractLevel<Buffer, KDefault, VDefault> {
+  extends AbstractLevel<string | Buffer | Uint8Array, KDefault, VDefault> {
   /**
    * Database constructor.
    *
@@ -20,12 +41,56 @@ export class RaveLevel<KDefault = string, VDefault = string>
     location: string,
     options?: DatabaseOptions<KDefault, VDefault> | undefined
   )
+
+  /**
+   * Whether this instance is the leader process for the database.
+   */
+  isLeader: boolean
+
+  open (): Promise<void>
+  open (options: OpenOptions): Promise<void>
+
+  get (key: KDefault): Promise<VDefault | undefined>
+  get<K = KDefault, V = VDefault> (key: K, options: GetOptions<K, V>): Promise<V | undefined>
+
+  getSync (key: KDefault): VDefault | undefined
+  getSync<K = KDefault, V = VDefault> (key: K, options: GetOptions<K, V>): V | undefined
+
+  getMany (keys: KDefault[]): Promise<Array<VDefault | undefined>>
+  getMany<K = KDefault, V = VDefault> (keys: K[], options: GetManyOptions<K, V>): Promise<Array<V | undefined>>
+
+  has (key: KDefault): Promise<boolean>
+  has<K = KDefault> (key: K, options: HasOptions<K>): Promise<boolean>
+
+  hasMany (keys: KDefault[]): Promise<boolean[]>
+  hasMany<K = KDefault> (keys: K[], options: HasManyOptions<K>): Promise<boolean[]>
+
+  put (key: KDefault, value: VDefault): Promise<void>
+  put<K = KDefault, V = VDefault> (key: K, value: V, options: PutOptions<K, V>): Promise<void>
+
+  del (key: KDefault): Promise<void>
+  del<K = KDefault> (key: K, options: DelOptions<K>): Promise<void>
+
+  batch (operations: Array<BatchOperation<typeof this, KDefault, VDefault>>): Promise<void>
+  batch<K = KDefault, V = VDefault> (operations: Array<BatchOperation<typeof this, K, V>>, options: BatchOptions<K, V>): Promise<void>
+  batch (): ChainedBatch<typeof this, KDefault, VDefault>
+
+  iterator (): Iterator<typeof this, KDefault, VDefault>
+  iterator<K = KDefault, V = VDefault> (options: IteratorOptions<K, V>): Iterator<typeof this, K, V>
+
+  keys (): KeyIterator<typeof this, KDefault>
+  keys<K = KDefault> (options: KeyIteratorOptions<K>): KeyIterator<typeof this, K>
+
+  values (): ValueIterator<typeof this, KDefault, VDefault>
+  values<K = KDefault, V = VDefault> (options: ValueIteratorOptions<K, V>): ValueIterator<typeof this, K, V>
+
+  snapshot (options?: any | undefined): Snapshot
 }
 
 /**
  * Options for the {@link RaveLevel} constructor.
  */
-declare interface DatabaseOptions<K, V> extends
+export interface DatabaseOptions<K, V> extends
   Omit<AbstractDatabaseOptions<K, V>, 'createIfMissing' | 'errorIfExists'> {
   /**
    * If true, operations are retried upon connecting to a new leader. If false,
@@ -35,4 +100,31 @@ declare interface DatabaseOptions<K, V> extends
    * @defaultValue `true`
    */
   retry?: boolean
+
+  /**
+   * Custom socket path used for follower connections.
+   */
+  raveSocketPath?: string | undefined
 }
+
+export type OpenOptions = AbstractOpenOptions
+export type GetOptions<K, V> = AbstractGetOptions<K, V>
+export type GetManyOptions<K, V> = AbstractGetManyOptions<K, V>
+export type HasOptions<K> = AbstractHasOptions<K>
+export type HasManyOptions<K> = AbstractHasManyOptions<K>
+export type PutOptions<K, V> = AbstractPutOptions<K, V>
+export type DelOptions<K> = AbstractDelOptions<K>
+export type BatchOptions<K, V> = AbstractBatchOptions<K, V>
+export type ChainedBatchWriteOptions = AbstractChainedBatchWriteOptions
+
+export type BatchOperation<TDatabase, K, V> = AbstractBatchOperation<TDatabase, K, V>
+export type ChainedBatch<TDatabase, K, V> = AbstractChainedBatch<TDatabase, K, V>
+export type Iterator<TDatabase, K, V> = AbstractIterator<TDatabase, K, V>
+export type KeyIterator<TDatabase, K> = AbstractKeyIterator<TDatabase, K>
+export type ValueIterator<TDatabase, K, V> = AbstractValueIterator<TDatabase, K, V>
+
+export type IteratorOptions<K, V> = AbstractIteratorOptions<K, V>
+export type KeyIteratorOptions<K> = AbstractKeyIteratorOptions<K>
+export type ValueIteratorOptions<K, V> = AbstractValueIteratorOptions<K, V>
+
+export type Snapshot = AbstractSnapshot