const EventEmitter = require('events')
const sameData = require('same-data')
const unixPathResolve = require('unix-path-resolve')
const streamEquals = require('binary-stream-equals')
const speedometer = require('speedometer')
const { pipelinePromise, isStream } = require('streamx')
const RabinStream = require('rabin-stream')
const SAME = 0
const DIFF = 1
const DIFF_META = 2
class Monitor extends EventEmitter {
constructor(mirror, { interval = 250 } = {}) {
super()
this.mirror = mirror
this.interval = setInterval(this.update.bind(this), interval)
this.stats = null
this.index = mirror.monitors.push(this) - 1
this.update() // populate latest stats
}
get preloaded() {
return this.mirror.preloaded
}
get destroyed() {
return this.index === -1
}
update() {
if (this.index === -1) return
// NOTE: immutable (append-only) data structure
this.stats = {
peers: this.mirror.peers.length,
download: {
bytes: this.mirror.downloadedBytes,
blocks: this.mirror.downloadedBlocks,
speed: this.mirror.downloadSpeed(),
progress: this.mirror.downloadProgress
},
upload: {
bytes: this.mirror.uploadedBytes,
blocks: this.mirror.uploadedBlocks,
speed: this.mirror.uploadSpeed()
}
}
this.emit('update', this.stats)
}
destroy() {
if (this.index === -1) return
clearInterval(this.interval)
const head = this.mirror.monitors.pop()
if (head !== this) {
this.mirror.monitors[this.index] = head
head.index = this.index
}
this.index = -1
this.emit('destroy')
}
}
module.exports = class MirrorDrive {
constructor(src, dst, opts = {}) {
this.src = src
this.dst = dst
this.prefix = toArray(opts.prefix || '/')
this.dedup = !!opts.dedup
this.dryRun = !!opts.dryRun
this.prune = opts.prune !== false
this.preload = opts.preload !== false && (!!src.getBlobs || !!dst.getBlobs)
this.preloaded = false
this.includeProgress = !!opts.progress && !!src.getBlobs
this.includeEquals = !!opts.includeEquals
this.filter = opts.filter || null
this.metadataEquals = opts.metadataEquals || null
this.batch = !!opts.batch
this.entries = opts.entries || null
this.transformers = opts.transformers || []
this.count = { files: 0, add: 0, remove: 0, change: 0 }
this.bytesRemoved = 0
this.bytesAdded = 0
this.ignore = opts.ignore ? toIgnoreFunction(opts.ignore) : null
this.finished = false
this.downloadedBlocks = 0
this.downloadedBlocksEstimate = 0
this.downloadedBytes = 0
this.downloadSpeed = this.includeProgress ? speedometer() : null
this.uploadedBlocks = 0
this.uploadedBytes = 0
this.uploadSpeed = this.includeProgress ? speedometer() : null
this.monitors = []
this.iterator = this._init()
}
[Symbol.asyncIterator]() {
return this.iterator
}
get peers() {
return this.src.core?.peers || []
}
get downloadProgress() {
if (this.finished) return 1
if (!this.downloadedBlocksEstimate) return 0
// leave 3% incase our estimatation is wrong - then at least it wont appear done...
return Math.min(0.99, this.downloadedBlocks / this.downloadedBlocksEstimate)
}
monitor(opts) {
this.includeProgress = true
if (this.downloadSpeed === null) this.downloadSpeed = speedometer()
if (this.uploadSpeed === null) this.uploadSpeed = speedometer()
return new Monitor(this, opts)
}
async done() {
while (true) {
const { done } = await this.iterator.next()
if (done) break
}
}
_onupload(index, byteLength) {
this.uploadedBlocks++
this.uploadedBytes += byteLength
this.uploadSpeed(byteLength)
}
_ondownload(index, byteLength) {
this.downloadedBlocks++
this.downloadedBytes += byteLength
this.downloadSpeed(byteLength)
}
async _getBlobRange(blobs, entry) {
const blob = entry.value.blob
if (!blob || !blob.byteLength) return null
if (!blob.byteLength) {
const dl = blobs.core.download({ start: blob.blockOffset, length: blob.blockLength })
await dl.ready()
return dl
}
const map = await blobs.getBlockMap(blob)
if (!map) return null
const blocks = []
for (const b of map.blocks) blocks.push(b.index)
const dl = blobs.core.download({ blocks })
await dl.ready()
return dl
}
async _flushPreload(blobs, entries) {
const promises = []
for (const entry of entries) {
promises.push(this._getBlobRange(blobs, entry))
}
const ranges = await Promise.all(promises)
this.downloadedBlocksEstimate = this.downloadedBlocks
for (const dl of ranges) {
if (!dl || !dl.request.context) continue
this.downloadedBlocksEstimate += dl.request.context.end - dl.request.context.start
}
this.preloaded = true
for (const m of this.monitors) {
m.emit('preloaded')
}
for (const dl of ranges) {
if (!dl) continue
await dl.done()
}
}
async *_init() {
try {
for await (const out of this._mirror()) yield out
} finally {
while (this.monitors.length) {
this.monitors[this.monitors.length - 1].destroy()
}
}
}
async *_mirror() {
await this.src.ready()
await this.dst.ready()
if (this.dst.core && !this.dst.core.writable) throw new Error('Destination must be writable')
const blobs = this.includeProgress ? await this.src.getBlobs() : null
const onupload = this._onupload.bind(this)
const ondownload = this._ondownload.bind(this)
if (blobs) {
blobs.core.on('upload', onupload)
blobs.core.on('download', ondownload)
}
const dst = this.batch ? this.dst.batch() : this.dst
const dstBlobs = dst.getBlobs ? await dst.getBlobs() : null
const srcBlobs = this.src.getBlobs ? await this.src.getBlobs() : null
if (this.preload) {
const blobs = srcBlobs || dstBlobs
const entries = []
const maps = []
// we are gonna upgrade how much inflight blobs cause maps are tiny, remember old
const inflight = blobs.core.replicator.inflightRange
const d = srcBlobs ? null : dst
for await (const [, srcEntry, dstEntry] of this._list(this.src, d, this.filter)) {
const entry = d ? dstEntry : srcEntry
if (!entry) continue
entries.push(entry)
const blob = entry.value.blob
if (blob && blob.blockMap && blob.blockLength) {
if (maps.length === 0) {
// bump it
blobs.core.replicator.setInflightRange(256, 512)
}
maps.push(blobs.core.download({ start: blob.blockOffset, length: blob.blockLength }))
}
}
for (const m of maps) await m.done()
blobs.core.replicator.setInflightRange(inflight)
// flush in bg
this._flushPreload(blobs, entries).catch(noop)
}
if (this.prune) {
for await (const [key, dstEntry, srcEntry] of this._list(this.dst, this.src, null)) {
if (srcEntry) continue
const removed = await blobLength(dstEntry, dstBlobs)
this.count.remove++
this.bytesRemoved += removed
yield { op: 'remove', key, bytesRemoved: removed, bytesAdded: 0 }
if (!this.dryRun) await dst.del(key)
}
}
for await (const [key, srcEntry, dstEntry] of this._list(this.src, dst, this.filter)) {
if (!srcEntry) continue // Due entries option, src entry might not exist probably because it was pruned
this.count.files++
// If transformers are provided, we can't know if same before running them
const hasTransformers = this.transformers && this.transformers.length > 0
const s = hasTransformers === false ? await same(this, srcEntry, dstEntry) : DIFF
if (s === SAME) {
if (this.includeEquals) {
yield { op: 'equal', key, bytesRemoved: 0, bytesAdded: 0 }
}
continue
}
const onlyMetadata = s === DIFF_META && !!dst.putEntry
const dedup = !!dst.putEntry && this.dedup
if (dstEntry) {
const removed = onlyMetadata ? 0 : await blobLength(dstEntry, dstBlobs)
const added = onlyMetadata ? 0 : await blobLength(srcEntry, srcBlobs)
this.count.change++
this.bytesRemoved += removed
this.bytesAdded += added
yield {
op: 'change',
key,
bytesRemoved: removed,
bytesAdded: added
}
} else {
const added = await blobLength(srcEntry, srcBlobs)
this.count.add++
this.bytesAdded += added
yield { op: 'add', key, bytesRemoved: 0, bytesAdded: added }
}
if (this.dryRun) {
continue
}
const transformers = []
for (const transformer of this.transformers) {
if (typeof transformer !== 'function') throw new Error('transformer must be a function')
const stream = transformer(key)
if (stream === null) continue
if (!isStream(stream)) throw new Error('transformer must return a stream')
transformers.push(stream)
}
if (srcEntry.value.linkname) {
await dst.symlink(key, srcEntry.value.linkname)
} else if (!onlyMetadata) {
if (dedup) transformers.push(new RabinStream())
await pipelinePromise(
this.src.createReadStream(srcEntry),
...transformers,
dst.createWriteStream(key, {
dedup,
executable: srcEntry.value.executable,
metadata: srcEntry.value.metadata
})
)
} else {
await dst.putEntry(key, {
executable: srcEntry.value.executable,
linkname: srcEntry.value.linkname,
blob: dstEntry.value.blob,
metadata: srcEntry.value.metadata
})
}
}
if (this.batch) await dst.flush()
if (blobs) {
blobs.core.off('upload', onupload)
blobs.core.off('download', ondownload)
}
this.finished = true
}
async *_list(a, b, filter) {
const lists = []
for (const prefix of this.prefix) {
if (this.entries) {
lists.push(this.entries)
} else {
const stream = a.list(prefix, { ignore: this.ignore })
if (stream.on) {
stream.on('error', noop)
stream.resume()
stream.pause()
}
lists.push(stream)
}
}
for (let i = 0; i < this.prefix.length; i++) {
const prefix = this.prefix[i]
const list = lists[i]
for await (const entry of list) {
const key = typeof entry === 'object' ? entry.key : entry
if (filter && !filter(key)) continue
const entryA = await a.entry(entry)
const entryB = b ? await b.entry(key) : null
yield [key, entryA, entryB]
}
if (prefix !== '/' && (!filter || filter(prefix))) {
const entryA = await a.entry(prefix)
const entryB = b ? await b.entry(prefix) : null
if (!entryA && !entryB) continue
yield [prefix, entryA, entryB]
}
}
}
}
function blobLength(entry, blobs) {
const blob = entry.value.blob
if (!blob) return 0
if (!blob.blockMap) return blob.byteLength
return blobs.getByteLength(blob)
}
async function same(m, srcEntry, dstEntry) {
if (!dstEntry) return DIFF
if (srcEntry.value.linkname || dstEntry.value.linkname) {
return srcEntry.value.linkname === dstEntry.value.linkname ? SAME : DIFF
}
if (!maybeEquals(srcEntry, dstEntry)) return DIFF
const eq = await streamEquals(m.src.createReadStream(srcEntry), m.dst.createReadStream(dstEntry))
const diff = eq ? DIFF_META : DIFF
if (srcEntry.value.executable !== dstEntry.value.executable) return diff
if (!metadataEquals(m, srcEntry, dstEntry)) return diff
return eq ? SAME : DIFF
}
function maybeEquals(srcEntry, dstEntry) {
const srcBlob = srcEntry.value.blob
const dstBlob = dstEntry.value.blob
if (!srcBlob && !dstBlob) return true
if (!srcBlob || !dstBlob) return false
if (srcBlob.blockMap || dstBlob.blockMap) return true
return srcBlob.byteLength === dstBlob.byteLength
}
function metadataEquals(m, srcEntry, dstEntry) {
if (!m.src.supportsMetadata || !m.dst.supportsMetadata) return true
const srcMetadata = srcEntry.value.metadata
const dstMetadata = dstEntry.value.metadata
if (m.metadataEquals) {
return m.metadataEquals(srcMetadata, dstMetadata)
}
const noMetadata = !srcMetadata && !dstMetadata
const identicalMetadata = !!(srcMetadata && dstMetadata && sameData(srcMetadata, dstMetadata))
return noMetadata || identicalMetadata
}
function toIgnoreFunction(ignore) {
if (typeof ignore === 'function') return ignore
const all = [].concat(ignore).map((e) => unixPathResolve('/', e))
return (key) => all.some((path) => path === key || key.startsWith(path + '/'))
}
function toArray(prefix) {
return Array.isArray(prefix) ? prefix : [prefix]
}
function noop() {}