<commit_before>import Bluebird from 'bluebird'
import filter from 'lodash.filter'
import forEach from 'lodash.foreach'
import includes from 'lodash.includes'
import isEmpty from 'lodash.isempty'
import isString from 'lodash.isstring'
import map from 'lodash.map'
import proxyRequest from 'proxy-http-request'
import XoCollection from 'xo-collection'
import XoUniqueIndex from 'xo-collection/unique-index'
// import XoView from 'xo-collection/view'
import {createClient as createRedisClient} from 'then-redis'
import {EventEmitter} from 'events'
import {parse as parseUrl} from 'url'

import * as xapiObjectsToXo from './xapi-objects-to-xo'
import Connection from './connection'
import User, {Users} from './models/user'
import Xapi from './xapi'
import {Acls} from './models/acl'
import {autobind} from './decorators'
import {generateToken} from './utils'
import {Groups} from './models/group'
import {JsonRpcError, NoSuchObject} from './api-errors'
import {ModelAlreadyExists} from './collection'
import {Servers} from './models/server'
import {Tokens} from './models/token'

// ===================================================================

class NoSuchAuthenticationToken extends NoSuchObject {
  constructor (id) {
    super(id, 'authentication token')
  }
}

class NoSuchGroup extends NoSuchObject {
  constructor (id) {
    super(id, 'group')
  }
}

class NoSuchUser extends NoSuchObject {
  constructor (id) {
    super(id, 'user')
  }
}

class NoSuchXenServer extends NoSuchObject {
  constructor (id) {
    super(id, 'xen server')
  }
}

// ===================================================================

export default class Xo extends EventEmitter {
  constructor () {
    super()

    this._objects = new XoCollection()
    this._objects.createIndex('byRef', new XoUniqueIndex('ref'))

    // These will be initialized in start()
    //
    // TODO: remove and put everything in the `_objects` collection.
    this._acls = null
    this._groups = null
    this._servers = null
    this._tokens = null
    this._users = null
    this._UUIDsToKeys = null

    // Connections to Xen servers.
    this._xapis = Object.create(null)

    // Connections to users.
    this._nextConId = 0
    this._connections = Object.create(null)

    this._httpRequestWatchers = Object.create(null)

    // TODO: remove when no longer necessary.
    this._proxyRequests = Object.create(null)

    this._authenticationProviders = new Set()

    this._watchObjects()
  }

  // -----------------------------------------------------------------

  async start (config) {
    // Connects to Redis.
    const redis = createRedisClient(config.redis && config.redis.uri)

    // Creates persistent collections.
    this._acls = new Acls({
      connection: redis,
      prefix: 'xo:acl',
      indexes: ['subject', 'object']
    })
    this._groups = new Groups({
      connection: redis,
      prefix: 'xo:group'
    })
    this._servers = new Servers({
      connection: redis,
      prefix: 'xo:server',
      indexes: ['host']
    })
    this._tokens = new Tokens({
      connection: redis,
      prefix: 'xo:token',
      indexes: ['user_id']
    })
    this._users = new Users({
      connection: redis,
      prefix: 'xo:user',
      indexes: ['email']
    })

    // Proxies tokens/users related events to XO and removes tokens
    // when their related user is removed.
    this._tokens.on('remove', ids => {
      for (let id of ids) {
        this.emit(`token.revoked:${id}`)
      }
    })
    this._users.on('remove', async function (ids) {
      for (let id of ids) {
        this.emit(`user.revoked:${id}`)
        const tokens = await this._tokens.get({ user_id: id })
        for (let token of tokens) {
          this._tokens.remove(token.id)
        }
      }
    }.bind(this))

    // Connects to existing servers.
    for (let server of await this._servers.get()) {
      this.connectXenServer(server.id).catch(error => {
        console.error(
          `[WARN] ${server.host}:`,
          error[0] || error.stack || error.code || error
        )
      })
    }
  }

  // -----------------------------------------------------------------

  async addAcl (subjectId, objectId, action) {
    try {
      await this._acls.create(subjectId, objectId, action)
    } catch (error) {
      if (!(error instanceof ModelAlreadyExists)) {
        throw error
      }
    }
  }

  async removeAcl (subjectId, objectId, action) {
    await this._acls.delete(subjectId, objectId, action)
  }

  async getAclsForUser (userId) {
    const subjects = (await this.getUser(userId)).groups.concat(userId)

    const acls = []
    const pushAcls = (function (push) {
      return function (entries) {
        push.apply(acls, entries)
      }
    })(acls.push)

    const {_acls: collection} = this
    await Promise.all(map(
      subjects,
      subject => collection.get({subject}).then(pushAcls)
    ))

    return acls
  }

  // TODO: remove when new collection.
  async getAllAcls () {
    return this._acls.get()
  }

