<commit_before>import Promise from 'bluebird'
import { createClient as createRedisClient } from 'redis'
import _ from 'lodash'
import IoServer from 'socket.io'
import redis_adapter from 'socket.io-redis'
import jwt from 'jsonwebtoken'

import models from './models'
import config_loader from '../config/config'

export default class PubsubListener {
  constructor(server, app) {
    this.app = app

    var config = config_loader.load()

    var redisPub = createRedisClient(config.redis.port, config.redis.host, config.redis.options)
      , redisSub = createRedisClient(config.redis.port, config.redis.host, _.extend(config.redis.options, { detect_buffers: true }))

    redisPub.on('error', function(err) { console.error('redisPub error', err) })
    redisSub.on('error', function(err) { console.error('redisSub error', err) })

    this.io = IoServer(server)
    this.io.adapter(redis_adapter({
      pubClient: redisPub,
      subClient: redisSub
    }))

    this.io.sockets.on('error', function(err) { console.error('socket.io error', err) })
    this.io.sockets.on('connection', this.onConnect.bind(this))

    var redisClient = createRedisClient(config.redis.port, config.redis.host, {})
    redisClient.on('error', function(err) { console.error('redis error', err) })
    redisClient.subscribe('post:new', 'post:destroy', 'post:update',
      'comment:new', 'comment:destroy', 'comment:update',
      'like:new', 'like:remove', 'post:hide', 'post:unhide')

    redisClient.on('message', this.onRedisMessage.bind(this))
  }

