const { EventEmitter: e } = require("events"),
t = require("index-encoder"),
i = require("sub-encoder"),
s = require("mutexify/promise"),
r = require("b4a"),
{ pipeline: n, Transform: a } = require("streamx"),
o = r.alloc(1)
class c extends e {
constructor(e, t) {
;(super(),
(this.index = e),
(this.watcher = t),
(this.node = null),
this.watcher.on("error", (e) => this.emit("error", e)),
this.watcher.on("update", () => {
;(this._updateNode(), this.emit("update"))
}),
this._updateNode())
}
_updateNode() {
const e = this.watcher.node
this.node = e ? { key: e.key, value: e.value } : null
}
close() {
return this.watcher.close()
}
}
class h extends e {
constructor(e, t) {
;(super(),
(this.collection = e),
(this.watcher = t),
(this.node = null),
this.watcher.on("error", (e) => this.emit("error", e)),
this.watcher.on("update", () => {
;(this._updateNode(), this.emit("update"))
}),
this._updateNode())
}
_updateNode() {
this.node = this.watcher.node
? { _id: this.watcher.node.key, value: this.watcher.node.value }
: null
}
close() {
return this.watcher.close()
}
}
class l {
constructor(e, t) {
;((this.bee = e), (this.encodings = t))
}
async get(e) {
const t = await this.bee.get(e, this.encodings)
return t ? { key: t.key, value: t.value } : null
}
async delete(e) {
await this.bee.del(e, this.encodings)
}
insert(e, t = o) {
return this.bee.put(e, t, this.encodings)
}
async peek(e) {
const t = await this.bee.peek(e, this.encodings)
return t ? { key: t.key, value: t.value } : null
}
createReadStream(e = {}) {
const t = this.bee.createReadStream(e, this.encodings)
return n(
t,
new a({
transform: (e, t) => {
t(null, { key: e.key, value: e.value })
},
}),
)
}
createDiffStream(e, t = {}) {
const i = this.bee.createDiffStream(e, { ...t, ...this.encodings })
return n(
i,
new a({
transform: (e, t) => {
t(null, {
left: e.left ? { key: e.left.key, value: e.left.value } : null,
right: e.right ? { key: e.right.key, value: e.right.value } : null,
})
},
}),
)
}
async getAndWatch(e, t) {
const i = await this.bee.getAndWatch(e, { ...t, ...this.encodings })
return new c(this, i)
}
watch(e, t) {
return this.bee.watch(e, { ...t, ...this.encodings })
}
transaction(e) {
return new l(e.bee, this.encodings)
}
}
class u {
constructor(e, t, { indexes: i, indexer: s, transaction: r, autoincrement: n } = {}) {
;((this.sub = e),
(this.primary = t),
(this.indexes = i || null),
(this.indexer = s || null),
(this.insert = this._locked(this._insert.bind(this))),
(this.update = this._locked(this._update.bind(this))),
(this.delete = this._locked(this._delete.bind(this))),
(this._autoincrement = !!n),
(this._transaction = r))
}
_locked(e) {
return async (t, i, s) => {
if (!this._transaction) throw new Error("Can only mutate a collection with a transaction")
const r = await this._transaction.writeLock(this)
try {
return await e(t, i, s)
} finally {
r()
}
}
}
async _autoincrementId() {
const e = await this.primary.peek({ reverse: !0 })
return e ? e.key + 1 : 0
}
async _update(e, t, { upsert: i = !1 } = {}) {
const s = await this.primary.get(e)
if (!s && !i) throw new Error("Cannot update a record that does not exist")
return (
await this.primary.insert(e, t),
this.indexer && (await this.indexer(this._transaction, { id: e, prev: s?.value, next: t })),
e
)
}
async _insert(e) {
const t = this._autoincrement ? await this._autoincrementId() : e._id
return this._update(t, e, { upsert: !0 })
}
async _delete(e) {
const t = await this.primary.get(e)
t &&
(await this.primary.delete(e),
this.indexer && (await this.indexer(this._transaction, { id: e, prev: t.value, next: null })))
}
async get(e) {
const t = await this.primary.get(e)
return t ? { _id: t.key, value: t.value } : null
}
async peek(e) {
const t = await this.primary.peek(e)
return t ? { _id: t.key, value: t.value } : null
}
createReadStream(e) {
const t = this.primary.createReadStream(e)
return n(t, new a({ transform: (e, t) => t(null, { _id: e.key, value: e.value }) }))
}
createDiffStream(e, t) {
const i = this.primary.createDiffStream(e, t)
return n(
i,
new a({
transform: (e, t) =>
t(null, {
left: e.left ? { _id: e.left.key, value: e.left.value } : null,
right: e.right ? { _id: e.right.key, value: e.right.value } : null,
}),
}),
)
}
async getAndWatch(e, t) {
const i = await this.primary.getAndWatch(e, t)
return new h(this, i)
}
watch(e, t) {
return this.primary.watch(e, t)
}
transaction(e) {
return new u(this.sub, this.primary.transaction(e), {
transaction: e,
indexer: this.indexer,
autoincrement: this._autoincrement,
indexes: this.indexes && this.indexes.map((t) => t.transaction(e)),
})
}
createIndexes(e, t) {
if (this.indexes || this.indexer) throw new Error("Already created indexes")
;((this.indexes = []), (this.indexer = t))
let i = 1
for (const t of e)
if (Array.isArray(t)) {
const e = []
for (const s of t) e.push(w(this.primary.bee, this.sub, i++, s))
this.indexes.push(e)
} else this.indexes.push(w(this.primary.bee, this.sub, i++, t))
return this.indexes
}
}
class d {
constructor(e, { collections: t, sub: r, writable: n, release: a, offset: o = 0 } = {}) {
;((this.bee = e),
(this.sub = r || new i()),
(this.collections = t ? t.map((e) => e.transaction(this)) : null),
(this._writable = !!n),
(this._writeLock = this.collections ? null : s()),
(this._collectionLocks = this._writable ? new Map() : null),
(this._release = a),
(this._offset = o))
}
get version() {
return this.bee.version
}
async writeTransaction(e) {
const t = await this._writeLock()
return new this.constructor(this.bee.batch(e), {
collections: this.collections,
writable: !0,
sub: this.sub,
release: t,
})
}
readTransaction(e) {
return new this.constructor(this.bee.batch(e), {
collections: this.collections,
writable: !1,
sub: this.sub,
})
}
createCollections(e) {
if (this.collections) throw new Error("Already created collections")
this.collections = []
for (let t = 0; t < e.length; t++) {
const i = this.sub.sub(r.from([t + this._offset])),
s = { ...e[t], transaction: this._writable ? this : null },
n = w(this.bee, i, 0, s)
this.collections.push(new u(i, n, s))
}
return this.collections
}
async writeLock(e) {
if (!this._writable) throw new Error("Transaction is not writable")
let t = this._collectionLocks.get(e)
return (t || ((t = s()), this._collectionLocks.set(e, t)), t())
}
async commit() {
if (!this._writable) throw new Error("Transaction is not writable")
;(await Promise.all([...this._collectionLocks.values()].map((e) => e())),
await this.bee.flush(),
this._releaseLock())
}
async close() {
;(await this.bee.close(), this._releaseLock())
}
_releaseLock() {
const e = this._release
;((this._release = null), e && e())
}
static UINT = t.UINT
static STRING = t.STRING
static BUFFER = t.BUFFER
}
function w(e, i, s, n) {
let { keyEncoding: a, valueEncoding: o, autoincrement: c } = n
c && (a = d.UINT)
const h = Array.isArray(a) ? new t(a) : a,
u = { keyEncoding: i.sub(r.from([s]), h), valueEncoding: o }
return new l(e, u)
}
module.exports = d