  async hasPermission (userId, objectId, permission) {
    const user = await this.getUser(userId)

    // Special case for super XO administrators.
    //
    // TODO: restore when necessary, for now it is already implemented
    // in resolveParams().
    // if (user.permission === 'admin') {
    //   return true
    // }

    const subjects = user.groups.concat(userId)
    let actions = (await this.getRolesForPermission(permission)).concat(permission)

    const promises = []
    {
      const {_acls: acls} = this
      const throwIfFail = function (success) {
        if (!success) {
          // We don't care about an error object.
          /* eslint no-throw-literal: 0 */
          throw null
        }
      }
      forEach(subjects, subject => {
        forEach(actions, action => {
          promises.push(
            acls.aclExists(subject, objectId, action).then(throwIfFail)
          )
        })
      })
    }

    try {
      await Bluebird.any(promises)
      return true
    } catch (_) {
      return false
    }
  }

  // -----------------------------------------------------------------

  async createUser ({email, password, permission}) {
    // TODO: use plain objects
    const user = await this._users.create(email, password, permission)

    return user.properties
  }

  async deleteUser (id) {
    if (!await this._users.remove(id)) {
      throw new NoSuchUser(id)
    }
  }

  async updateUser (id, {email, password, permission}) {
    const user = await this._getUser(id)

    if (email) user.set('email', email)
    if (permission) user.set('permission', permission)
    if (password) {
      await user.setPassword(password)
    }

    await this._users.save(user.properties)
  }

  // Merge this method in getUser() when plain objects.
  async _getUser (id) {
    const user = await this._users.first(id)
    if (!user) {
      throw new NoSuchUser(id)
    }

    return user
  }

  // TODO: this method will no longer be async when users are
  // integrated to the main collection.
  async getUser (id) {
    return (await this._getUser(id)).properties
  }

  // -----------------------------------------------------------------

  async createGroup ({name}) {
    // TODO: use plain objects.
    const group = (await this._groups.create(name)).properties

    group.users = JSON.parse(group.users)
    return group
  }

  async deleteGroup (id) {
    if (!await this._groups.remove(id)) {
      throw new NoSuchGroup(id)
    }
  }

  async updateGroup (id, {name}) {
    const group = await this.getGroup(id)

    if (name) group.name = name

    await this._groups.save(group)
  }

  async getGroup (id) {
    const group = (await this._groups.first(id))
    if (!group) {
      throw new NoSuchGroup(id)
    }

    return group.properties
  }

  async addUserToGroup (userId, groupId) {
    const [user, group] = await Promise.all([
      this.getUser(userId),
      this.getGroup(groupId)
    ])

    const {groups} = user
    if (!includes(groups, groupId)) {
      user.groups.push(groupId)
    }

    const {users} = group
    if (!includes(users, userId)) {
      group.users.push(userId)
    }

    await Promise.all([
      this._users.save(user),
      this._groups.save(group)
    ])
  }

  async removeUserFromGroup (userId, groupId) {
    const [user, group] = await Promise.all([
      this.getUser(userId),
      this.getGroup(groupId)
    ])

    // TODO: maybe not iterating through the whole arrays?
    user.groups = filter(user.groups, id => id !== groupId)
    group.users = filter(group.users, id => id !== userId)

    await Promise.all([
      this._users.save(user),
      this._groups.save(group)
    ])
  }

  async setGroupUsers (groupId, userIds) {
    const group = await this.getGroup(groupId)

    const newUsersIds = Object.create(null)
    const oldUsersIds = Object.create(null)
    forEach(userIds, id => {
      newUsersIds[id] = null
    })
    forEach(group.users, id => {
      if (id in newUsersIds) {
        delete newUsersIds[id]
      } else {
        oldUsersIds[id] = null
      }
    })

    const [newUsers, oldUsers] = await Promise.all([
      Promise.all(map(newUsersIds, (_, id) => this.getUser(id))),
      Promise.all(map(oldUsersIds, (_, id) => this.getUser(id)))
    ])

    forEach(newUsers, user => {
      const {groups} = user
      if (!includes(groups, groupId)) {
        user.groups.push(groupId)
      }
    })
    forEach(oldUsers, user => {
      user.groups = filter(user.groups, id => id !== groupId)
    })

    group.users = userIds

    await Promise.all([
      Promise.all(map(newUsers, this._users.save, this._users)),
      Promise.all(map(oldUsers, this._users.save, this._users)),
      this._groups.save(group)
    ])
  }

  // -----------------------------------------------------------------

  // TODO: delete when merged with the new collection.
  async getRoles () {
    return [
      {
        id: 'viewer',
        name: 'Viewer',
        permissions: [
          'view'
        ]
      },
      {
        id: 'operator',
        name: 'Operator',
        permissions: [
          'view',
          'operate'
        ]
      },
      {
        id: 'admin',
        name: 'Admin',
        permissions: [
          'view',
          'operate',
          'administrate'
        ]
      }
    ]
  }

  // Returns an array of roles which have a given permission.
  async getRolesForPermission (permission) {
    const roles = []

    forEach(await this.getRoles(), role => {
      if (includes(role.permissions, permission)) {
        roles.push(role.id)
      }
    })

    return roles
  }

  // -----------------------------------------------------------------

  async createAuthenticationToken ({userId}) {
    // TODO: use plain objects
    const token = await this._tokens.generate(userId)

    return token.properties
  }

