const e = require("path"),
t = require("fs"),
a = require("hypercore-blob-server"),
i = require("ready-resource"),
r = require("safety-catch"),
l = require("hypercore-id-encoding"),
{ pipelinePromise: o, Readable: n } = require("streamx"),
s = require("b4a"),
c = 6048e5
module.exports = class extends i {
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 || r),
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 l = 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(r), n.push(s), i ? await e.delete(l, t) : await e.insert(l, { ...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),
l = this.server.clear(t, a)
;(l.catch(r), i.push(l))
}
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 = s.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),
n = this.keet.internalOpenExistingRoom(a)
try {
await this.keet.internalBlobsActiveInRoom(a)
const e = t.createReadStream(i),
s = n.externalBlobs.createWriteStream(r)
return (await o(e, s), this._createFile(n, s.id, { name: l, ...r }))
} finally {
this.keet.internalReleaseRoom(n)
}
}
subscribeAddFile(a, i, l = {}) {
const o = e.basename(i),
s = this,
c = new n()
return (
t.stat(i, (e, n) => {
if (c.destroyed) return
if (e) return c.destroy(e)
let d = 0
const y = n.size,
h = t.createReadStream(i),
w = s.keet.internalOpenExistingRoom(a),
p = w.externalBlobs.createWriteStream(l)
this.keet.internalBlobsActiveInRoom(a).catch(r)
let f = setInterval(() => {
c.destroyed || (d < y && c.push({ bytesAdded: d, byteLength: y, result: null }))
}, l?.debounceTime ?? 500)
;(h.pipe(p),
h.on("data", (e) => {
d += e.byteLength
}),
h.on("error", (e) => c.destroy(e)),
p.on("close", async () => {
if ((f && clearInterval(f), (f = null), !c.destroyed))
try {
const e = await s._createFile(w, p.id, { name: o, ...l })
;(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", () => {
;(f && clearInterval(f),
(f = null),
h.destroy(),
p.destroy(),
s.keet.internalReleaseRoom(w))
}),
l.onFileStat && l.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() - c,
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(l.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(l.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 o = r?.encryptionKey
if (!o) {
const e = await i.get("@local/rooms", { roomKey: t })
o = e?.encryptionKey
}
const n = {
...r,
cleared: r?.cleared ?? !1,
roomKey: t,
encryptionKey: o,
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", l.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 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))
}
}