server/models/pod.ts

   1 import { each, waterfall } from 'async'
   2 import { map } from 'lodash'
   3 import * as Sequelize from 'sequelize'
   4
   5 import { FRIEND_SCORE, PODS_SCORE } from '../initializers'
   6 import { logger, isHostValid } from '../helpers'
   7
   8 import { addMethodsToModel } from './utils'
   9 import {
  10   PodClass,
  11   PodInstance,
  12   PodAttributes,
  13
  14   PodMethods
  15 } from './pod-interface'
  16
  17 let Pod: Sequelize.Model<PodInstance, PodAttributes>
  18 let toFormatedJSON: PodMethods.ToFormatedJSON
  19 let countAll: PodMethods.CountAll
  20 let incrementScores: PodMethods.IncrementScores
  21 let list: PodMethods.List
  22 let listAllIds: PodMethods.ListAllIds
  23 let listRandomPodIdsWithRequest: PodMethods.ListRandomPodIdsWithRequest
  24 let listBadPods: PodMethods.ListBadPods
  25 let load: PodMethods.Load
  26 let loadByHost: PodMethods.LoadByHost
  27 let removeAll: PodMethods.RemoveAll
  28 let updatePodsScore: PodMethods.UpdatePodsScore
  29
  30 export default function (sequelize, DataTypes) {
  31   Pod = sequelize.define('Pod',
  32     {
  33       host: {
  34         type: DataTypes.STRING,
  35         allowNull: false,
  36         validate: {
  37           isHost: function (value) {
  38             const res = isHostValid(value)
  39             if (res === false) throw new Error('Host not valid.')
  40           }
  41         }
  42       },
  43       publicKey: {
  44         type: DataTypes.STRING(5000),
  45         allowNull: false
  46       },
  47       score: {
  48         type: DataTypes.INTEGER,
  49         defaultValue: FRIEND_SCORE.BASE,
  50         allowNull: false,
  51         validate: {
  52           isInt: true,
  53           max: FRIEND_SCORE.MAX
  54         }
  55       },
  56       email: {
  57         type: DataTypes.STRING(400),
  58         allowNull: false,
  59         validate: {
  60           isEmail: true
  61         }
  62       }
  63     },
  64     {
  65       indexes: [
  66         {
  67           fields: [ 'host' ],
  68           unique: true
  69         },
  70         {
  71           fields: [ 'score' ]
  72         }
  73       ]
  74     }
  75   )
  76
  77   const classMethods = [
  78     associate,
  79
  80     countAll,
  81     incrementScores,
  82     list,
  83     listAllIds,
  84     listRandomPodIdsWithRequest,
  85     listBadPods,
  86     load,
  87     loadByHost,
  88     updatePodsScore,
  89     removeAll
  90   ]
  91   const instanceMethods = [ toFormatedJSON ]
  92   addMethodsToModel(Pod, classMethods, instanceMethods)
  93
  94   return Pod
  95 }
  96
  97 // ------------------------------ METHODS ------------------------------
  98
  99 toFormatedJSON = function () {
 100   const json = {
 101     id: this.id,
 102     host: this.host,
 103     email: this.email,
 104     score: this.score,
 105     createdAt: this.createdAt
 106   }
 107
 108   return json
 109 }
 110
 111 // ------------------------------ Statics ------------------------------
 112
 113 function associate (models) {
 114   Pod.belongsToMany(models.Request, {
 115     foreignKey: 'podId',
 116     through: models.RequestToPod,
 117     onDelete: 'cascade'
 118   })
 119 }
 120
 121 countAll = function (callback) {
 122   return Pod.count().asCallback(callback)
 123 }
 124
 125 incrementScores = function (ids, value, callback) {
 126   if (!callback) callback = function () { /* empty */ }
 127
 128   const update = {
 129     score: Sequelize.literal('score +' + value)
 130   }
 131
 132   const options = {
 133     where: {
 134       id: {
 135         $in: ids
 136       }
 137     },
 138     // In this case score is a literal and not an integer so we do not validate it
 139     validate: false
 140   }
 141
 142   return Pod.update(update, options).asCallback(callback)
 143 }
 144
 145 list = function (callback) {
 146   return Pod.findAll().asCallback(callback)
 147 }
 148
 149 listAllIds = function (transaction, callback) {
 150   if (!callback) {
 151     callback = transaction
 152     transaction = null
 153   }
 154
 155   const query: any = {
 156     attributes: [ 'id' ]
 157   }
 158
 159   if (transaction) query.transaction = transaction
 160
 161   return Pod.findAll(query).asCallback(function (err, pods) {
 162     if (err) return callback(err)
 163
 164     return callback(null, map(pods, 'id'))
 165   })
 166 }
 167
 168 listRandomPodIdsWithRequest = function (limit, tableWithPods, tableWithPodsJoins, callback) {
 169   if (!callback) {
 170     callback = tableWithPodsJoins
 171     tableWithPodsJoins = ''
 172   }
 173
 174   Pod.count().asCallback(function (err, count) {
 175     if (err) return callback(err)
 176
 177     // Optimization...
 178     if (count === 0) return callback(null, [])
 179
 180     let start = Math.floor(Math.random() * count) - limit
 181     if (start < 0) start = 0
 182
 183     const query = {
 184       attributes: [ 'id' ],
 185       order: [
 186         [ 'id', 'ASC' ]
 187       ],
 188       offset: start,
 189       limit: limit,
 190       where: {
 191         id: {
 192           $in: [
 193             Sequelize.literal(`SELECT DISTINCT "${tableWithPods}"."podId" FROM "${tableWithPods}" ${tableWithPodsJoins}`)
 194           ]
 195         }
 196       }
 197     }
 198
 199     return Pod.findAll(query).asCallback(function (err, pods) {
 200       if (err) return callback(err)
 201
 202       return callback(null, map(pods, 'id'))
 203     })
 204   })
 205 }
 206
 207 listBadPods = function (callback) {
 208   const query = {
 209     where: {
 210       score: { $lte: 0 }
 211     }
 212   }
 213
 214   return Pod.findAll(query).asCallback(callback)
 215 }
 216
 217 load = function (id, callback) {
 218   return Pod.findById(id).asCallback(callback)
 219 }
 220
 221 loadByHost = function (host, callback) {
 222   const query = {
 223     where: {
 224       host: host
 225     }
 226   }
 227
 228   return Pod.findOne(query).asCallback(callback)
 229 }
 230
 231 removeAll = function (callback) {
 232   return Pod.destroy().asCallback(callback)
 233 }
 234
 235 updatePodsScore = function (goodPods, badPods) {
 236   logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
 237
 238   if (goodPods.length !== 0) {
 239     incrementScores(goodPods, PODS_SCORE.BONUS, function (err) {
 240       if (err) logger.error('Cannot increment scores of good pods.', { error: err })
 241     })
 242   }
 243
 244   if (badPods.length !== 0) {
 245     incrementScores(badPods, PODS_SCORE.MALUS, function (err) {
 246       if (err) logger.error('Cannot decrement scores of bad pods.', { error: err })
 247       removeBadPods()
 248     })
 249   }
 250 }
 251
 252 // ---------------------------------------------------------------------------
 253
 254 // Remove pods with a score of 0 (too many requests where they were unreachable)
 255 function removeBadPods () {
 256   waterfall([
 257     function findBadPods (callback) {
 258       listBadPods(function (err, pods) {
 259         if (err) {
 260           logger.error('Cannot find bad pods.', { error: err })
 261           return callback(err)
 262         }
 263
 264         return callback(null, pods)
 265       })
 266     },
 267
 268     function removeTheseBadPods (pods, callback) {
 269       each(pods, function (pod: any, callbackEach) {
 270         pod.destroy().asCallback(callbackEach)
 271       }, function (err) {
 272         return callback(err, pods.length)
 273       })
 274     }
 275   ], function (err, numberOfPodsRemoved) {
 276     if (err) {
 277       logger.error('Cannot remove bad pods.', { error: err })
 278     } else if (numberOfPodsRemoved) {
 279       logger.info('Removed %d pods.', numberOfPodsRemoved)
 280     } else {
 281       logger.info('No need to remove bad pods.')
 282     }
 283   })
 284 }