  async deleteAuthenticationToken (id) {
    if (!await this._tokens.remove(id)) {
      throw new NoSuchAuthenticationToken(id)
    }
  }

  async getAuthenticationToken (id) {
    const token = await this._tokens.first(id)
    if (!token) {
      throw new NoSuchAuthenticationToken(id)
    }

    return token.properties
  }

  // -----------------------------------------------------------------

  async registerXenServer ({host, username, password}) {
    // FIXME: We are storing passwords which is bad!
    //        Could we use tokens instead?
    // TODO: use plain objects
    const server = await this._servers.add({host, username, password})

    return server.properties
  }

  async unregisterXenServer (id) {
    this.disconnectXenServer(id).catch(() => {})

    if (!await this._servers.remove(id)) {
      throw new NoSuchXenServer(id)
    }
  }

  async updateXenServer (id, {host, username, password}) {
    const server = await this._getXenServer(id)

    if (host) server.set('host', host)
    if (username) server.set('username', username)
    if (password) server.set('password', password)

    await this._servers.update(server)
  }

  // TODO: this method will no longer be async when servers are
  // integrated to the main collection.
  async _getXenServer (id) {
    const server = await this._servers.first(id)
    if (!server) {
      throw new NoSuchXenServer(id)
    }

    return server
  }

  @autobind
  _onXenAdd (xapiObjects) {
    const {_objects: objects} = this
    forEach(xapiObjects, (xapiObject, id) => {
      const transform = xapiObjectsToXo[xapiObject.$type]
      if (!transform) {
        return
      }

      const xoObject = transform(xapiObject)
      xoObject.id = id
      xoObject.ref = xapiObject.$ref
      if (!xoObject.type) {
        xoObject.type = xapiObject.$type
      }

      const {$pool: pool} = xapiObject
      Object.defineProperties(xoObject, {
        poolRef: { value: pool.$ref },
        $poolId: {
          enumerable: true,
          value: pool.$id
        },
        ref: { value: xapiObject.$ref }
      })

      objects.set(id, xoObject)
    })
  }

  @autobind
  _onXenRemove (xapiObjects) {
    const {_objects: objects} = this
    forEach(xapiObjects, (_, id) => {
      if (objects.has(id)) {
        objects.remove(id)
      }
    })
  }

  // TODO the previous state should be marked as connected.
  async connectXenServer (id) {
    const server = (await this._getXenServer(id)).properties

    const xapi = this._xapis[server.id] = new Xapi({
      url: server.host,
      auth: {
        user: server.username,
        password: <PASSWORD>
      }
    })

    const {objects} = xapi
    objects.on('add', this._onXenAdd)
    objects.on('update', this._onXenAdd)
    objects.on('remove', this._onXenRemove)

    // Each time objects are refreshed, registers the connection with
    // the pool identifier.
    objects.on('finish', () => {
      this._xapis[xapi.pool.$id] = xapi
    })

    try {
      await xapi.connect()
    } catch (error) {
      if (error.code === 'SESSION_AUTHENTICATION_FAILED') {
        throw new JsonRpcError('authentication failed')
      }
      if (error.code === 'EHOSTUNREACH') {
        throw new JsonRpcError('host unreachable')
      }
      throw error
    }
  }

  // TODO the previous state should be marked as disconnected.
  async disconnectXenServer (id) {
    const xapi = this._xapis[id]
    if (!xapi) {
      throw new NoSuchXenServer(id)
    }

    delete this._xapis[id]
    if (xapi.pool) {
      delete this._xapis[xapi.pool.id]
    }

    return xapi.disconnect()
  }

  // Returns the XAPI connection associated to an object.
  getXAPI (object, type) {
    if (isString(object)) {
      object = this.getObject(object, type)
    }

    const {$poolId: poolId} = object
    if (!poolId) {
      throw new Error(`object ${object.id} does not belong to a pool`)
    }

    const xapi = this._xapis[poolId]
    if (!xapi) {
      throw new Error(`no connection found for object ${object.id}`)
    }

    return xapi
  }

  // -----------------------------------------------------------------

  // Returns an object from its key or UUID.
  //
  // TODO: should throw a NoSuchObject error on failure.
  getObject (key, type) {
    const {
      all,
      indexes: {
        byRef
      }
    } = this._objects

    const obj = all[key] || byRef[key]
    if (!obj) {
      throw new NoSuchObject(key, type)
    }

    if (type != null && (
      isString(type) && type !== obj.type ||
      !includes(type, obj.type) // Array
    )) {
      throw new NoSuchObject(key, type)
    }

    return obj
  }

  getObjects (keys) {
    const {
      all,
      indexes: {
        byRef
      }
    } = this._objects

    // Returns all objects if no keys have been passed.
    if (!keys) {
      return all
    }

    // Fetches all objects and ignores those missing.
    const result = []
    forEach(keys, key => {
      const object = all[key] || byRef[key]
      if (object) {
        result.push(object)
      }
    })
    return result
  }

  // -----------------------------------------------------------------

