client/src/assets/player/shared/webtorrent/peertube-chunk-store.ts

   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
   4
   5 import Dexie from 'dexie'
   6 import { EventEmitter } from 'events'
   7 import { logger } from '@root-helpers/logger'
   8
   9 class ChunkDatabase extends Dexie {
  10   chunks: Dexie.Table<{ id: number, buf: Buffer }, number>
  11
  12   constructor (dbname: string) {
  13     super(dbname)
  14
  15     this.version(1).stores({
  16       chunks: 'id'
  17     })
  18   }
  19 }
  20
  21 class ExpirationDatabase extends Dexie {
  22   databases: Dexie.Table<{ name: string, expiration: number }, number>
  23
  24   constructor () {
  25     super('webtorrent-expiration')
  26
  27     this.version(1).stores({
  28       databases: 'name,expiration'
  29     })
  30   }
  31 }
  32
  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
  37
  38   chunkLength: number
  39
  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
  51
  52   constructor (chunkLength: number, opts: any) {
  53     super()
  54
  55     this.databaseName = 'webtorrent-chunks-'
  56
  57     if (!opts) opts = {}
  58     if (opts.torrent?.infoHash) this.databaseName += opts.torrent.infoHash
  59     else this.databaseName += '-default'
  60
  61     this.setMaxListeners(100)
  62
  63     this.chunkLength = Number(chunkLength)
  64     if (!this.chunkLength) throw new Error('First argument must be a chunk length')
  65
  66     this.length = Number(opts.length) || Infinity
  67
  68     if (this.length !== Infinity) {
  69       this.lastChunkLength = (this.length % this.chunkLength) || this.chunkLength
  70       this.lastChunkIndex = Math.ceil(this.length / this.chunkLength) - 1
  71     }
  72
  73     this.db = new ChunkDatabase(this.databaseName)
  74     // Track databases that expired
  75     this.expirationDB = new ExpirationDatabase()
  76
  77     this.runCleaner()
  78   }
  79
  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))
  84     }
  85     if (!isLastChunk && buf.length !== this.chunkLength) {
  86       return this.nextTick(cb, new Error('Chunk length must be ' + this.chunkLength))
  87     }
  88
  89     // Specify we have this chunk
  90     this.memoryChunks[index] = true
  91
  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
  96
  97     // Plan a future bulk insert
  98     this.putBulkTimeout = setTimeout(async () => {
  99       const processing = this.pendingPut
 100       this.pendingPut = []
 101       this.putBulkTimeout = undefined
 102
 103       try {
 104         await this.db.transaction('rw', this.db.chunks, () => {
 105           return this.db.chunks.bulkPut(processing.map(p => ({ id: p.id, buf: p.buf })))
 106         })
 107       } catch (err) {
 108         logger.info('Cannot bulk insert chunks. Store them in memory.', err)
 109
 110         processing.forEach(p => {
 111           this.memoryChunks[p.id] = p.buf
 112         })
 113       } finally {
 114         processing.forEach(p => p.cb())
 115       }
 116     }, PeertubeChunkStore.BUFFERING_PUT_MS)
 117   }
 118
 119   get (index: number, opts: any, cb: (err?: Error, buf?: Buffer) => void): void {
 120     if (typeof opts === 'function') return this.get(index, null, opts)
 121
 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
 127
 128       return process.nextTick(() => cb(err))
 129     }
 130
 131     // Chunk in memory
 132     if (memoryChunk !== true) return cb(null, memoryChunk)
 133
 134     // Chunk in store
 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))
 138
 139       const buf = result.buf
 140       if (!opts) return this.nextTick(cb, null, buf)
 141
 142       const offset = opts.offset || 0
 143       const len = opts.length || (buf.length - offset)
 144       return cb(null, buf.slice(offset, len + offset))
 145     })
 146     .catch(err => {
 147       logger.error(err)
 148       return cb(err)
 149     })
 150   }
 151
 152   close (cb: (err?: Error) => void) {
 153     return this.destroy(cb)
 154   }
 155
 156   async destroy (cb: (err?: Error) => void) {
 157     try {
 158       if (this.pendingPut) {
 159         clearTimeout(this.putBulkTimeout)
 160         this.pendingPut = null
 161       }
 162       if (this.cleanerInterval) {
 163         clearInterval(this.cleanerInterval)
 164         this.cleanerInterval = null
 165       }
 166
 167       if (this.db) {
 168         this.db.close()
 169
 170         await this.dropDatabase(this.databaseName)
 171       }
 172
 173       if (this.expirationDB) {
 174         this.expirationDB.close()
 175         this.expirationDB = null
 176       }
 177
 178       return cb()
 179     } catch (err) {
 180       logger.error('Cannot destroy peertube chunk store.', err)
 181       return cb(err)
 182     }
 183   }
 184
 185   private runCleaner () {
 186     this.checkExpiration()
 187
 188     this.cleanerInterval = setInterval(() => {
 189       this.checkExpiration()
 190     }, PeertubeChunkStore.CLEANER_INTERVAL_MS)
 191   }
 192
 193   private async checkExpiration () {
 194     let databasesToDeleteInfo: { name: string }[] = []
 195
 196     try {
 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
 202         })
 203
 204         const now = new Date().getTime()
 205         databasesToDeleteInfo = await this.expirationDB.databases.where('expiration').below(now).toArray()
 206       })
 207     } catch (err) {
 208       logger.error('Cannot update expiration of fetch expired databases.', err)
 209     }
 210
 211     for (const databaseToDeleteInfo of databasesToDeleteInfo) {
 212       await this.dropDatabase(databaseToDeleteInfo.name)
 213     }
 214   }
 215
 216   private async dropDatabase (databaseName: string) {
 217     const dbToDelete = new ChunkDatabase(databaseName)
 218     logger.info(`Destroying IndexDB database ${databaseName}`)
 219
 220     try {
 221       await dbToDelete.delete()
 222
 223       await this.expirationDB.transaction('rw', this.expirationDB.databases, () => {
 224         return this.expirationDB.databases.where({ name: databaseName }).delete()
 225       })
 226     } catch (err) {
 227       logger.error(`Cannot delete ${databaseName}.`, err)
 228     }
 229   }
 230
 231   private nextTick <T> (cb: (err?: Error, val?: T) => void, err: Error, val?: T) {
 232     process.nextTick(() => cb(err, val), undefined)
 233   }
 234 }