1 // From https://github.com/MinEduTDF/idb-chunk-store
2 // We use temporary IndexDB (all data are removed on destroy) to avoid RAM issues
3 // Thanks @santiagogil and @Feross
5 import Dexie from 'dexie'
6 import { EventEmitter } from 'events'
7 import { logger } from '@root-helpers/logger'
9 class ChunkDatabase extends Dexie {
10 chunks: Dexie.Table<{ id: number, buf: Buffer }, number>
12 constructor (dbname: string) {
15 this.version(1).stores({
21 class ExpirationDatabase extends Dexie {
22 databases: Dexie.Table<{ name: string, expiration: number }, number>
25 super('webtorrent-expiration')
27 this.version(1).stores({
28 databases: 'name,expiration'
33 export class PeertubeChunkStore extends EventEmitter {
34 private static readonly BUFFERING_PUT_MS = 1000
35 private static readonly CLEANER_INTERVAL_MS = 1000 * 60 // 1 minute
36 private static readonly CLEANER_EXPIRATION_MS = 1000 * 60 * 5 // 5 minutes
40 private pendingPut: { id: number, buf: Buffer, cb: (err?: Error) => void }[] = []
41 // If the store is full
42 private memoryChunks: { [ id: number ]: Buffer | true } = {}
43 private databaseName: string
44 private putBulkTimeout: any
45 private cleanerInterval: any
46 private db: ChunkDatabase
47 private expirationDB: ExpirationDatabase
48 private readonly length: number
49 private readonly lastChunkLength: number
50 private readonly lastChunkIndex: number
52 constructor (chunkLength: number, opts: any) {
55 this.databaseName = 'webtorrent-chunks-'
58 if (opts.torrent?.infoHash) this.databaseName += opts.torrent.infoHash
59 else this.databaseName += '-default'
61 this.setMaxListeners(100)
63 this.chunkLength = Number(chunkLength)
64 if (!this.chunkLength) throw new Error('First argument must be a chunk length')
66 this.length = Number(opts.length) || Infinity
68 if (this.length !== Infinity) {
69 this.lastChunkLength = (this.length % this.chunkLength) || this.chunkLength
70 this.lastChunkIndex = Math.ceil(this.length / this.chunkLength) - 1
73 this.db = new ChunkDatabase(this.databaseName)
74 // Track databases that expired
75 this.expirationDB = new ExpirationDatabase()
80 put (index: number, buf: Buffer, cb: (err?: Error) => void) {
81 const isLastChunk = (index === this.lastChunkIndex)
82 if (isLastChunk && buf.length !== this.lastChunkLength) {
83 return this.nextTick(cb, new Error('Last chunk length must be ' + this.lastChunkLength))
85 if (!isLastChunk && buf.length !== this.chunkLength) {
86 return this.nextTick(cb, new Error('Chunk length must be ' + this.chunkLength))
89 // Specify we have this chunk
90 this.memoryChunks[index] = true
92 // Add it to the pending put
93 this.pendingPut.push({ id: index, buf, cb })
94 // If it's already planned, return
95 if (this.putBulkTimeout) return
97 // Plan a future bulk insert
98 this.putBulkTimeout = setTimeout(async () => {
99 const processing = this.pendingPut
101 this.putBulkTimeout = undefined
104 await this.db.transaction('rw', this.db.chunks, () => {
105 return this.db.chunks.bulkPut(processing.map(p => ({ id: p.id, buf: p.buf })))
108 logger.info('Cannot bulk insert chunks. Store them in memory.', err)
110 processing.forEach(p => {
111 this.memoryChunks[p.id] = p.buf
114 processing.forEach(p => p.cb())
116 }, PeertubeChunkStore.BUFFERING_PUT_MS)
119 get (index: number, opts: any, cb: (err?: Error, buf?: Buffer) => void): void {
120 if (typeof opts === 'function') return this.get(index, null, opts)
122 // IndexDB could be slow, use our memory index first
123 const memoryChunk = this.memoryChunks[index]
124 if (memoryChunk === undefined) {
125 const err = new Error('Chunk not found') as any
126 err['notFound'] = true
128 return process.nextTick(() => cb(err))
132 if (memoryChunk !== true) return cb(null, memoryChunk)
135 this.db.transaction('r', this.db.chunks, async () => {
136 const result = await this.db.chunks.get({ id: index })
137 if (result === undefined) return cb(null, Buffer.alloc(0))
139 const buf = result.buf
140 if (!opts) return this.nextTick(cb, null, buf)
142 const offset = opts.offset || 0
143 const len = opts.length || (buf.length - offset)
144 return cb(null, buf.slice(offset, len + offset))
152 close (cb: (err?: Error) => void) {
153 return this.destroy(cb)
156 async destroy (cb: (err?: Error) => void) {
158 if (this.pendingPut) {
159 clearTimeout(this.putBulkTimeout)
160 this.pendingPut = null
162 if (this.cleanerInterval) {
163 clearInterval(this.cleanerInterval)
164 this.cleanerInterval = null
170 await this.dropDatabase(this.databaseName)
173 if (this.expirationDB) {
174 this.expirationDB.close()
175 this.expirationDB = null
180 logger.error('Cannot destroy peertube chunk store.', err)
185 private runCleaner () {
186 this.checkExpiration()
188 this.cleanerInterval = setInterval(() => {
189 this.checkExpiration()
190 }, PeertubeChunkStore.CLEANER_INTERVAL_MS)
193 private async checkExpiration () {
194 let databasesToDeleteInfo: { name: string }[] = []
197 await this.expirationDB.transaction('rw', this.expirationDB.databases, async () => {
198 // Update our database expiration since we are alive
199 await this.expirationDB.databases.put({
200 name: this.databaseName,
201 expiration: new Date().getTime() + PeertubeChunkStore.CLEANER_EXPIRATION_MS
204 const now = new Date().getTime()
205 databasesToDeleteInfo = await this.expirationDB.databases.where('expiration').below(now).toArray()
208 logger.error('Cannot update expiration of fetch expired databases.', err)
211 for (const databaseToDeleteInfo of databasesToDeleteInfo) {
212 await this.dropDatabase(databaseToDeleteInfo.name)
216 private async dropDatabase (databaseName: string) {
217 const dbToDelete = new ChunkDatabase(databaseName)
218 logger.info(`Destroying IndexDB database ${databaseName}`)
221 await dbToDelete.delete()
223 await this.expirationDB.transaction('rw', this.expirationDB.databases, () => {
224 return this.expirationDB.databases.where({ name: databaseName }).delete()
227 logger.error(`Cannot delete ${databaseName}.`, err)
231 private nextTick <T> (cb: (err?: Error, val?: T) => void, err: Error, val?: T) {
232 process.nextTick(() => cb(err, val), undefined)