  createUserConnection () {
    const {_connections: connections} = this

    const connection = new Connection()
    const id = connection.id = this._nextConId++

    connections[id] = connection
    connection.on('close', () => {
      delete connections[id]
    })

    return connection
  }

  // -----------------------------------------------------------------

  _handleHttpRequest (req, res, next) {
    const {url} = req

    const {_httpRequestWatchers: watchers} = this
    const watcher = watchers[url]
    if (!watcher) {
      next()
      return
    }
    delete watchers[url]

    const {fn, data} = watcher
    Bluebird.try(fn, [req, res, data]).then(
      result => {
        if (result != null) {
          res.end(JSON.stringify(result))
        }
      },
      error => {
        console.error('HTTP request error', error.stack || error)

        if (!res.headersSent) {
          res.writeHead(500)
        }
        res.end('unknown error')
      }
    )
  }

  async registerHttpRequest (fn, data) {
    const {_httpRequestWatchers: watchers} = this

    const url = await (function generateUniqueUrl () {
      return generateToken().then(token => {
        const url = `/api/${token}`

        return url in watchers ?
          generateUniqueUrl() :
          url
      })
    })()

    watchers[url] = {
      fn,
      data
    }

    return url
  }

  // -----------------------------------------------------------------

  // TODO: remove when no longer necessary.
  _handleProxyRequest (req, res, next) {
    const {url} = req
    const request = this._proxyRequests[url]
    if (!request || req.method !== request.proxyMethod) {
      next()
      return
    }

    // A proxy request can only be used once.
    delete this._proxyRequests[url]

    proxyRequest(request, req, res)

    if (request.onSuccess) {
      res.on('finish', request.onSuccess)
    }

    const onFailure = request.onFailure || (() => {})

    req.on('close', onFailure)

    const closeConnection = () => {
      if (!res.headerSent) {
        res.writeHead(500)
      }
      res.end()

      onFailure()
    }
    req.on('error', error => {
      console.warn('request error', error.stack || error)
      closeConnection()
    })
    res.on('error', error => {
      console.warn('response error', error.stack || error)
      closeConnection()
    })
  }
  async registerProxyRequest (opts) {
    if (isString(opts)) {
      opts = parseUrl(opts)
    } else {
      opts.method = opts.method != null ?
        opts.method.toUpperCase() :
        'GET'

      opts.proxyMethod = opts.proxyMethod != null ?
        opts.proxyMethod.toUpperCase() :
        opts.method
    }

    opts.createdAt = Date.now()

    const url = `/${await generateToken()}`
    this._proxyRequests[url] = opts

    return url
  }

  // -----------------------------------------------------------------

  registerAuthenticationProvider (provider) {
    return this._authenticationProviders.add(provider)
  }

  unregisterAuthenticationProvider (provider) {
    return this._authenticationProviders.remove(provider)
  }

  async authenticateUser (credentials) {
    // TODO: remove when email has been replaced by username.
    if (credentials.email) {
      credentials.username = credentials.email
    } else if (credentials.username) {
      credentials.email = credentials.username
    }

    for (let provider of this._authenticationProviders) {
      try {
        const result = await provider(credentials)

        if (result instanceof User) {
          return result
        }

        // TODO: replace by email by username.
        if (result.username) {
          result.email = result.username
          delete result.username
        }

        const user = await this._users.first(result)
        if (user) return user

        return this._users.create(result.email)
      } catch (error) {
        // Authentication providers may just throw `null` to indicate
        // they could not authenticate the user without any special
        // errors.
        if (error) console.error(error.stack || error)
      }
    }

    return false
  }

  // -----------------------------------------------------------------

  // Watches objects changes.
  //
  // Some should be forwarded to connected clients.
  // Some should be persistently saved.
  _watchObjects () {
    const {
      _connections: connections,
      _objects: objects
    } = this

    let entered, exited
    function reset () {
      entered = Object.create(null)
      exited = Object.create(null)
    }
    reset()

    function onAdd (items) {
      forEach(items, (item, id) => {
        entered[id] = item
      })
    }
    objects.on('add', onAdd)
    objects.on('update', onAdd)

    objects.on('remove', (items) => {
      forEach(items, (_, id) => {
        // We don't care about the value here, so we choose `0`
        // because it is small in JSON.
        exited[id] = 0
      })
    })

    objects.on('finish', () => {
      const enteredMessage = !isEmpty(entered) && {
        type: 'enter',
        items: entered
      }
      const exitedMessage = !isEmpty(exited) && {
        type: 'exit',
        items: exited
      }

      if (!enteredMessage && !exitedMessage) {
        return
      }

      forEach(connections, connection => {
        // Notifies only authenticated clients.
        if (connection.has('user_id')) {
          if (enteredMessage) {
            connection.notify('all', enteredMessage)
          }
          if (exitedMessage) {
            connection.notify('all', exitedMessage)
          }
        }
      })

      reset()
    })
  }
}
<commit_msg>Work around babel/babel#1850.<commit_after>import Bluebird from 'bluebird'
import filter from 'lodash.filter'
import forEach from 'lodash.foreach'
import includes from 'lodash.includes'
import isEmpty from 'lodash.isempty'
import isString from 'lodash.isstring'
import map from 'lodash.map'
import proxyRequest from 'proxy-http-request'
import XoCollection from 'xo-collection'
import XoUniqueIndex from 'xo-collection/unique-index'
// import XoView from 'xo-collection/view'
import {createClient as createRedisClient} from 'then-redis'
import {EventEmitter} from 'events'
import {parse as parseUrl} from 'url'

