DodoTransfer/lib/db.js

'use strict';
const fsp = require('fs-promise');
const path = require('path');
const debug = require('debug')('psitransfer:db');
const config = require('../config');

module.exports = class DB {

  constructor(uploadDir, store) {
    this.initialized = false;
    this.db = {};
    // fids of one-time files with a running download
    this.claims = new Set();

    this.uploadDir = uploadDir;
    this.store = store;

    // delete expired files
    const gc = () => {
      const now = Date.now();
      for (const sid of Object.keys(this.db)) {
        for (const f of this.db[sid].slice()) {
          if(this.expiresAt(f) <= now) {
            debug(`Expired ${sid}++${f.key}`);
            this.remove(sid, f.key).catch(e => console.error(e));
          }
        }
      }
    };
    setInterval(gc, 60 * 1000).unref();

  }


  /**
   * Point in time (ms) a file is deleted. The garbage collector and the
   * expiry shown to users share this function so they can not differ.
   * One-time downloads that are never fetched expire after maxAge.
   */
  expiresAt(file) {
    const created = +file.metadata.createdAt;
    const maxExpire = created + config.maxAge * 1000;
    const retention = +file.metadata.retention;
    if(Number.isInteger(retention)) {
      return Math.min(maxExpire, created + retention * 1000);
    }
    return maxExpire;
  }


  init() {
    if(this.initialized) return;
    this.initialized = true;

    try {
      this._sync();
    } catch(e) {
      this.initialized = false;
      e.message = `db initialization failed with error ${e.message}`;
      throw e;
    }
  }


  /**
   * @private
   */
  _sync() {
    fsp.ensureDirSync(this.uploadDir);

    fsp.readdirSync(this.uploadDir).forEach((sid) => {
      // hidden folders (e.g. .settings) are no buckets
      if(sid.startsWith('.')) return;
      this._import(sid);
    });
  }


  /**
   * @private
   */
  _import(sid) {
    const p = path.resolve(this.uploadDir, sid);
    const stat = fsp.statSync(p);
    if(!stat.isDirectory()) return;

    // synchronous on purpose: the index is complete when init() returns
    fsp.readdirSync(p).forEach((key) => {
      if(path.extname(key) !== '') return;
      try {
        const file = this.store.getFilename(`${sid}++${key}`);
        const info = fsp.readJsonSync(file + '.json');
        const fileStat = fsp.statSync(file);
        info.size = fileStat.size;
        info.offset = fileStat.size;
        this.add(sid, key, info);
      } catch(e) {
        console.error(e);
      }
    });
  }


  add(sid, key, data) {
    if(!this.initialized) throw new Error('DB not initialized_');
    if(!this.db[sid]) this.db[sid] = [];
    data.key = key;
    const old = this.db[sid].findIndex(i => i.key === key);
    if(old !== -1) {
      this.db[sid].splice(old, 1, data);
      debug(`Updated ${sid}++${key}`);
    } else {
      this.db[sid].push(data);
      debug(`Added ${sid}++${key}`);
    }
  }


  async remove(sid, key) {
    if(!this.initialized) throw new Error('DB not initialized');
    debug(`Remove ${sid}++${key}`);
    await this.store.del(sid + '++' + key);
    // tolerate concurrent removal of the same file
    const bucket = this.get(sid);
    if(!bucket) return;
    const i = bucket.findIndex(item => item.key === key);
    if(i === -1) return;
    bucket.splice(i, 1);
    if(bucket.length === 0) {
      delete this.db[sid];
      try {
        await fsp.rmdir(path.resolve(this.uploadDir, sid));
      } catch(e) {
        if(e.code !== 'ENOENT' && e.code !== 'ENOTEMPTY') throw e;
      }
    }
  }


  /**
   * Atomically claim a file for a one-time download.
   * Must stay synchronous: no await between check and set.
   * @returns {boolean} false if already claimed or not (anymore) in the db
   */
  claim(sid, key) {
    const fid = `${sid}++${key}`;
    if(this.claims.has(fid)) return false;
    const bucket = this.get(sid);
    if(!bucket || !bucket.some(item => item.key === key)) return false;
    this.claims.add(fid);
    return true;
  }


  release(sid, key) {
    this.claims.delete(`${sid}++${key}`);
  }


  isClaimed(sid, key) {
    return this.claims.has(`${sid}++${key}`);
  }


  async updateLastDownload(sid, key) {
    debug(`Update last download ${sid}++${key}`);
    const data = this.get(sid).find(item => item.key === key);
    if(!data) return;
    data.metadata.lastDownload = Date.now();
    await this.store.update(`${sid}++${key}`, data);
  }


  async updateMetadata(sid, key, data) {
    debug(`Update metadata ${ sid }++${ key }`);
    const file = this.get(sid).find(item => item.key === key);
    if (!file) return;
    file.metadata = { ...file.metadata, ...data };
    await this.store.update(`${ sid }++${ key }`, file);
  }

  async lock(sid) {
    const files = this.get(sid);
    if(!files) return;
    await Promise.all(files.map(async file => {
      await this.updateMetadata(sid, file.key, { bucketLocked: true });
    }));
  }

  isLocked(sid) {
    const files = this.get(sid);
    if(!files) return false;
    // "buckedLocked" is the old (misspelled) name, still found in existing data
    return files.some(file => file.metadata.bucketLocked || file.metadata.buckedLocked);
  }

  /** Remove every file of a bucket. */
  async removeBucket(sid) {
    const files = this.get(sid);
    if(!files) return 0;
    const keys = files.map(f => f.key);
    await Promise.all(keys.map(key => this.remove(sid, key)));
    return keys.length;
  }

  get(sid) {
    return this.db[sid];
  }

  bucketSize(sid) {
    const bucket = this.get(sid);
    if(!bucket) return 0;
    return bucket.reduce((v, file) => v + +file.metadata.uploadLength, 0);
  }

};