mongodb.js

const client = require('mongodb').MongoClient
const ObjectID = require('mongodb').ObjectID
const mongodbUri = require('mongodb-uri')

const Interface = require('./interface')
const config = require('../../config.json')
const flatten = require('flat')

const _ = require('lodash')
const debug = require('debug')('db:mongodb')

class IMongoDB extends Interface {
  constructor () {
    super('mongodb')

    this.connected = false
    this.client = null
    this.dbName = mongodbUri.parse(config.database.mongodb.url).database

    this.connect()

    if (debug.enabled) debug('MongoDB initialized')
  }

  async connect () {
    this.client = await client.connect(config.database.mongodb.url, { poolSize: _.get(config, 'database.mongodb.poolSize', 5) })

    // create indexes
    let db = this.client.db(this.dbName)
    await db.collection('users.bits').createIndex('timestamp')
    await db.collection('users.tips').createIndex('timestamp')
    await db.collection('users').createIndex('username')
    await db.collection('cache').createIndex('key')
    await db.collection('api.stats').createIndex('timestamp')
    await db.collection('stats').createIndex('whenOnline')

    this.connected = true
  }

  async find (table, where) {
    where = where || {}
    if (!_.isNil(where._id)) {
      let regexp = new RegExp('^[0-9a-fA-F]{24}$')
      if (regexp.test(where._id)) where._id = new ObjectID(where._id)
      else return {}
    } else where = flatten(where)
    try {
      let db = this.client.db(this.dbName)
      let items = await db.collection(table).find(where)
      return items.toArray()
    } catch (e) {
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async findOne (table, where) {
    where = where || {}
    if (!_.isNil(where._id)) {
      let regexp = new RegExp('^[0-9a-fA-F]{24}$')
      if (regexp.test(where._id)) where._id = new ObjectID(where._id)
      else return {}
    } else where = flatten(where)

    try {
      let db = this.client.db(this.dbName)
      let item = await db.collection(table).findOne(where)
      return item || {}
    } catch (e) {
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async insert (table, object) {
    if (_.isEmpty(object)) throw Error('Object cannot be empty')
    delete object._id

    try {
      let db = this.client.db(this.dbName)
      let item = await db.collection(table).insert(flatten.unflatten(object))
      return item.ops[0]
    } catch (e) {
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async incrementOne (table, where, object) {
    where = where || {}
    if (!_.isNil(where._id)) where._id = new ObjectID(where._id)
    else where = flatten(where)

    if (_.isEmpty(object)) throw Error('Object to update cannot be empty')
    delete object._id

    const threadHash = await this.waitForThread(table, where)
    try {
      let db = this.client.db(this.dbName)
      let item = await db.collection(table).findAndModify(
        where,
        { _id: 1 },
        // DON'T EVER DELETE flatten ON OBJECT - with flatten object get updated and not replaced
        { $inc: flatten(object) },
        { new: true } // will return updated item
      )
      this.freeThread(table, where, threadHash)
      return item.value
    } catch (e) {
      this.freeThread(table, where, threadHash)
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async increment (table, where, object) {
    where = where || {}
    if (!_.isNil(where._id)) where._id = new ObjectID(where._id)
    else where = flatten(where)

    if (_.isEmpty(object)) throw Error('Object to update cannot be empty')
    delete object._id

    const threadHash = await this.waitForThread(table, where)
    try {
      let db = this.client.db(this.dbName)

      await db.collection(table).update(
        where,
        // DON'T EVER DELETE flatten ON OBJECT - with flatten object get updated and not replaced
        { $inc: flatten(object) }, {
          upsert: true,
          multi: _.isEmpty(where)
        }
      )

      // workaround for return of updated objects
      let items = await db.collection(table).find(where).toArray()
      this.freeThread(table, where, threadHash)
      return items
    } catch (e) {
      this.freeThread(table, where, threadHash)
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async remove (table, where) {
    if (!_.isNil(where._id)) where._id = new ObjectID(where._id)
    else where = flatten(where)

    const threadHash = await this.waitForThread(table, where)
    try {
      let db = this.client.db(this.dbName)
      let result = await db.collection(table).deleteMany(where)
      this.freeThread(table, where, threadHash)
      return result.result.n
    } catch (e) {
      this.freeThread(table, where, threadHash)
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }

  async update (table, where, object) {
    if (_.isEmpty(object)) throw Error('Object to update cannot be empty')

    if (!_.isNil(where._id)) where._id = new ObjectID(where._id)
    else where = flatten(where)

    // remove _id from object
    delete object._id

    const threadHash = await this.waitForThread(table, where)
    if (debug.enabled) debug('update() \n\ttable: %s \n\twhere: %j \n\thash: %s', table, where, threadHash)

    try {
      let db = this.client.db(this.dbName)

      if (_.size(where) === 0) {
        // DON'T EVER DELETE flatten ON OBJECT - with flatten object get updated and not replaced
        await db.collection(table).updateMany({}, { $set: flatten(object, { safe: true }) })
      } else {
        await db.collection(table).updateOne(
          where,
          // DON'T EVER DELETE flatten ON OBJECT - with flatten object get updated and not replaced
          { $set: flatten(object, { safe: true }) }, {
            upsert: _.isNil(where._id)
          }
        )
      }

      // workaround for return of updated objects
      let items = await db.collection(table).find(where).toArray()
      this.freeThread(table, where, threadHash)
      return items.length === 1 ? items[0] : items
    } catch (e) {
      this.freeThread(table, where, threadHash)
      global.log.error(e.stack)
      if (e.message.match(/EPIPE/g)) {
        global.log.error(`Something went wrong with mongodb instance (EPIPE error)`)
        process.exit()
      }
    }
  }
}

module.exports = IMongoDB