import * as xapiObjectsToXo from './xapi-objects-to-xo'
import Connection from './connection'
import User, {Users} from './models/user'
import Xapi from './xapi'
import {Acls} from './models/acl'
import {autobind} from './decorators'
import {generateToken} from './utils'
import {Groups} from './models/group'
import {JsonRpcError, NoSuchObject} from './api-errors'
import {ModelAlreadyExists} from './collection'
import {Servers} from './models/server'
import {Tokens} from './models/token'

// ===================================================================

class NoSuchAuthenticationToken extends NoSuchObject {
  constructor (id) {
    super(id, 'authentication token')
  }
}

class NoSuchGroup extends NoSuchObject {
  constructor (id) {
    super(id, 'group')
  }
}

class NoSuchUser extends NoSuchObject {
  constructor (id) {
    super(id, 'user')
  }
}

class NoSuchXenServer extends NoSuchObject {
  constructor (id) {
    super(id, 'xen server')
  }
}

// ===================================================================

export default class Xo extends EventEmitter {
  constructor () {
    super()

    this._objects = new XoCollection()
    this._objects.createIndex('byRef', new XoUniqueIndex('ref'))

    // These will be initialized in start()
    //
    // TODO: remove and put everything in the `_objects` collection.
    this._acls = null
    this._groups = null
    this._servers = null
    this._tokens = null
    this._users = null
    this._UUIDsToKeys = null

    // Connections to Xen servers.
    this._xapis = Object.create(null)

    // Connections to users.
    this._nextConId = 0
    this._connections = Object.create(null)

    this._httpRequestWatchers = Object.create(null)

    // TODO: remove when no longer necessary.
    this._proxyRequests = Object.create(null)

    this._authenticationProviders = new Set()

    this._watchObjects()
  }

  // -----------------------------------------------------------------

  async start (config) {
    // Connects to Redis.
    const redis = createRedisClient(config.redis && config.redis.uri)

    // Creates persistent collections.
    this._acls = new Acls({
      connection: redis,
      prefix: 'xo:acl',
      indexes: ['subject', 'object']
    })
    this._groups = new Groups({
      connection: redis,
      prefix: 'xo:group'
    })
    this._servers = new Servers({
      connection: redis,
      prefix: 'xo:server',
      indexes: ['host']
    })
    this._tokens = new Tokens({
      connection: redis,
      prefix: 'xo:token',
      indexes: ['user_id']
    })
    this._users = new Users({
      connection: redis,
      prefix: 'xo:user',
      indexes: ['email']
    })

    // Proxies tokens/users related events to XO and removes tokens
    // when their related user is removed.
    this._tokens.on('remove', ids => {
      for (let id of ids) {
        this.emit(`token.revoked:${id}`)
      }
    })
    this._users.on('remove', async function (ids) {
      for (let id of ids) {
        this.emit(`user.revoked:${id}`)
        const tokens = await this._tokens.get({ user_id: id })
        for (let token of tokens) {
          this._tokens.remove(token.id)
        }
      }
    }.bind(this))

    // Connects to existing servers.
    const servers = await this._servers.get()
    for (let server of servers) {
      this.connectXenServer(server.id).catch(error => {
        console.error(
          `[WARN] ${server.host}:`,
          error[0] || error.stack || error.code || error
        )
      })
    }
  }

  // -----------------------------------------------------------------

  async addAcl (subjectId, objectId, action) {
    try {
      await this._acls.create(subjectId, objectId, action)
    } catch (error) {
      if (!(error instanceof ModelAlreadyExists)) {
        throw error
      }
    }
  }

  async removeAcl (subjectId, objectId, action) {
    await this._acls.delete(subjectId, objectId, action)
  }

  async getAclsForUser (userId) {
    const subjects = (await this.getUser(userId)).groups.concat(userId)

    const acls = []
    const pushAcls = (function (push) {
      return function (entries) {
        push.apply(acls, entries)
      }
    })(acls.push)

    const {_acls: collection} = this
    await Promise.all(map(
      subjects,
      subject => collection.get({subject}).then(pushAcls)
    ))

    return acls
  }

  // TODO: remove when new collection.
  async getAllAcls () {
    return this._acls.get()
  }

