const RocksDB = require('rocksdb-native')
const rrp = require('resolve-reject-promise')
const ScopeLock = require('scope-lock')
const DeviceFile = require('device-file')
const path = require('path')
const fs = require('fs')
const View = require('./lib/view.js')
const VERSION = 2
const COLUMN_FAMILY = 'corestore'
const { store, core } = require('./lib/keys.js')
const { CorestoreRX, CorestoreTX, CoreTX, CoreRX } = require('./lib/tx.js')
const {
createCoreStream,
createAliasStream,
createDiscoveryKeyStream,
createBlockStream,
createBitfieldStream,
createUserDataStream,
createTreeNodeStream,
createMarkStream,
createLocalStream,
createGroupUpdateStream
} = require('./lib/streams.js')
const EMPTY = new View()
class Atom {
constructor(db) {
this.db = db
this.view = new View()
this.flushedPromise = null
this.flushing = false
this.preflushes = []
this.flushes = []
}
preflush(fn) {
this.preflushes.push(fn)
}
onflush(fn) {
this.flushes.push(fn)
}
flushed() {
if (!this.flushing) return Promise.resolve()
if (this.flushedPromise !== null) return this.flushedPromise.promise
this.flushedPromise = rrp()
return this.flushedPromise.promise
}
_resolve() {
const f = this.flushedPromise
this.flushedPromise = null
f.resolve()
}
async flush() {
if (this.flushing) throw new Error('Atom already flushing')
this.flushing = true
try {
const plen = this.preflushes.length
for (let i = 0; i < plen; i++) this.preflushes[i]()
await View.flush(this.view.changes, this.db)
const promises = []
const len = this.flushes.length // in case of reentry
for (let i = 0; i < len; i++) promises.push(this.flushes[i]())
await Promise.all(promises)
} finally {
this.view.reset()
this.flushing = false
if (this.flushedPromise !== null) this._resolve()
}
}
}
class HypercoreStorage {
constructor(store, db, core, view, atom) {
this.store = store
this.db = db
this.core = core
this.view = view
this.atom = atom
this.view.readStart()
}
get readOnly() {
return this.store.readOnly
}
get dependencies() {
return this.core.dependencies
}
getDependencyLength() {
return this.core.dependencies.length
? this.core.dependencies[this.core.dependencies.length - 1].length
: -1
}
getDependency(length) {
for (let i = this.core.dependencies.length - 1; i >= 0; i--) {
const dep = this.core.dependencies[i]
if (dep.length < length) return dep
}
return null
}
setDependencyHead(dep) {
if (dep === null) {
this.core.dependencies = []
return
}
const deps = this.core.dependencies
for (let i = deps.length - 1; i >= 0; i--) {
const d = deps[i]
if (d.dataPointer !== dep.dataPointer) continue
// check if nothing changed
if (d.length === dep.length && i === deps.length - 1) return
this.core = {
corePointer: this.core.corePointer,
dataPointer: this.core.dataPointer,
dependencies: deps.slice(0, i + 1)
}
this.core.dependencies[i] = {
dataPointer: dep.dataPointer,
length: dep.length
}
return // updated head so done
}
this.core.dependencies = [
{
dataPointer: dep.dataPointer,
length: dep.length
}
]
}
// TODO: this might have to be async if the dependents have changed, but prop ok for now
updateDependencyLength(length, truncated) {
const deps = this.core.dependencies
const i = this.findDependencyIndex(length, truncated)
if (i === -1) throw new Error('Dependency not found')
this.core = {
corePointer: this.core.corePointer,
dataPointer: this.core.dataPointer,
dependencies: deps.slice(0, i + 1)
}
if (this.core.dependencies[i].length !== length) {
this.core.dependencies[i] = {
dataPointer: deps[i].dataPointer,
length
}
}
}
findDependencyIndex(length, truncated) {
const deps = this.core.dependencies
if (truncated) {
for (let i = 0; i < deps.length; i++) {
if (deps[i].length >= length) return i
}
return -1
}
for (let i = deps.length - 1; i >= 0; i--) {
if (deps[i].length <= length) return i
}
return -1
}
get snapshotted() {
return this.db._snapshot !== null
}
snapshot() {
return new HypercoreStorage(
this.store,
this.db.snapshot(),
this.core,
this.view.snapshot(),
this.atom
)
}
compact() {
return Promise.all([
this.db.compactRange(core.core(this.core.corePointer), core.core(this.core.corePointer)),
this.db.compactRange(core.data(this.core.dataPointer), core.data(this.core.dataPointer))
])
}
atomize(atom) {
if (this.atom && this.atom !== atom) {
throw new Error('Cannot atomize and atomized session with a new atom')
}
return new HypercoreStorage(this.store, this.db.session(), this.core, atom.view, atom)
}
createAtom() {
return this.store.createAtom()
}
createBlockStream(opts) {
return createBlockStream(this.core, this.db, this.view, opts)
}
createTreeNodeStream(opts) {
return createTreeNodeStream(this.core, this.db, this.view, opts)
}
createBitfieldStream(opts) {
return createBitfieldStream(this.core, this.db, this.view, opts)
}
createUserDataStream(opts) {
return createUserDataStream(this.core, this.db, this.view, opts)
}
createMarkStream(opts) {
return createMarkStream(this.core, this.db, this.view, opts)
}
createLocalStream(opts) {
return createLocalStream(this.core, this.db, this.view, opts)
}
async resumeSession(name) {
const rx = this.read()
const existingSessionsPromise = rx.getSessions()
rx.tryFlush()
const existingSessions = await existingSessionsPromise
const sessions = existingSessions || []
const session = getSession(sessions, name, false)
if (session === null) return null
const core = {
corePointer: this.core.corePointer,
dataPointer: session.dataPointer,
dependencies: []
}
const coreRx = new CoreRX(core, this.db, this.view)
const dependencyPromise = coreRx.getDependency()
coreRx.tryFlush()
const dependency = await dependencyPromise
if (dependency) core.dependencies = this._addDependency(dependency)
return new HypercoreStorage(
this.store,
this.db.session(),
core,
this.atom ? this.view : new View(),
this.atom
)
}
async createSession(name, head) {
const rx = this.read()
const existingSessionsPromise = rx.getSessions()
const existingHeadPromise = rx.getHead()
rx.tryFlush()
const [existingSessions, existingHead] = await Promise.all([
existingSessionsPromise,
existingHeadPromise
])
if (head === null) head = existingHead
if (existingHead !== null && head.length > existingHead.length) {
throw new Error('Invalid head passed, ahead of core')
}
const sessions = existingSessions || []
const session = getSession(sessions, name, true)
const fresh = session.dataPointer === -1
if (fresh) {
session.dataPointer = await this.store._allocData()
}
const tx = this.write()
tx.setSessions(sessions)
const length = head === null ? 0 : head.length
const core = {
corePointer: this.core.corePointer,
dataPointer: session.dataPointer,
dependencies: this._addDependency({
dataPointer: this.core.dataPointer,
length
})
}
const coreTx = new CoreTX(core, this.db, tx.view, tx.changes)
if (length > 0) coreTx.setHead(head)
coreTx.setDependency(core.dependencies[core.dependencies.length - 1])
if (!fresh) {
// nuke all existing state...
coreTx.deleteBlockRange(0, -1)
coreTx.deleteTreeNodeRange(0, -1)
coreTx.deleteBitfieldPageRange(0, -1)
}
await tx.flush()
return new HypercoreStorage(
this.store,
this.db.session(),
core,
this.atom ? this.view : new View(),
this.atom
)
}
async createAtomicSession(atom, head) {
const length = head === null ? 0 : head.length
const core = {
corePointer: this.core.corePointer,
dataPointer: this.core.dataPointer,
dependencies: this._addDependency(null)
}
const coreTx = new CoreTX(core, this.db, atom.view, [])
if (length > 0) coreTx.setHead(head)
await coreTx.flush()
return this.atomize(atom)
}
_addDependency(dep) {
const deps = []
for (let i = 0; i < this.core.dependencies.length; i++) {
const d = this.core.dependencies[i]
if (dep !== null && d.length > dep.length) {
if (d.dataPointer !== dep.dataPointer) {
deps.push({ dataPointer: d.dataPointer, length: dep.length })
}
return deps
}
deps.push(d)
}
if (
dep !== null &&
(deps.length === 0 || deps[deps.length - 1].dataPointer !== dep.dataPointer)
) {
deps.push(dep)
}
return deps
}
async deleteSessions() {
const rx = this.read()
const existingSessionsPromise = rx.getSessions()
rx.tryFlush()
const existingSessions = await existingSessionsPromise
// Remove batches
const coreTx = this.write()
coreTx.setSessions([]) // Clear sessions record
await coreTx.flush()
const tx = this.db.write({ autoDestroy: true })
for (const { dataPointer } of existingSessions) {
const start = core.data(dataPointer)
const end = core.data(dataPointer + 1)
tx.tryDeleteRange(start, end)
}
await tx.flush()
}
read() {
return new CoreRX(this.core, this.db, this.view)
}
write() {
return new CoreTX(this.core, this.db, this.atom ? this.view : null, [])
}
close() {
if (this.view !== null) {
this.view.readStop()
this.view = null
}
return this.db.close()
}
static async export(ptr, db, { batches = false } = {}) {
const rx = new CoreRX(ptr, db, EMPTY)
const core = {
head: null,
auth: null,
sessions: [],
data: null
}
const sessionsPromise = rx.getSessions()
const headPromise = rx.getHead()
const authPromise = rx.getAuth()
rx.tryFlush()
const [sessions, head, auth] = await Promise.all([sessionsPromise, headPromise, authPromise])
core.head = head
core.auth = { ...auth, keyPair: null }
if (sessions) core.sessions = sessions.map((s) => s.name)
const data = []
data.push(exportData(ptr, db))
if (batches) {
for (const { dataPointer } of sessions) {
data.push(exportData({ dataPointer, dependencies: [] }, db))
}
}
core.data = await Promise.all(data)
return core
}
static isParentStorage(storage, parent) {
const last = getLastDependency(storage)
return !!last && last.dataPointer === parent.core.dataPointer
}
}
class CorestoreStorage {
constructor(db, opts = {}) {
const storage = typeof db === 'string' ? db : null
this.bootstrap = storage !== null
this.path = storage !== null ? storage : path.join(db.path, '..')
this.alwaysRecover = !!opts.alwaysRecover
this.readOnly = !!opts.readOnly
this.allowBackup = !!opts.allowBackup
this.deviceFile = null
this.wait = !!opts.wait
// tmp sync fix for simplicty since not super deployed yet
if (this.bootstrap && !this.readOnly) tmpFixStorage(this.path)
this.id = opts.id || null
this.view = null
this.enters = 0
this.lock = new ScopeLock()
this.flushing = null
this.version = 0
this.migrating = null
if ((this.bootstrap && !this.readOnly && !this.allowBackup) || this.wait) {
const corestoreFile = path.join(this.path, 'CORESTORE')
this.deviceFile = new DeviceFile(corestoreFile, {
lock: true,
wait: this.wait,
data: { id: this.id }
})
}
const dbPath = path.join(this.path, 'db')
this.rocks = storage === null ? db : new RocksDB(dbPath, { ...opts, lock: this.deviceFile })
this.db = createColumnFamily(this.rocks, opts)
}
get opened() {
return this.db.opened
}
get closed() {
return this.db.closed
}
async ready() {
if (this.version === 0) await this._migrateStore()
return this.db.ready()
}
compact(opts) {
return this.db.compactRange(opts)
}
async audit() {
for await (const { core } of this.createCoreStream()) {
const coreRx = new CoreRX(core, this.db, EMPTY)
const authPromise = coreRx.getAuth()
coreRx.tryFlush()
const auth = await authPromise
if (!auth.manifest || auth.manifest.version > 0) continue
if (auth.manifest.linked === null) continue
auth.manifest.linked = null
const coreTx = new CoreTX(core, this.db, null, [])
coreTx.setAuth(auth)
await coreTx.flush()
}
}
async deleteCore(ptr) {
const rx = new CoreRX(ptr, this.db, EMPTY)
const authPromise = rx.getAuth()
const sessionsPromise = rx.getSessions()
rx.tryFlush()
const auth = await authPromise
const sessions = await sessionsPromise
// no core stored here
if (!auth) return
const tx = this.db.write({ autoDestroy: true })
tx.tryDelete(store.core(auth.discoveryKey))
// clear core
const start = core.core(ptr.corePointer)
const end = core.core(ptr.corePointer + 1)
tx.tryDeleteRange(start, end)
if (sessions) {
for (const { dataPointer } of sessions) {
const start = core.data(dataPointer)
const end = core.data(dataPointer + 1)
tx.tryDeleteRange(start, end)
}
}
return tx.flush()
}
static isCoreStorage(db) {
return isCorestoreStorage(db)
}
static from(db) {
if (isCorestoreStorage(db)) return db
return new this(db)
}
async _flush() {
while (this.enters > 0) {
await this.lock.lock()
await this.lock.unlock()
}
}
async _maybeRecover(view) {
const r = path.join(this.db.path, 'RECOVERING')
if (!(await fileExists(r))) return
await this.setRecovering(view)
await fs.promises.unlink(r)
}
_maybeRecoverLater(err) {
if (!shouldRecover(err) && !this.alwaysRecover) return
fs.writeFileSync(path.join(this.db.path, 'RECOVERING'), '')
for (const file of fs.readdirSync(this.db.path)) {
if (file.endsWith('.log')) fs.unlinkSync(path.join(this.db.path, file))
}
}
// runs pre any other mutation and read
async _migrateStore() {
const view = await this._enter()
try {
if (this.version === VERSION) return
await this._preopen
try {
await this.db.ready()
} catch (err) {
this._maybeRecoverLater(err)
throw err
}
await this._maybeRecover(view)
const rx = new CorestoreRX(this.db, view)
const headPromise = rx.getHead()
rx.tryFlush()
const head = await headPromise
const version = head === null ? 0 : head.version
if (version === VERSION) {
this.version = VERSION
return
}
const target = { version: VERSION, dryRun: false }
switch (version) {
case 0: {
await require('./migrations/0').store(this, target)
break
}
case 1: {
// implicit migration on write
break
}
default: {
throw new Error(
'Unsupported version: ' + version + ' - you should probably upgrade your dependencies'
)
}
}
this.version = VERSION
} finally {
await this._exit()
}
}
// runs pre the core is returned to the user
async _migrateCore(core, discoveryKey, version, locked) {
const view = locked ? this.view : await this._enter()
try {
if (version === VERSION) return
const target = { version: VERSION, dryRun: false }
switch (version) {
case 0: {
await require('./migrations/0').core(core, target)
break
}
case 1: {
await require('./migrations/1').core(core, target)
break
}
default: {
throw new Error(
'Unsupported version: ' + version + ' - you should probably upgrade your dependencies'
)
}
}
if (locked === false) return
// if its locked, then move the core state into the memview
// in case the core is reopened from the memview, pre flush
const rx = new CorestoreRX(this.db, EMPTY)
const tx = new CorestoreTX(view)
const corePromise = rx.getCore(discoveryKey)
rx.tryFlush()
tx.putCore(discoveryKey, await corePromise)
tx.apply()
} finally {
if (!locked) await this._exit()
}
}
async _enter() {
this.enters++
await this.lock.lock()
if (this.view === null) this.view = new View()
return this.view
}
async _exit() {
this.enters--
if (this.flushing === null) this.flushing = rrp()
const flushed = this.flushing.promise
if (this.enters === 0 || this.view.size() > 128) {
try {
await View.flush(this.view.changes, this.db)
this.flushing.resolve()
} catch (err) {
this.flushing.reject(err)
} finally {
this.flushing = null
this.view = null
}
}
this.lock.unlock()
return flushed
}
// when used with core catches this isnt transactional for simplicity, HOWEVER, its just a number
// so worth the tradeoff
async _allocData() {
let dataPointer = 0
const view = await this._enter()
const tx = new CorestoreTX(view)
try {
const head = await this._getHead(view)
dataPointer = head.datas++
tx.setHead(head)
tx.apply()
} finally {
await this._exit()
}
return dataPointer
}
// exposes here so migrations can easily access the head in an init state
async _getHead(view) {
const rx = new CorestoreRX(this.db, view)
const headPromise = rx.getHead()
rx.tryFlush()
const head = await headPromise
return head === null ? initStoreHead() : head
}
createAtom() {
return new Atom(this.db)
}
async flush() {
await this.rocks.flush()
}
async close() {
if (this.db.closed) return
await this._flush()
await this.db.close()
await this.rocks.close()
if (this.deviceFile) await this.deviceFile.close()
}
async clear() {
if (this.version === 0) await this._migrateStore()
const view = await this._enter()
const tx = new CorestoreTX(view)
tx.clear()
tx.apply()
await this._exit()
}
createCoreStream() {
// TODO: be nice to run the mgiration here also, but too much plumbing atm
return createCoreStream(this.db, EMPTY)
}
createAliasStream(namespace) {
// TODO: be nice to run the mgiration here also, but too much plumbing atm
return createAliasStream(this.db, EMPTY, namespace)
}
createDiscoveryKeyStream(namespace) {
return createDiscoveryKeyStream(this.db, EMPTY, namespace)
}
createGroupUpdateStream(group, { since, reverse } = {}) {
return createGroupUpdateStream(this.db, EMPTY, group, { gte: since, reverse })
}
async getAlias(alias) {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const discoveryKeyPromise = rx.getCoreByAlias(alias)
rx.tryFlush()
return discoveryKeyPromise
}
async getSeed() {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const headPromise = rx.getHead()
rx.tryFlush()
const head = await headPromise
return head === null ? null : head.seed
}
async getGroup(topic) {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const groupPromise = rx.getGroup(topic)
rx.tryFlush()
return groupPromise
}
async setSeed(seed, { overwrite = true } = {}) {
if (this.version === 0) await this._migrateStore()
const view = await this._enter()
const tx = new CorestoreTX(view)
try {
const rx = new CorestoreRX(this.db, view)
const headPromise = rx.getHead()
rx.tryFlush()
const head = (await headPromise) || initStoreHead()
if (head.seed === null || overwrite) head.seed = seed
tx.setHead(head)
tx.apply()
return head.seed
} finally {
await this._exit()
}
}
async setRecovering(view = null) {
const recovering = Date.now()
let keys = []
for await (const data of this.createDiscoveryKeyStream()) {
keys.push(data)
if (keys.length > 16 * 1024) {
await this._flushRecovering(recovering, view, keys)
keys = []
}
}
if (keys.length) await this._flushRecovering(recovering, view, keys)
}
async _flushRecovering(recovering, view, keys) {
const own = !view
if (own) view = await this._enter()
const tx = new CorestoreTX(view)
try {
const infos = await this._getInfos(keys, { auth: false, head: false, hints: true })
for (const inf of infos) {
tx.setCoreHints(inf.core, { ...inf.hints, recovering })
}
tx.apply()
} finally {
if (own) await this._exit()
}
}
async getDefaultDiscoveryKey() {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const headPromise = rx.getHead()
rx.tryFlush()
const head = await headPromise
return head === null ? null : head.defaultDiscoveryKey
}
async setDefaultDiscoveryKey(discoveryKey, { overwrite = true } = {}) {
if (this.version === 0) await this._migrateStore()
const view = await this._enter()
const tx = new CorestoreTX(view)
try {
const rx = new CorestoreRX(this.db, view)
const headPromise = rx.getHead()
rx.tryFlush()
const head = (await headPromise) || initStoreHead()
if (head.defaultDiscoveryKey === null || overwrite) head.defaultDiscoveryKey = discoveryKey
tx.setHead(head)
tx.apply()
return head.defaultDiscoveryKey
} finally {
await this._exit()
}
}
async hasCore(discoveryKey, { ifMigrated = false } = {}) {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const promise = rx.getCore(discoveryKey)
rx.tryFlush()
const core = await promise
if (core === null) return false
if (core.version === 0 && ifMigrated) return false
return true
}
async getAuth(discoveryKey) {
if (this.version === 0) await this._migrateStore()
const rx = new CorestoreRX(this.db, EMPTY)
const corePromise = rx.getCore(discoveryKey)
rx.tryFlush()
const core = await corePromise
if (core === null) return null
const read = this.db.read({ autoDestroy: true })
const authPromise = CoreRX.getAuth(read, core)
read.tryFlush()
return authPromise
}
async getInfo(discoveryKey, opts) {
return (await this.getInfos([discoveryKey], opts))[0]
}
async getInfos(discoveryKeys, opts) {
if (this.version === 0) await this._migrateStore()
return this._getInfos(discoveryKeys, opts)
}
async _getInfos(discoveryKeys, { auth = true, head = true, hints = true } = {}) {
const rx = new CorestoreRX(this.db, EMPTY)
const corePromises = new Array(discoveryKeys.length)
for (let i = 0; i < discoveryKeys.length; i++) {
corePromises[i] = rx.getCore(discoveryKeys[i])
}
rx.tryFlush()
const cores = await Promise.all(corePromises)
const read = this.db.read({ autoDestroy: true })
const resultPromises = new Array(cores.length)
for (let i = 0; i < cores.length; i++) {
resultPromises[i] = cores[i]
? getInfoFromBatch(read, cores[i], discoveryKeys[i], auth, head, hints)
: null
}
read.tryFlush()
return Promise.all(resultPromises)
}
async suspend({ log = noop } = {}) {
await log('Suspending hypercore storage...')
await this.db.suspend()
await log('db suspended')
if (this.deviceFile) {
await log('suspend deviceFile')
await this.deviceFile.suspend()
await log('deviceFile suspended')
}
}
async resume() {
if (this.deviceFile) await this.deviceFile.resume()
await this.db.resume()
}
async resumeCore(discoveryKey) {
if (this.version === 0) await this._migrateStore()
if (!discoveryKey) {
discoveryKey = await this.getDefaultDiscoveryKey()
if (!discoveryKey) return null
}
const rx = new CorestoreRX(this.db, EMPTY)
const corePromise = rx.getCore(discoveryKey)
rx.tryFlush()
const core = await corePromise
if (core === null) return null
return this._resumeFromPointers(EMPTY, discoveryKey, false, core)
}
async export(discoveryKey, opts) {
const rx = new CorestoreRX(this.db, EMPTY)
const corePromise = rx.getCore(discoveryKey)
rx.tryFlush()
const core = await corePromise
if (core === null) return null
let { dataPointer, corePointer } = core
const ptr = { corePointer, dataPointer, dependencies: [] }
while (true) {
const rx = new CoreRX({ dataPointer, corePointer: 0, dependencies: [] }, this.db, EMPTY)
const dependencyPromise = rx.getDependency()
rx.tryFlush()
const dependency = await dependencyPromise
if (!dependency) break
ptr.dependencies.push(dependency)
dataPointer = dependency.dataPointer
}
const session = this.db.session()
try {
return await HypercoreStorage.export(ptr, session, opts)
} finally {
await session.close()
}
}
async _resumeFromPointers(view, discoveryKey, create, { version, corePointer, dataPointer }) {
const core = { corePointer, dataPointer, dependencies: [] }
while (true) {
const rx = new CoreRX({ dataPointer, corePointer: 0, dependencies: [] }, this.db, view)
const dependencyPromise = rx.getDependency()
rx.tryFlush()
const dependency = await dependencyPromise
if (!dependency) break
core.dependencies.push(dependency)
dataPointer = dependency.dataPointer
}
const result = new HypercoreStorage(this, this.db.session(), core, EMPTY, null)
if (version < VERSION) await this._migrateCore(result, discoveryKey, version, create)
return result
}
// not allowed to throw validation errors as its a shared tx!
async _create(
view,
{ key, manifest, keyPair, encryptionKey, discoveryKey, alias, userData, core: ptrs }
) {
const rx = new CorestoreRX(this.db, view)
const tx = new CorestoreTX(view)
const corePromise = rx.getCore(discoveryKey)
const headPromise = rx.getHead()
rx.tryFlush()
let [core, head] = await Promise.all([corePromise, headPromise])
if (core) return this._resumeFromPointers(view, discoveryKey, true, core)
if (head === null) head = initStoreHead()
if (head.defaultDiscoveryKey === null) head.defaultDiscoveryKey = discoveryKey
const corePointer = ptrs ? ptrs.corePointer : head.cores++
const dataPointer = ptrs ? ptrs.dataPointer : head.datas++
core = { version: VERSION, corePointer, dataPointer, alias }
tx.setHead(head)
tx.putCore(discoveryKey, core)
if (alias) tx.putCoreByAlias(alias, discoveryKey)
const ptr = { corePointer, dataPointer, dependencies: [] }
const ctx = new CoreTX(ptr, this.db, view, tx.changes)
ctx.setAuth({
key,
discoveryKey,
manifest,
keyPair,
encryptionKey
})
if (userData) {
for (const { key, value } of userData) {
ctx.putUserData(key, value)
}
}
tx.apply()
return new HypercoreStorage(this, this.db.session(), ptr, EMPTY, null)
}
async createCore(data) {
if (this.version === 0) await this._migrateStore()
const view = await this._enter()
try {
return await this._create(view, data)
} finally {
await this._exit()
}
}
// not allowed to throw validation errors as its a shared tx!
async _createGroup(view, topic) {
const rx = new CorestoreRX(this.db, view)
const tx = new CorestoreTX(view)
const groupPromise = rx.getGroup(topic)
const headPromise = rx.getHead()
rx.tryFlush()
let group = await groupPromise
if (group !== null) {
return {
key: topic,
pointer: group
}
}
let head = await headPromise
if (head === null) head = initStoreHead()
group = head.groups++
tx.putGroup(topic, group)
tx.setHead(head)
tx.apply()
return {
key: topic,
pointer: group
}
}
async createGroup(topic) {
if (this.version === 0) await this._migrateStore()
const view = await this._enter()
try {
return await this._createGroup(view, topic)
} finally {
await this._exit()
}
}
static isParentStorage(storage, parent) {
return HypercoreStorage.isParentStorage(storage, parent)
}
}
module.exports = CorestoreStorage
function initStoreHead() {
return {
version: 0, // cause we wanna run the migration
datas: 0,
cores: 0,
groups: 0,
seed: null,
defaultDiscoveryKey: null
}
}
function getSession(sessions, name, alloc) {
for (let i = 0; i < sessions.length; i++) {
if (sessions[i].name === name) return sessions[i]
}
if (!alloc) return null
const result = { name, dataPointer: -1 }
sessions.push(result)
return result
}
function isCorestoreStorage(s) {
return typeof s === 'object' && !!s && typeof s.setDefaultDiscoveryKey === 'function'
}
function createColumnFamily(db, opts = {}) {
const {
tableCacheIndexAndFilterBlocks = true,
blockCache = true,
optimizeFiltersForMemory = false,
blobFileSize = 256 * 1024 * 1024,
blobGarbageCollectionAgeCutOff = 0.25,
blobGarbageCollectionForceThreshold = 1.0
} = opts
const col = new RocksDB.ColumnFamily(COLUMN_FAMILY, {
enableBlobFiles: true,
minBlobSize: 4096,
blobFileSize,
enableBlobGarbageCollection: true,
tableBlockSize: 8192,
tableCacheIndexAndFilterBlocks,
tableFormatVersion: 6,
optimizeFiltersForMemory,
blockCache,
blobGarbageCollectionAgeCutOff,
blobGarbageCollectionForceThreshold
})
return db.columnFamily(col)
}
function getLastDependency(storage) {
return storage.dependencies.length ? storage.dependencies[storage.dependencies.length - 1] : null
}
// TODO: remove in like 3-6 mo
function tmpFixStorage(p) {
// if CORESTORE file is written, new format
if (fs.existsSync(path.join(p, 'CORESTORE'))) return
let files = []
try {
files = fs.readdirSync(p)
} catch {}
const notRocks = new Set([
'CORESTORE',
'primary-key',
'cores',
'app-preferences',
'cache',
'preferences.json',
'db',
'clone',
'core',
'notifications'
])
for (const f of files) {
if (notRocks.has(f)) continue
try {
fs.mkdirSync(path.join(p, 'db'))
} catch {}
fs.renameSync(path.join(p, f), path.join(p, 'db', f))
}
}
async function exportData(ptr, db, opts) {
// just need dataPointer
const reads = [
toArray(createBlockStream(ptr, db, EMPTY, opts)),
toArray(createTreeNodeStream(ptr, db, EMPTY, opts)),
toArray(createBitfieldStream(ptr, db, EMPTY, opts))
]
const [blocks, tree, bitfield] = await Promise.all(reads)
return {
blocks,
tree,
bitfield
}
}
async function toArray(stream) {
const all = []
for await (const e of stream) all.push(e)
return all
}
async function fileExists(path) {
try {
await fs.promises.stat(path)
return true
} catch {
return false
}
}
function shouldRecover(err) {
return err.message.toLowerCase().indexOf('sequence number is being set backwards') > -1
}
function noop() {}
async function getInfoFromBatch(db, c, discoveryKey, getAuth, getHead, getHints) {
const authPromise = getAuth ? CoreRX.getAuth(db, c) : null
const headPromise = getHead ? CoreRX.getHead(db, c) : null
const hintsPromise = getHints ? CoreRX.getHints(db, c) : null
// ensure no uncaughts
if (authPromise) authPromise.catch(noop)
if (headPromise) headPromise.catch(noop)
if (hintsPromise) hintsPromise.catch(noop)
return {
discoveryKey,
core: c,
auth: await authPromise,
head: await headPromise,
hints: await hintsPromise
}
}