  async onConnect(socket) {
    let authToken = socket.handshake.query.token
    let config = config_loader.load()
    let secret = config.secret
    let logger = this.app.logger

    let jwtAsync = Promise.promisifyAll(jwt)
    try {
      let decoded = await jwtAsync.verifyAsync(authToken, secret)
      socket.user = await models.User.findById(decoded.userId)
    } catch(e) {
      socket.user = { id: null }
    }

    socket.on('subscribe', function(data) {
      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has subscribed to ' + id + ' ' + channel)

              socket.join(channel + ':' + id)
            }
          })
        }
      }
    })

    socket.on('unsubscribe', function(data) {
      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has unsubscribed from ' + id + ' ' + channel)

              socket.leave(channel + ':' + id)
            }
          })
        }
      }
    })
  }

  onRedisMessage(channel, msg) {
    const messageRoutes = {
      'post:new':         this.onPostNew.bind(this),
      'post:update':      this.onPostUpdate.bind(this),
      'post:destroy':     this.onPostDestroy.bind(this),
      'post:hide':        this.onPostHide.bind(this),
      'post:unhide':      this.onPostUnhide.bind(this),

      'comment:new':      this.onCommentNew.bind(this),
      'comment:update':   this.onCommentUpdate.bind(this),
      'comment:destroy':  this.onCommentDestroy.bind(this),

      'like:new':         this.onLikeNew.bind(this),
      'like:remove':      this.onLikeRemove.bind(this)
    }

    messageRoutes[channel](
      this.io.sockets,
      JSON.parse(msg)
    ).catch(e => {console.error('onRedisMessage error', e, e.stack)})
  }

  async validateAndEmitMessage(sockets, room, type, json, post) {
    if (!(room in sockets.adapter.rooms)) {
      return
    }

    let clientIds = Object.keys(sockets.adapter.rooms[room])

    await Promise.all(clientIds.map(async (clientId) => {
      let socket = sockets.connected[clientId]
      let user = socket.user

      let valid = await post.validateCanShow(user.id)

      if (valid)
        socket.emit(type, json)
    }))
  }

  // Message-handlers follow
  async onPostDestroy(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = { meta: { postId: data.postId } }

    sockets.in('timeline:' + data.timelineId).emit('post:destroy', json)
    sockets.in('post:' + data.postId).emit('post:destroy', json)

    let type = 'post:destroy'
    let room = `timeline:${data.timelineId}`
    await this.validateAndEmitMessage(sockets, room, type, json, post)

    room = `post:${data.postId}`
    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onPostNew(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = await new models.PostSerializer(post).promiseToJSON()

    let type = 'post:new'
    let room = `timeline:${data.timelineId}`

    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onPostUpdate(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = await new models.PostSerializer(post).promiseToJSON()

    let type = 'post:update'
    let room

    if (data.timelineId) {
      room = `timeline:${data.timelineId}`
    } else {
      room = `post:${data.postId}`
    }
    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onCommentNew(sockets, data) {
    let comment = await models.Comment.findById(data.commentId)
    let post = await models.Post.findById(comment.postId)
    let json = await new models.PubsubCommentSerializer(comment).promiseToJSON()

    let type = 'comment:new'
    let room

    if (data.timelineId) {<commit_msg>ignore stale "new comment" events<commit_after>import Promise from 'bluebird'
import { createClient as createRedisClient } from 'redis'
import _ from 'lodash'
import IoServer from 'socket.io'
import redis_adapter from 'socket.io-redis'
import jwt from 'jsonwebtoken'

import models from './models'
import config_loader from '../config/config'

export default class PubsubListener {
  constructor(server, app) {
    this.app = app

    var config = config_loader.load()

    var redisPub = createRedisClient(config.redis.port, config.redis.host, config.redis.options)
      , redisSub = createRedisClient(config.redis.port, config.redis.host, _.extend(config.redis.options, { detect_buffers: true }))

    redisPub.on('error', function(err) { console.error('redisPub error', err) })
    redisSub.on('error', function(err) { console.error('redisSub error', err) })

    this.io = IoServer(server)
    this.io.adapter(redis_adapter({
      pubClient: redisPub,
      subClient: redisSub
    }))

    this.io.sockets.on('error', function(err) { console.error('socket.io error', err) })
    this.io.sockets.on('connection', this.onConnect.bind(this))

    var redisClient = createRedisClient(config.redis.port, config.redis.host, {})
    redisClient.on('error', function(err) { console.error('redis error', err) })
    redisClient.subscribe('post:new', 'post:destroy', 'post:update',
      'comment:new', 'comment:destroy', 'comment:update',
      'like:new', 'like:remove', 'post:hide', 'post:unhide')

    redisClient.on('message', this.onRedisMessage.bind(this))
  }

  async onConnect(socket) {
    let authToken = socket.handshake.query.token
    let config = config_loader.load()
    let secret = config.secret
    let logger = this.app.logger

    let jwtAsync = Promise.promisifyAll(jwt)
    try {
      let decoded = await jwtAsync.verifyAsync(authToken, secret)
      socket.user = await models.User.findById(decoded.userId)
    } catch(e) {
      socket.user = { id: null }
    }

    socket.on('subscribe', function(data) {
      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has subscribed to ' + id + ' ' + channel)

              socket.join(channel + ':' + id)
            }
          })
        }
      }
    })

    socket.on('unsubscribe', function(data) {
      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has unsubscribed from ' + id + ' ' + channel)

              socket.leave(channel + ':' + id)
            }
          })
        }
      }
    })
  }

  onRedisMessage(channel, msg) {
    const messageRoutes = {
      'post:new':         this.onPostNew.bind(this),
      'post:update':      this.onPostUpdate.bind(this),
      'post:destroy':     this.onPostDestroy.bind(this),
      'post:hide':        this.onPostHide.bind(this),
      'post:unhide':      this.onPostUnhide.bind(this),

      'comment:new':      this.onCommentNew.bind(this),
      'comment:update':   this.onCommentUpdate.bind(this),
      'comment:destroy':  this.onCommentDestroy.bind(this),

      'like:new':         this.onLikeNew.bind(this),
      'like:remove':      this.onLikeRemove.bind(this)
    }

    messageRoutes[channel](
      this.io.sockets,
      JSON.parse(msg)
    ).catch(e => {console.error('onRedisMessage error', e, e.stack)})
  }

  async validateAndEmitMessage(sockets, room, type, json, post) {
    if (!(room in sockets.adapter.rooms)) {
      return
    }

    let clientIds = Object.keys(sockets.adapter.rooms[room])

    await Promise.all(clientIds.map(async (clientId) => {
      let socket = sockets.connected[clientId]
      let user = socket.user

      let valid = await post.validateCanShow(user.id)

      if (valid)
        socket.emit(type, json)
    }))
  }

  // Message-handlers follow
  async onPostDestroy(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = { meta: { postId: data.postId } }

    sockets.in('timeline:' + data.timelineId).emit('post:destroy', json)
    sockets.in('post:' + data.postId).emit('post:destroy', json)

    let type = 'post:destroy'
    let room = `timeline:${data.timelineId}`
    await this.validateAndEmitMessage(sockets, room, type, json, post)

    room = `post:${data.postId}`
    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onPostNew(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = await new models.PostSerializer(post).promiseToJSON()

    let type = 'post:new'
    let room = `timeline:${data.timelineId}`

    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onPostUpdate(sockets, data) {
    let post = await models.Post.findById(data.postId)
    let json = await new models.PostSerializer(post).promiseToJSON()

    let type = 'post:update'
    let room

    if (data.timelineId) {
      room = `timeline:${data.timelineId}`
    } else {
      room = `post:${data.postId}`
    }
    await this.validateAndEmitMessage(sockets, room, type, json, post)
  }

  async onCommentNew(sockets, data) {
    let comment = await models.Comment.findById(data.commentId)

    if (!comment) {
      // might be outdated event
      return
    }

    let post = await models.Post.findById(comment.postId)
    let json = await new models.PubsubCommentSerializer(comment).promiseToJSON()

    let type = 'comment:new'
    let room

    if (data.timelineId) {
{'fraction_non_alphanumeric': 0.13791554357592092, 'fraction_numerical': 0.0, 'mean_word_length': 2.982468694096601, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '4402278', 'n_tokens_mistral': 3621, 'n_tokens_neox': 3566, 'n_words': 711}