const Hyperbee = require('hyperbee')
const SubEncoder = require('sub-encoder')
const ReadyResource = require('ready-resource')
const b4a = require('b4a')
const c = require('compact-encoding')
const crypto = require('hypercore-crypto')
const { Info, Member } = require('./messages.js')
const { DEFAULT_AUTOBASE_VERSION } = require('./caps.js')
const subs = new SubEncoder()
const DIGEST = subs.sub(b4a.from([0]))
const MEMBERS = subs.sub(b4a.from([1]))
const INFO_KEY = DIGEST.encode('info')
const [GENESIS_ENTROPY, NS_ENTROPY] = crypto.namespace('autobase/entropy', 2)
module.exports = class SystemView extends ReadyResource {
constructor(core, { checkout = 0, empty = false } = {}) {
super()
this.core = core
// sessions is a workaround for batches not having sessions atm...
this.db = new Hyperbee(core, {
keyEncoding: 'binary',
extension: false,
checkout,
sessions: typeof core.session === 'function'
})
this.version = DEFAULT_AUTOBASE_VERSION
this.members = 0
this.pendingIndexers = []
this.indexers = []
this.heads = []
this.views = []
this.encryptionLength = 0
this.entropy = null
this.indexerUpdate = false
this._empty = empty
this._fork = 0
this._length = 0
this._changes = []
this._indexerMap = new Map()
this._clockUpdates = new Map()
}
static GENESIS_ENTROPY = GENESIS_ENTROPY
static async getIndexedInfo(core, length) {
const sys = new this(core.session())
try {
return await sys.getIndexedInfo(length)
} finally {
await sys.close()
}
}
static async *flushes(core, { reverse, lt = core.length, gte = 0, wait = true } = {}) {
if (lt <= 0) return
// ensure block
await core.get(lt - 1)
const sys = new SystemView(core)
try {
for await (const data of sys.db.createHistoryStream({ lt, gte, wait, reverse: true })) {
if (!b4a.equals(data.key, INFO_KEY)) continue
const info = c.decode(Info, data.value)
yield { length: data.seq + 1, info }
}
} finally {
await sys.close()
}
}
get bootstrapping() {
return this.members === 0
}
async checkout(length) {
const checkout = new SystemView(this.core.session(), {
checkout: length,
empty: length === 0
})
await checkout.ready()
return checkout
}
async _open() {
// this should NEVER fail, if so we have a bug elsewhere (should always be consistent)
const info = this._empty
? null
: await this.db.get('info', {
valueEncoding: Info,
keyEncoding: DIGEST,
update: false,
wait: false
})
await this._reset(info)
}
async _close() {
await this.db.close()
}
async getIndexedInfo(length = this.core.signedLength) {
if (this.opened === false) await this.ready()
if (length === this.core.length) {
return {
version: this.version,
members: this.members,
pendingIndexers: this.pendingIndexers,
indexers: this.indexers,
heads: this.heads,
views: this.views,
encryptionLength: this.encryptionLength,
entropy: this.entropy
}
}
const node = length === 0 ? null : await this.db.getBySeq(length - 1)
if (node === null) {
return {
version: DEFAULT_AUTOBASE_VERSION,
members: 0,
pendingIndexers: [],
indexers: [],
heads: [],
views: [],
encryptionLength: 0,
entropy: null
}
}
return c.decode(Info, node.value)
}
static sameIndexers(a, b) {
if (a.length !== b.length) return false
for (let i = 0; i < a.length; i++) {
if (!b4a.equals(a[i].key, b[i].core.key)) return false
}
return true
}
sameIndexers(indexers) {
return SystemView.sameIndexers(this.indexers, indexers)
}
async history(since) {
const checkout = this.db.checkout(since)
const seen = new Map()
const nodes = []
const updates = []
const checkoutNode = since === 0 ? null : await this.db.getBySeq(since - 1)
let prevInfo = checkoutNode === null ? null : c.decode(Info, checkoutNode.value)
let updateBatch = 0
for await (const data of this.db.createHistoryStream({ gte: since })) {
if (b4a.equals(data.key, INFO_KEY)) {
const info = c.decode(Info, data.value)
updates.push({
batch: updateBatch,
indexers: prevInfo === null || !sameIndexers(info, prevInfo),
systemLength: data.seq + 1
})
prevInfo = info
updateBatch = 0
continue
}
const key = data.key.subarray(2)
const hex = b4a.toString(key, 'hex')
const len = c.decode(Member, data.value).length
if (!seen.has(hex)) {
const node = await checkout.get(data.key)
if (node === null) {
seen.set(hex, 0)
} else {
const { length } = c.decode(Member, node.value)
seen.set(hex, length)
}
}
const prev = seen.get(hex)
const batch = len - prev
seen.set(hex, len)
if (batch === 0) continue
updateBatch += batch
if (nodes.length > 0) {
const top = nodes[nodes.length - 1]
if (b4a.equals(top.key, key)) {
top.length = len
top.batch += batch
continue
}
}
nodes.push({
key,
length: len,
batch
})
}
await checkout.close()
return { updates, nodes }
}
async update() {
if (this.opened === false) await this.ready()
if (this._fork === this.core.fork && this._length === this.core.length) return false
await this._reset(
this._empty ? null : await this.db.get('info', { valueEncoding: Info, keyEncoding: DIGEST })
)
return true
}
async _reset(info) {
this.version = info === null ? DEFAULT_AUTOBASE_VERSION : info.value.version
this.members = info === null ? 0 : info.value.members
this.pendingIndexers = info === null ? [] : info.value.pendingIndexers
this.indexers = info === null ? [] : info.value.indexers
this.heads = info === null ? [] : info.value.heads
this.views = info === null ? [] : info.value.views
this.encryptionLength = info === null ? 0 : info.value.encryptionLength
this.entropy = info === null ? null : info.value.entropy
this.indexerUpdate = false
this._indexerMap.clear()
this._clockUpdates.clear()
this._length = this.core.length
this._fork = this.core.fork
this._changes = []
for (const idx of this.indexers) {
this._indexerMap.set(b4a.toString(idx.key, 'hex'), idx)
}
}
// helper used by apply-state
flushLength() {
const header = this.core.length ? 0 : 1 // account for header
let updates = this._changes.length + this._clockUpdates.size
for (const { key } of this._changes) {
const hex = b4a.toString(key, 'hex')
if (this._clockUpdates.has(hex)) updates-- // shared are skipped
}
return header + this.core.length + updates + 1
}
async flush(views) {
const batch = this.db.batch({ update: false })
for (let i = 0; i < this._changes.length; i++) {
const c = this._changes[i]
if (this._clockUpdates.has(b4a.toString(c.key, 'hex'))) continue
await batch.put(c.key, c.value, { valueEncoding: Member, keyEncoding: MEMBERS })
}
for (const [hex, length] of this._clockUpdates) {
const isIndexer = this._indexerMap.get(hex) !== undefined
const key = b4a.from(hex, 'hex')
const info = await this._get(key, 0)
const value = { isIndexer, isRemoved: info ? info.isRemoved : true, length }
await batch.put(key, value, { valueEncoding: Member, keyEncoding: MEMBERS })
}
if (this.indexerUpdate) {
this._refreshEntropy(this.core.length + batch.length + 1)
}
this._clockUpdates.clear()
let maxIndex = -1
for (const view of views) {
if (view.mappedIndex > maxIndex) maxIndex = view.mappedIndex
}
while (this.views.length > maxIndex + 1) this.views.pop()
for (const view of views) {
const length = view.core ? view.core.length : view.length
if (!length) continue
const v = { key: view.key, length }
if (view.mappedIndex !== -1) {
this.views[view.mappedIndex] = v
} else {
view.mappedIndex = this.views.push(v) - 1
}
}
const info = {
version: this.version,
members: this.members,
pendingIndexers: this.pendingIndexers,
indexers: this.indexers,
heads: this.heads,
views: this.views,
encryptionLength: this.encryptionLength,
entropy: this.entropy
}
await batch.put('info', info, { valueEncoding: Info, keyEncoding: DIGEST })
await batch.flush()
this._length = this.core.length // should be ok
this._changes = []
if (this.indexerUpdate) this.indexerUpdate = false
}
getEntropy(indexers, length) {
return generateEntropy(indexers, length, this.entropy)
}
_refreshEntropy(length) {
if (this.version < 2) return
this.entropy = this.getEntropy(this.indexers, length)
}
checkpoint() {
return {
version: this.version,
members: this.members,
pendingIndexers: this.pendingIndexers.slice(0),
indexers: cloneNodes(this.indexers),
heads: cloneNodes(this.heads),
views: cloneNodes(this.views),
indexerUpdate: this.indexerUpdate,
encryptionLength: this.encryptionLength,
entropy: this.entropy ? b4a.from(this.entropy) : null,
changes: this._changes.slice(0),
clockUpdates: new Map([...this._clockUpdates])
}
}
applyCheckpoint(checkpoint) {
this.version = checkpoint.version
this.members = checkpoint.members
this.pendingIndexers = checkpoint.pendingIndexers
this.indexers = checkpoint.indexers
this.heads = checkpoint.heads
this.views = checkpoint.views
this.indexerUpdate = checkpoint.indexerUpdate
this.encryptionLength = checkpoint.encryptionLength
this.entropy = checkpoint.entropy
this._changes = checkpoint.changes
this._clockUpdates = checkpoint.clockUpdates
this._indexerMap = new Map()
for (const idx of this.indexers) {
this._indexerMap.set(b4a.toString(idx.key, 'hex'), idx)
}
}
addHead(node) {
const h = { key: node.writer.core.key, length: node.length }
for (let i = 0; i < this.heads.length; i++) {
const head = this.heads[i]
if (!hasDependency(node, head)) {
if (!b4a.equals(node.writer.core.key, head.key)) continue
// todo: remove in next major because bug was fixed here:
// https://github.com/holepunchto/autobase-next/pull/237
// filter out any bad heads introduced by a bug to
// prevent inconsistencies being written to the oplog
if (head.length > h.length) return false
}
const popped = this.heads.pop()
if (popped !== head) this.heads[i--] = popped
}
this.heads.push(h)
const hex = b4a.toString(h.key, 'hex')
this._clockUpdates.set(hex, h.length)
if (this.pendingIndexers.length > 0) {
for (let i = 0; i < this.pendingIndexers.length; i++) {
if (!b4a.equals(this.pendingIndexers[i], h.key)) continue
this._updateIndexer(h.key, h.length, true, i)
return true
}
}
const idx = this._indexerMap.get(hex)
if (idx !== undefined) {
idx.length = h.length
}
return false
}
_updateIndexer(key, length, isIndexer, i) {
const hex = b4a.toString(key, 'hex')
if (!isIndexer) {
const existing = this._indexerMap.get(hex)
if (existing) {
this.indexerUpdate = true
this.indexers.splice(this.indexers.indexOf(existing), 1)
this._indexerMap.delete(hex)
}
return
}
for (; i < this.pendingIndexers.length; i++) {
if (b4a.equals(this.pendingIndexers[i], key)) {
break
}
}
if (length === 0) {
if (i >= this.pendingIndexers.length) this.pendingIndexers.push(key)
return
}
if (i < this.pendingIndexers.length) {
const top = this.pendingIndexers.pop()
if (i < this.pendingIndexers.length) this.pendingIndexers[i] = top
}
const idx = this._indexerMap.get(hex)
if (idx === undefined) {
const newIdx = { key, length }
this._indexerMap.set(hex, newIdx)
this.indexers.push(newIdx)
// bootstrap is "silently" added so that initial views have no prologue
if (!this.bootstrapping) this.indexerUpdate = true
} else {
idx.length = length
}
}
_seenLength(key) {
return this._clockUpdates.get(b4a.toString(key, 'hex')) || 0
}
async ack(key) {
const value = await this._get(key, 0)
const length = this._seenLength(key)
if (value && value.length === length) return
const isIndexer = value ? value.isIndexer : false
const isRemoved = value ? value.isRemoved : true
this._changes.push({ key, value: { isIndexer, isRemoved, length } })
}
async add(key, { isIndexer = false, length = this._seenLength(key) } = {}) {
let value = null
let found = false
let changed = true
for (let i = this._changes.length - 1; i >= 0; i--) {
const c = this._changes[i]
if (b4a.equals(key, c.key)) {
value = c.value
found = true
break
}
}
if (!found) {
const node = await this.db.get(key, { valueEncoding: Member, keyEncoding: MEMBERS })
if (node) {
value = node.value
found = true
}
}
let wasTracked = false
let wasIndexer = false
if (found) {
if (!value.isRemoved) wasTracked = true
if (value.isIndexer) wasIndexer = true
if (length < value.length) length = value.length
if (value.length === length && value.isIndexer === isIndexer && value.isRemoved === false)
changed = false
}
if (changed) {
this._changes.push({ key, value: { isIndexer, isRemoved: false, length } })
}
if (!wasTracked) this.members++
if (wasIndexer || isIndexer) this._updateIndexer(key, length, isIndexer, 0)
}
async remove(key) {
let isIndexer = false
for (const idx of this.indexers) {
isIndexer = b4a.equals(idx.key, key)
if (isIndexer) break
}
if (isIndexer) this._updateIndexer(key, null, false, 0)
let value = null
let found = false
for (let i = this._changes.length - 1; i >= 0; i--) {
const c = this._changes[i]
if (b4a.equals(key, c.key)) {
value = c.value
found = true
break
}
}
if (!found) {
const node = await this.db.get(key, { valueEncoding: Member, keyEncoding: MEMBERS })
if (node) {
value = node.value
found = true
}
}
const changed = !found || !value.isRemoved
const wasTracked = found && !value.isRemoved
const length = found ? value.length : 0
if (changed) {
this._changes.push({ key, value: { isIndexer: false, isRemoved: true, length } })
}
if (wasTracked) this.members--
return isIndexer
}
async linkable(key, length) {
const len = this._seenLength(key)
if (len > 0) return length > len
const info = await this._get(key, 0)
const prevLength = info ? info.length : 0
return length > prevLength
}
async has(key, opts) {
// could be optimised...
return (await this.get(key, opts)) !== null
}
async _get(key, timeout) {
let value = null
let found = false
for (let i = this._changes.length - 1; i >= 0; i--) {
const c = this._changes[i]
if (b4a.equals(key, c.key)) {
value = c.value
found = true
break
}
}
if (!found) {
const node = await this.db.get(key, { timeout, valueEncoding: Member, keyEncoding: MEMBERS })
if (node) {
value = node.value
found = true
}
}
return found ? value : null
}
async get(key, opts = {}) {
if (this._empty) return null
let value = await this._get(key, opts.timeout || 0)
if (opts.unflushed) {
const hex = b4a.toString(key, 'hex')
for (let i = this._changes.length - 1; i >= 0; i--) {
const c = this._changes[i]
if (b4a.equals(c.key, key)) value = c.value
}
if (this._clockUpdates.has(hex)) value.length = this._clockUpdates.get(hex)
}
return opts.onlyActive !== false || !value.isRemoved ? value : null
}
async hasLocal(key) {
if (this._empty) return false
try {
const node = await this.db.get(key, {
valueEncoding: Member,
keyEncoding: MEMBERS,
update: false,
wait: false
})
return node !== null
} catch {
return false
}
}
async getLocalLength(key) {
if (this._empty) return 0
try {
const node = await this.db.get(key, {
valueEncoding: Member,
keyEncoding: MEMBERS,
update: false,
wait: false
})
return node === null ? 0 : node.value.length
} catch {
return 0
}
}
list() {
// note, NOT safe to use during apply atm
return this.db.createReadStream({
valueEncoding: Member,
keyEncoding: MEMBERS
})
}
async isIndexed(key, length) {
const co = this.db.checkout(this.core.indexedLength)
try {
const node = await co.get(key, { valueEncoding: Member, keyEncoding: MEMBERS })
return node !== null && node.value.length >= length
} finally {
await co.close()
}
}
async fork(indexers, manifests, encryptionLength, views) {
this.indexers = []
this.pendingIndexers = []
this.views = views
if (this.version < 2) this.version = 2 // bump to enable entropy
for (const { key } of indexers) {
// todo: probably need to check isRemoved etc
const { length } = await this.get(key)
this.indexers.push({ key, length })
}
this._refreshEntropy(this.core.length)
const blocks = []
// truncate needs to respect hypercore batch
let truncate = this.core.length - 1
for await (const data of this.db.createHistoryStream({ lt: truncate, reverse: true })) {
if (b4a.equals(data.key, INFO_KEY)) {
truncate = data.seq + 1
break
}
blocks.push(data)
}
await this.core.truncate(truncate, this.core.fork)
const batch = this.db.batch({ update: false })
await batch.ready()
for (const { key, value } of blocks) {
await batch.put(key, value)
}
const info = {
version: this.version,
members: this.members,
pendingIndexers: this.pendingIndexers,
indexers: this.indexers,
heads: this.heads,
views: this.views,
encryptionLength,
entropy: this.entropy
}
await batch.put('info', info, { valueEncoding: Info, keyEncoding: DIGEST })
await batch.flush()
await this._reset(await this.db.get('info', { valueEncoding: Info, keyEncoding: DIGEST }))
return truncate
}
}
function hasDependency(node, dep) {
for (const h of node.actualHeads) {
if (sameNode(h, dep)) return true
}
return false
}
function sameNode(a, b) {
return b4a.equals(a.key, b.key) && a.length === b.length
}
function sameIndexers(a, b) {
if (a.views.length > 0 && b.views.length > 0) return b4a.equals(a.views[0].key, b.views[0].key)
if (a.indexers.length !== b.indexers.length) return false
for (let i = 0; i < a.indexers.length; i++) {
if (!b4a.equals(a.indexers[i].key, b.indexers[i].key)) return false
}
return true
}
function cloneNodes(arr) {
const c = []
for (let i = 0; i < arr.length; i++) {
c.push({ key: arr[i].key, length: arr[i].length })
}
return c
}
function generateEntropy(indexers, length, entropy) {
const buffer = b4a.alloc(8 + 32 + indexers.length * 32)
const state = { start: 0, end: buffer.byteLength, buffer }
c.uint64.encode(state, length)
c.fixed32.encode(state, entropy || GENESIS_ENTROPY)
for (const { key } of indexers) c.fixed32.encode(state, key)
return crypto.hash([NS_ENTROPY, buffer])
}