hyper-note / lib / media-store.js

import Hyperblobs from 'hyperblobs'
import Hyperbee from 'hyperbee'
import crypto from 'bare-crypto'
import b4a from 'b4a'

export class MediaStore {
  constructor (corestore, opts = {}) {
    this._store = corestore.namespace('media-store')
    this._blobCore = this._store.get({ name: 'blobs' })
    this._blobs = new Hyperblobs(this._blobCore)
    this._indexCore = this._store.get({ name: 'index' })
    this._index = new Hyperbee(this._indexCore, {
      keyEncoding: 'utf-8',
      valueEncoding: 'json'
    })

    this._orphanQueue = new Map()
    this._graceMs = opts.graceMs != null ? opts.graceMs : 7 * 24 * 60 * 60 * 1000
    this.isReferencedAnywhere = opts.isReferencedAnywhere || (async () => false)
  }

  async ready () {
    await this._blobCore.ready()
    await this._index.ready()
  }

  _hash (buffer) {
    const digest = crypto.createHash('sha256').update(buffer).digest()
    return b4a.toString(digest, 'hex')
  }

  async put (buffer, mimeType) {
    const hash = this._hash(buffer)
    const existing = await this._index.get('blob:' + hash)
    if (existing) return hash

    const blobId = await this._blobs.put(buffer)
    await this._index.put('blob:' + hash, {
      blobId,
      size: buffer.length,
      mimeType: mimeType || 'application/octet-stream',
      createdAt: Date.now()
    })
    return hash
  }

  async get (hash) {
    const entry = await this._index.get('blob:' + hash)
    if (!entry) return null
    return this._blobs.get(entry.value.blobId)
  }

  async has (hash) {
    const entry = await this._index.get('blob:' + hash)
    return entry !== null
  }

  async getInfo (hash) {
    const entry = await this._index.get('blob:' + hash)
    return entry ? entry.value : null
  }

  async clear (hash) {
    const entry = await this._index.get('blob:' + hash)
    if (!entry) return 0

    const blobId = entry.value.blobId
    const result = await this._blobs.clear(blobId, { diff: true })
    await this._index.del('blob:' + hash)

    return result ? result.blocks : 0
  }

  getStream (hash, range) {
    const self = this
    return (async function * () {
      const entry = await self._index.get('blob:' + hash)
      if (!entry) return

      const blobId = entry.value.blobId
      const opts = {}
      if (range) {
        if (range.start !== undefined) opts.start = range.start
        if (range.end !== undefined) opts.end = range.end
        else if (range.length !== undefined) opts.length = range.length
      }

      const stream = self._blobs.createReadStream(blobId, opts)
      for await (const chunk of stream) {
        yield chunk
      }
    })()
  }

  scheduleOrphan (hash) {
    if (this._orphanQueue.has(hash)) return
    const timeoutId = setTimeout(() => this._clearIfStillOrphan(hash), this._graceMs)
    this._orphanQueue.set(hash, timeoutId)
  }

  cancelOrphan (hash) {
    const timeoutId = this._orphanQueue.get(hash)
    if (timeoutId) {
      clearTimeout(timeoutId)
      this._orphanQueue.delete(hash)
    }
  }

  async _clearIfStillOrphan (hash) {
    this._orphanQueue.delete(hash)
    const stillReferenced = await this.isReferencedAnywhere(hash)
    if (stillReferenced) return

    // Check if this is a video with associated HLS assets
    await this._cascadeClearVideo(hash)

    // Clear the original blob
    await this.clear(hash)
  }

  /**
   * If hash has a video index, cascade-clear all associated blobs:
   * poster, manifest, level playlists, and all segments.
   */
  async _cascadeClearVideo (hash) {
    const entry = await this._index.get('video:' + hash)
    if (!entry) return

    const video = entry.value
    let cleared = 0

    // Clear poster
    if (video.poster) {
      await this.clear(video.poster).catch(() => {})
      cleared++
    }

    // Clear manifest
    if (video.manifest) {
      await this.clear(video.manifest).catch(() => {})
      cleared++
    }

    // Clear all level playlists and segments
    if (video.levels) {
      for (const level of video.levels) {
        if (level.m3u8) {
          await this.clear(level.m3u8).catch(() => {})
          cleared++
        }
        if (level.segments) {
          for (const segHash of level.segments) {
            // segments may be strings (hashes) or objects with .hash
            const h = typeof segHash === 'string' ? segHash : segHash.hash
            if (h) {
              await this.clear(h).catch(() => {})
              cleared++
            }
          }
        }
      }
    }

    // Remove video index entry
    await this._index.del('video:' + hash).catch(() => {})

    if (cleared > 0) {
      console.log('[GC] Cascade cleared ' + cleared + ' video assets for ' + hash.substring(0, 16) + '...')
    }
  }

  async listAll () {
    const results = []
    for await (const entry of this._index.createReadStream({ gte: 'blob:', lt: 'blob:\xff' })) {
      results.push({
        hash: entry.key.replace('blob:', ''),
        ...entry.value
      })
    }
    return results
  }
}

export function parseRange (header, totalSize) {
  if (!header || !header.startsWith('bytes=')) return null
  const spec = header.replace('bytes=', '')
  const parts = spec.split('-')

  let start, end

  if (parts[0] === '') {
    const suffix = parseInt(parts[1], 10)
    if (isNaN(suffix)) return null
    start = totalSize - suffix
    end = totalSize - 1
  } else if (parts[1] === '') {
    start = parseInt(parts[0], 10)
    end = totalSize - 1
  } else {
    start = parseInt(parts[0], 10)
    end = parseInt(parts[1], 10)
  }

  if (isNaN(start) || isNaN(end) || start < 0 || end >= totalSize || start > end) return null
  return { start, end }
}

export default MediaStore