  async hasPermission (userId, objectId, permission) {
    const user = await this.getUser(userId)

    // Special case for super XO administrators.
    //
    // TODO: restore when necessary, for now it is already implemented
    // in resolveParams().
    // if (user.permission === 'admin') {
    //   return true
    // }

    const subjects = user.groups.concat(userId)
    let actions = (await this.getRolesForPermission(permission)).concat(permission)

    const promises = []
    {
      const {_acls: acls} = this
      const throwIfFail = function (success) {
        if (!success) {
          // We don't care about an error object.
          /* eslint no-throw-literal: 0 */
          throw null
        }
      }
      forEach(subjects, subject => {
        forEach(actions, action => {
          promises.push(
            acls.aclExists(subject, objectId, action).then(throwIfFail)
          )
        })
      })
    }

    try {
      await Bluebird.any(promises)
      return true
    } catch (_) {
      return false
    }
  }

  // -----------------------------------------------------------------

  async createUser ({email, password, permission}) {
    // TODO: use plain objects
    const user = await this._users.create(email, password, permission)

    return user.properties
  }

  async deleteUser (id) {
    if (!await this._users.remove(id)) {
      throw new NoSuchUser(id)
    }
  }

  async updateUser (id, {email, password, permission}) {
    const user = await this._getUser(id)

    if (email) user.set('email', email)
    if (permission) user.set('permission', permission)
    if (password) {
      await user.setPassword(password)
    }

    await this._users.save(user.properties)
  }

  // Merge this method in getUser() when plain objects.
  async _getUser (id) {
    const user = await this._users.first(id)
    if (!user) {
      throw new NoSuchUser(id)
    }

    return user
  }

  // TODO: this method will no longer be async when users are
  // integrated to the main collection.
  async getUser (id) {
    return (await this._getUser(id)).properties
  }

  // -----------------------------------------------------------------

  async createGroup ({name}) {
    // TODO: use plain objects.
    const group = (await this._groups.create(name)).properties

    group.users = JSON.parse(group.users)
    return group
  }

  async deleteGroup (id) {
    if (!await this._groups.remove(id)) {
      throw new NoSuchGroup(id)
    }
  }

  async updateGroup (id, {name}) {
    const group = await this.getGroup(id)

    if (name) group.name = name

    await this._groups.save(group)
  }

  async getGroup (id) {
    const group = (await this._groups.first(id))
    if (!group) {
      throw new NoSuchGroup(id)
    }

    return group.properties
  }

  async addUserToGroup (userId, groupId) {
    const [user, group] = await Promise.all([
      this.getUser(userId),
      this.getGroup(groupId)
    ])

    const {groups} = user
    if (!includes(groups, groupId)) {
      user.groups.push(groupId)
    }

    const {users} = group
    if (!includes(users, userId)) {
      group.users.push(userId)
    }

    await Promise.all([
      this._users.save(user),
      this._groups.save(group)
    ])
  }

  async removeUserFromGroup (userId, groupId) {
    const [user, group] = await Promise.all([
      this.getUser(userId),
      this.getGroup(groupId)
    ])

    // TODO: maybe not iterating through the whole arrays?
    user.groups = filter(user.groups, id => id !== groupId)
    group.users = filter(group.users, id => id !== userId)

    await Promise.all([
      this._users.save(user),
      this._groups.save(group)
    ])
  }

  async setGroupUsers (groupId, userIds) {
    const group = await this.getGroup(groupId)

    const newUsersIds = Object.create(null)
    const oldUsersIds = Object.create(null)
    forEach(userIds, id => {
      newUsersIds[id] = null
    })
    forEach(group.users, id => {
      if (id in newUsersIds) {
        delete newUsersIds[id]
      } else {
        oldUsersIds[id] = null
      }
    })

    const [newUsers, oldUsers] = await Promise.all([
      Promise.all(map(newUsersIds, (_, id) => this.getUser(id))),
      Promise.all(map(oldUsersIds, (_, id) => this.getUser(id)))
    ])

    forEach(newUsers, user => {
      const {groups} = user
      if (!includes(groups, groupId)) {
        user.groups.push(groupId)
      }
    })
    forEach(oldUsers, user => {
      user.groups = filter(user.groups, id => id !== groupId)
    })

    group.users = userIds

    await Promise.all([
      Promise.all(map(newUsers, this._users.save, this._users)),
      Promise.all(map(oldUsers, this._users.save, this._users)),
      this._groups.save(group)
    ])
  }

  // -----------------------------------------------------------------

  // TODO: delete when merged with the new collection.
  async getRoles () {
    return [
      {
        id: 'viewer',
        name: 'Viewer',
        permissions: [
          'view'
        ]
      },
      {
        id: 'operator',
        name: 'Operator',
        permissions: [
          'view',
          'operate'
        ]
      },
      {
        id: 'admin',
        name: 'Admin',
        permissions: [
          'view',
          'operate',
          'administrate'
        ]
      }
    ]
  }

  // Returns an array of roles which have a given permission.
  async getRolesForPermission (permission) {
    const roles = []

    forEach(await this.getRoles(), role => {
      if (includes(role.permissions, permission)) {
        roles.push(role.id)
      }
    })

    return roles
  }

  // -----------------------------------------------------------------

  async createAuthenticationToken ({userId}) {
    // TODO: use plain objects
    const token = await this._tokens.generate(userId)

    return token.properties
  }

