const e = require("path"),
t = require("fs"),
a = require("hypercore-blob-server"),
i = require("hyperblobs"),
r = require("ready-resource"),
l = require("safety-catch"),
o = require("hypercore-id-encoding"),
{ pipelinePromise: n, Readable: s } = require("streamx"),
c = require("b4a"),
d = 6048e5
module.exports = class extends r {
constructor(e, t = {}) {
;(super(),
(this.keet = e),
(this.local = e.local),
(this.store = e.store),
(this.server = new a(this.store.session(), {
resolve: this._resolveEncryptionKey.bind(this),
})),
(this._activeDownloads = new Map()),
(this._onerror = t.onerror || l),
this.ready().catch(this._onerror))
}
_open() {
return this.server.listen()
}
_close() {
return this.server.close()
}
async _resolveEncryptionKey(e, { blob: t, filename: a, drive: i, version: r }) {
const l = i ? "@local/drive-files" : "@local/blob-files",
o = {
pointer: {
externalDrive: { key: i, path: a && a.startsWith("/") ? a.slice(1) : a, version: r },
externalBlob: { key: e, blob: t },
},
},
n = await this.local.get(l, { file: o })
return (
n && (await this.keet.internalBlobsActiveInRoom(n.roomKey)),
n ? { key: e, encryptionKey: n.encryptionKey } : null
)
}
_getLinkOptions(e) {
if (!e.pointer?.externalBlob && !e.pointer?.externalDrive)
throw new Error("Invalid file pointer")
return e.pointer.externalBlob
? {
collection: "@local/blob-files",
key: e.pointer.externalBlob.key,
opts: { blob: e.pointer.externalBlob.blob, type: e.metadata?.mimetype },
}
: {
collection: "@local/drive-files",
key: e.pointer.externalDrive.key,
opts: {
filename: e.pointer.externalDrive.path,
version: e.pointer.externalDrive.version,
},
}
}
async _clearAllFiles(e, t, a, { leaving: i = !1 } = {}) {
const r = t ? "@local/blob-files" : "@local/drive-files",
o = t ? "@local/blob-files-by-room" : "@local/drive-files-by-room",
n = []
for await (const t of e.find(o, { gte: { roomKey: a }, lte: { roomKey: a } })) {
if (t.cleared) continue
const { key: a, opts: o } = this._getLinkOptions(t.file),
s = this.server.clear(a, o)
;(s.catch(l), n.push(s), i ? await e.delete(r, t) : await e.insert(r, { ...t, cleared: !0 }))
}
return (await Promise.allSettled(n))
.filter((e) => "fulfilled" === e.status)
.reduce((e, t) => e + t.value, 0)
}
async _findAndClearOldFiles(e, t) {
const a = []
for await (const i of this.local.find(e, { lte: { lastAccessed: t } })) i.cleared || a.push(i)
const i = []
for (const e of a) {
const { key: t, opts: a } = this._getLinkOptions(e.file),
r = this.server.clear(t, a)
;(r.catch(l), i.push(r))
}
return {
recordsToClear: a,
bytesCleared: (await Promise.allSettled(i))
.filter((e) => "fulfilled" === e.status)
.reduce((e, t) => e + t.value, 0),
}
}
async _createFile(e, t, a) {
const i = { metadata: a, pointer: { externalBlob: { key: e.externalBlobs.key, blob: t } } }
i.metadata?.userdata && (i.metadata.userdata = c.from(i.metadata.userdata, "base64"))
const r = await this.keet.internalEnterLocalTransaction()
try {
const t = {
file: i,
roomKey: e.key,
encryptionKey: e.encryptionKey,
cleared: !1,
own: !0,
lastAccessed: Date.now(),
}
;(await r.insert("@local/blob-files", t), await this._updateFileStats(r, t), await r.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
return i
}
async saveFile(a, i, r) {
this.opened || (await this.ready())
const l = e.basename(i),
o = this.keet.internalOpenExistingRoom(a)
try {
await this.keet.internalBlobsActiveInRoom(a)
const e = t.createReadStream(i),
s = o.externalBlobs.createWriteStream(r)
return (await n(e, s), this._createFile(o, s.id, { name: l, ...r }))
} finally {
this.keet.internalReleaseRoom(o)
}
}
subscribeAddFile(a, i, r = {}) {
const o = e.basename(i),
n = this,
c = new s()
return (
t.stat(i, (e, s) => {
if (c.destroyed) return
if (e) return c.destroy(e)
let d = 0
const y = s.size,
h = t.createReadStream(i),
p = n.keet.internalOpenExistingRoom(a),
w = p.externalBlobs.createWriteStream(r)
this.keet.internalBlobsActiveInRoom(a).catch(l)
let b = setInterval(() => {
c.destroyed || (d < y && c.push({ bytesAdded: d, byteLength: y, result: null }))
}, r?.debounceTime ?? 500)
;(h.pipe(w),
h.on("data", (e) => {
d += e.byteLength
}),
h.on("error", (e) => c.destroy(e)),
w.on("close", async () => {
if ((b && clearInterval(b), (b = null), !c.destroyed))
try {
const e = await n._createFile(p, w.id, { name: o, ...r })
;(c.push({ bytesAdded: d, byteLength: y, result: e }), c.push(null))
} catch (e) {
c.destroy(e)
}
}),
c.on("error", (e) => {
h.destroy(e)
}),
c.on("close", () => {
;(b && clearInterval(b),
(b = null),
h.destroy(),
w.destroy(),
n.keet.internalReleaseRoom(p))
}),
r.onFileStat && r.onFileStat())
}),
c
)
}
async saveFileBlob(e, t, a) {
this.opened || (await this.ready())
const i = this.keet.internalOpenExistingRoom(e)
try {
await this.keet.internalBlobsActiveInRoom(e)
const r = await i.externalBlobs.put(t)
return this._createFile(i, r, a)
} finally {
this.keet.internalReleaseRoom(i)
}
}
async clearAllFiles(e, t) {
let a = 0,
i = 0
const r = await this.keet.internalEnterLocalTransaction()
try {
;((a = await this._clearAllFiles(r, !0, e, t)),
(i = await this._clearAllFiles(r, !1, e, t)),
await r.insert("@local/file-stats", { roomKey: e, total: 0 }),
await r.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
return { bytesCleared: a + i }
}
async clearOldFiles() {
const e = Date.now() - d,
t = new Set(),
a = await this._findAndClearOldFiles("@local/blob-files-by-last-accessed", e),
i = await this._findAndClearOldFiles("@local/drive-files-by-last-accessed", e),
r = await this.keet.internalEnterLocalTransaction()
try {
for (const e of a.recordsToClear)
(await r.insert("@local/blob-files", { ...e, cleared: !0 }),
await this._updateFileStats(r, { ...e, cleared: !0 }),
t.add(o.encode(e.roomKey)))
for (const e of i.recordsToClear)
(await r.insert("@local/drive-files", { ...e, cleared: !0 }),
await this._updateFileStats(r, { ...e, cleared: !0 }),
t.add(o.encode(e.roomKey)))
;(await r.insert("@local/file-gc-info", { lastCleared: Date.now() }), await r.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
for (const e of t) this.keet.emit("local-files-changed", e)
return { bytesCleared: a.bytesCleared + i.bytesCleared }
}
async _ensureLocalFile(e, t, a) {
const i = await this.keet.internalEnterLocalTransaction()
try {
await this.keet.internalBlobsActiveInRoom(t)
const r = await i.get(e, { roomKey: t, file: a })
let l = r?.encryptionKey
if (!l) {
const e = await i.get("@local/rooms", { roomKey: t })
l = e?.encryptionKey
}
const n = {
...r,
cleared: r?.cleared ?? !1,
roomKey: t,
encryptionKey: l,
file: a,
lastAccessed: Date.now(),
}
return (
await i.insert(e, n),
r || (await this._updateFileStats(i, n)),
await i.flush(),
r || this.keet.emit("local-files-changed", o.encode(t)),
n
)
} finally {
await this.keet.internalExitLocalTransaction()
}
}
async _loadBlob(e, t) {
const a = this.keet.internalOpenExistingRoom(e)
try {
return (await a.ready(), await a.blobs.get(t))
} finally {
this.keet.internalReleaseRoom(a)
}
}
async getFileLink(e, t) {
if ((this.opened || (await this.ready()), t.pointer.blob)) {
const a = await this._loadBlob(e, t.pointer.blob)
return this.server.getBlobLink(a, { type: t.metadata?.mimetype })
}
const { collection: a, key: i, opts: r } = this._getLinkOptions(t)
return (await this._ensureLocalFile(a, e, t), this.server.getLink(i, r))
}
async getFileEntry(e, t, a = {}) {
if (!t) return null
this.opened || (await this.ready())
const i = { roomKey: e, encryptionKey: null, cleared: !1, httpLink: null }
if (!t.pointer.externalBlob && !t.pointer.externalDrive && !t.pointer.blob) return i
if (t.pointer.blob) {
const r = a.loadBlob
? await a.loadBlob(t.pointer.blob)
: await this._loadBlob(e, t.pointer.blob)
return ((i.httpLink = this.server.getBlobLink(r, { type: t.metadata?.mimetype })), i)
}
const { collection: r, key: l, opts: o } = this._getLinkOptions(t)
return { ...(await this._ensureLocalFile(r, e, t)), httpLink: this.server.getLink(l, o) }
}
async getFileEntryWithPreview(e, t, a = {}) {
this.opened || (await this.ready())
const i = await this.getFileEntry(e, t, a)
if (!t.preview) return i
const [r, l, o] = await Promise.all([
this.getFileEntry(e, t.preview.small, a),
this.getFileEntry(e, t.preview.medium, a),
this.getFileEntry(e, t.preview.large, a),
])
return ((i.preview = { small: r, medium: l, large: o }), i)
}
async getFileCleared(e, t) {
const { collection: a } = this._getLinkOptions(t),
i = await this.local.get(a, { file: t })
return !!i && i.cleared
}
async getLocalFiles(e) {
return (
await Promise.all([
this.local
.find("@local/blob-files-by-room", { gte: { roomKey: e }, lte: { roomKey: e } })
.toArray(),
this.local
.find("@local/drive-files-by-room", { gte: { roomKey: e }, lte: { roomKey: e } })
.toArray(),
])
).flat()
}
async clearFile(e, t) {
if (!t?.pointer?.externalBlob && !t?.pointer?.externalDrive) return
const { collection: a, key: i, opts: r } = this._getLinkOptions(t),
l = await this.keet.internalEnterLocalTransaction()
try {
const o = await l.get(a, { roomKey: e, file: t })
if (!o || o.cleared) return
await this.server.clear(i, r)
const n = { roomKey: e, file: t, lastAccessed: o.lastAccessed ?? Date.now(), cleared: !0 }
;(await l.insert(a, n), await this._updateFileStats(l, n), await l.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
}
async unclearFile(e, t) {
const { collection: a } = this._getLinkOptions(t),
i = await this.keet.internalEnterLocalTransaction()
try {
const r = await i.get(a, { roomKey: e, file: t })
if (!r || !r.cleared) return
const l = { roomKey: e, file: t, lastAccessed: Date.now(), cleared: !1 }
;(await i.insert(a, l), await this._updateFileStats(i, l), await i.flush())
} finally {
await this.keet.internalExitLocalTransaction()
}
}
async startDownload(e, t) {
this.opened || (await this.ready())
const { key: a, opts: i } = this._getLinkOptions(t),
r = this.server.getLink(a, { ...i, url: !1 }),
l = this.server.download(a, i)
return (
this._activeDownloads.set(r, l),
l.done().then(() => {
this._activeDownloads.get(r) === l && this._activeDownloads.delete(r)
}),
r
)
}
async cancelDownload(e, t) {
this.opened || (await this.ready())
const a = this._activeDownloads.get(t)
t &&
a &&
(await a.close().finally(() => {
this._activeDownloads.get(t) === a && this._activeDownloads.delete(t)
}))
}
async readFileStream(e, t) {
if (!t.pointer.externalBlob) throw new Error("Only external blob files support streaming")
this.opened || (await this.ready())
const a = t.pointer.externalBlob.blob,
r = t.pointer.externalBlob.key,
{ collection: o } = this._getLinkOptions(t),
n = await this._ensureLocalFile(o, e, t),
s = this.store.get({ key: r, encryptionKey: n.encryptionKey, encrypted: !0, active: !1 })
await s.ready()
const c = new i(s).createReadStream(a)
return (c.once("close", () => s.close().catch(l)), c)
}
async saveToDisk(e, a, i) {
this.opened || (await this.ready())
const r = await this.readFileStream(e, a),
l = t.createWriteStream(i)
await n(r, l)
}
async refreshLinks(e) {
this.opened || (await this.ready())
const t = []
for (const a of e) t.push(this.server.refreshLink(a))
return t
}
async monitor(e, t) {
this.opened || (await this.ready())
const { key: a, opts: i } = this._getLinkOptions(t)
return this.server.monitor(a, i)
}
async _updateFileStats(e, { roomKey: t, file: a, cleared: i }) {
const r = a.pointer.externalBlob?.blob.byteLength
if (!r) return
let l = await e.get("@local/file-stats", { roomKey: t })
l || (l = { roomKey: t, total: 0 })
const o = i ? -r : r
;((l.total = Math.max(l.total + o, 0)), await e.insert("@local/file-stats", l))
}
}