  async deleteAuthenticationToken (id) {
    if (!await this._tokens.remove(id)) {
      throw new NoSuchAuthenticationToken(id)
    }
  }

  async getAuthenticationToken (id) {
    const token = await this._tokens.first(id)
    if (!token) {
      throw new NoSuchAuthenticationToken(id)
    }

    return token.properties
  }

  // -----------------------------------------------------------------

  async registerXenServer ({host, username, password}) {
    // FIXME: We are storing passwords which is bad!
    //        Could we use tokens instead?
    // TODO: use plain objects
    const server = await this._servers.add({host, username, password})

    return server.properties
  }

  async unregisterXenServer (id) {
    this.disconnectXenServer(id).catch(() => {})

    if (!await this._servers.remove(id)) {
      throw new NoSuchXenServer(id)
    }
  }

  async updateXenServer (id, {host, username, password}) {
    const server = await this._getXenServer(id)

    if (host) server.set('host', host)
    if (username) server.set('username', username)
    if (password) server.set('password', password)

    await this._servers.update(server)
  }

  // TODO: this method will no longer be async when servers are
  // integrated to the main collection.
  async _getXenServer (id) {
    const server = await this._servers.first(id)
    if (!server) {
      throw new NoSuchXenServer(id)
    }

    return server
  }

  @autobind
  _onXenAdd (xapiObjects) {
    const {_objects: objects} = this
    forEach(xapiObjects, (xapiObject, id) => {
      const transform = xapiObjectsToXo[xapiObject.$type]
      if (!transform) {
        return
      }

      const xoObject = transform(xapiObject)
      xoObject.id = id
      xoObject.ref = xapiObject.$ref
      if (!xoObject.type) {
        xoObject.type = xapiObject.$type
      }

      const {$pool: pool} = xapiObject
      Object.defineProperties(xoObject, {
        poolRef: { value: pool.$ref },
        $poolId: {
          enumerable: true,
          value: pool.$id
        },
        ref: { value: xapiObject.$ref }
      })

      objects.set(id, xoObject)
    })
  }

  @autobind
  _onXenRemove (xapiObjects) {
    const {_objects: objects} = this
    forEach(xapiObjects, (_, id) => {
      if (objects.has(id)) {
        objects.remove(id)
      }
    })
  }

  // TODO the previous state should be marked as connected.
  async connectXenServer (id) {
    const server = (await this._getXenServer(id)).properties

    const xapi = this._xapis[server.id] = new Xapi({
      url: server.host,
      auth: {
        user: server.username,
        password: <PASSWORD>
      }
    })

    const {objects} = xapi
    objects.on('add', this._onXenAdd)
    objects.on('update', this._onXenAdd)
    objects.on('remove', this._onXenRemove)

    // Each time objects are refreshed, registers the connection with
    // the pool identifier.
    objects.on('finish', () => {
      this._xapis[xapi.pool.$id] = xapi
    })

    try {
      await xapi.connect()
    } catch (error) {
      if (error.code === 'SESSION_AUTHENTICATION_FAILED') {
        throw new JsonRpcError('authentication failed')
      }
      if (error.code === 'EHOSTUNREACH') {
        throw new JsonRpcError('host unreachable')
      }
      throw error
    }
  }

  // TODO the previous state should be marked as disconnected.
  async disconnectXenServer (id) {
    const xapi = this._xapis[id]
    if (!xapi) {
      throw new NoSuchXenServer(id)
    }

    delete this._xapis[id]
    if (xapi.pool) {
      delete this._xapis[xapi.pool.id]
    }

    return xapi.disconnect()
  }

  // Returns the XAPI connection associated to an object.
  getXAPI (object, type) {
    if (isString(object)) {
      object = this.getObject(object, type)
    }

    const {$poolId: poolId} = object
    if (!poolId) {
      throw new Error(`object ${object.id} does not belong to a pool`)
    }

    const xapi = this._xapis[poolId]
    if (!xapi) {
      throw new Error(`no connection found for object ${object.id}`)
    }

    return xapi
  }

  // -----------------------------------------------------------------

  // Returns an object from its key or UUID.
  //
  // TODO: should throw a NoSuchObject error on failure.
  getObject (key, type) {
    const {
      all,
      indexes: {
        byRef
      }
    } = this._objects

    const obj = all[key] || byRef[key]
    if (!obj) {
      throw new NoSuchObject(key, type)
    }

    if (type != null && (
      isString(type) && type !== obj.type ||
      !includes(type, obj.type) // Array
    )) {
      throw new NoSuchObject(key, type)
    }

    return obj
  }

  getObjects (keys) {
    const {
      all,
      indexes: {
        byRef
      }
    } = this._objects

    // Returns all objects if no keys have been passed.
    if (!keys) {
      return all
    }

    // Fetches all objects and ignores those missing.
    const result = []
    forEach(keys, key => {
      const object = all[key] || byRef[key]
      if (object) {
        result.push(object)
      }
    })
    return result
  }

  // -----------------------------------------------------------------

  createUserConnection () {
    const {_connections: connections} = this

    const connection = new Connection()
    const id = connection.id = this._nextConId++

    connections[id] = connection
    connection.on('close', () => {
      delete connections[id]
    })

    return connection
  }

  // -----------------------------------------------------------------

  _handleHttpRequest (req, res, next) {
    const {url} = req

    const {_httpRequestWatchers: watchers} = this
    const watcher = watchers[url]
    if (!watcher) {
      next()
      return
    }
    delete watchers[url]

    const {fn, data} = watcher
    Bluebird.try(fn, [req, res, data]).then(
      result => {
        if (result != null) {
          res.end(JSON.stringify(result))
        }
      },
      error => {
        console.error('HTTP request error', error.stack || error)

        if (!res.headersSent) {
          res.writeHead(500)
        }
        res.end('unknown error')
      }
    )
  }

  async registerHttpRequest (fn, data) {
    const {_httpRequestWatchers: watchers} = this

    const url = await (function generateUniqueUrl () {
      return generateToken().then(token => {
        const url = `/api/${token}`

        return url in watchers ?
          generateUniqueUrl() :
          url
      })
    })()

    watchers[url] = {
      fn,
      data
    }

    return url
  }

  // -----------------------------------------------------------------

  // TODO: remove when no longer necessary.
  _handleProxyRequest (req, res, next) {
    const {url} = req
    const request = this._proxyRequests[url]
    if (!request || req.method !== request.proxyMethod) {
      next()
      return
    }

    // A proxy request can only be used once.
    delete this._proxyRequests[url]

    proxyRequest(request, req, res)

    if (request.onSuccess) {
      res.on('finish', request.onSuccess)
    }

    const onFailure = request.onFailure || (() => {})

    req.on('close', onFailure)

    const closeConnection = () => {
      if (!res.headerSent) {
        res.writeHead(500)
      }
      res.end()

      onFailure()
    }
    req.on('error', error => {
      console.warn('request error', error.stack || error)
      closeConnection()
    })
    res.on('error', error => {
      console.warn('response error', error.stack || error)
      closeConnection()
    })
  }
  async registerProxyRequest (opts) {
    if (isString(opts)) {
      opts = parseUrl(opts)
    } else {
      opts.method = opts.method != null ?
        opts.method.toUpperCase() :
        'GET'

      opts.proxyMethod = opts.proxyMethod != null ?
        opts.proxyMethod.toUpperCase() :
        opts.method
    }

    opts.createdAt = Date.now()

    const url = `/${await generateToken()}`
    this._proxyRequests[url] = opts

    return url
  }

  // -----------------------------------------------------------------

  registerAuthenticationProvider (provider) {
    return this._authenticationProviders.add(provider)
  }

  unregisterAuthenticationProvider (provider) {
    return this._authenticationProviders.remove(provider)
  }

  async authenticateUser (credentials) {
    // TODO: remove when email has been replaced by username.
    if (credentials.email) {
      credentials.username = credentials.email
    } else if (credentials.username) {
      credentials.email = credentials.username
    }

    for (let provider of this._authenticationProviders) {
      try {
        const result = await provider(credentials)

        if (result instanceof User) {
          return result
        }

        // TODO: replace by email by username.
        if (result.username) {
          result.email = result.username
          delete result.username
        }

        const user = await this._users.first(result)
        if (user) return user

        return this._users.create(result.email)
      } catch (error) {
        // Authentication providers may just throw `null` to indicate
        // they could not authenticate the user without any special
        // errors.
        if (error) console.error(error.stack || error)
      }
    }

    return false
  }

  // -----------------------------------------------------------------

  // Watches objects changes.
  //
  // Some should be forwarded to connected clients.
  // Some should be persistently saved.
  _watchObjects () {
    const {
      _connections: connections,
      _objects: objects
    } = this

    let entered, exited
    function reset () {
      entered = Object.create(null)
      exited = Object.create(null)
    }
    reset()

    function onAdd (items) {
      forEach(items, (item, id) => {
        entered[id] = item
      })
    }
    objects.on('add', onAdd)
    objects.on('update', onAdd)

    objects.on('remove', (items) => {
      forEach(items, (_, id) => {
        // We don't care about the value here, so we choose `0`
        // because it is small in JSON.
        exited[id] = 0
      })
    })

    objects.on('finish', () => {
      const enteredMessage = !isEmpty(entered) && {
        type: 'enter',
        items: entered
      }
      const exitedMessage = !isEmpty(exited) && {
        type: 'exit',
        items: exited
      }

      if (!enteredMessage && !exitedMessage) {
        return
      }

      forEach(connections, connection => {
        // Notifies only authenticated clients.
        if (connection.has('user_id')) {
          if (enteredMessage) {
            connection.notify('all', enteredMessage)
          }
          if (exitedMessage) {
            connection.notify('all', exitedMessage)
          }
        }
      })

      reset()
    })
  }
}

{'fraction_non_alphanumeric': 0.16276179516685846, 'fraction_numerical': 0.0005983889528193326, 'mean_word_length': 2.0096973055343907, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5159300', 'n_tokens_mistral': 13007, 'n_tokens_neox': 12699, 'n_words': 3237}