<commit_before>{EventEmitter} = require 'microemitter'

module.exports = class AuthWorker extends EventEmitter

  AuthedClient = require './authedclient'

  AUTH_EXCHANGE_OPTIONS =
    type        : 'fanout'
    autoDelete  : yes

  REROUTING_EXCHANGE_OPTIONS =
  USERS_PRESENCE_CONTROL_EXCHANGE_OPTIONS =
    type        : 'fanout'
    autoDelete  : yes

  NOTIFICATION_EXCHANGE_OPTIONS =
    type        : 'topic'
    autoDelete  : yes

  constructor: (@bongo, options = {}) ->

    # instance options
    { @servicesPresenceExchange, @reroutingExchange
      @notificationExchange, @usersPresenceControlExchange
      @presenceTimeoutAmount, @authExchange, @authAllExchange } = options

    # initialize defaults:
    @servicesPresenceExchange     ?= 'services-presence'
    @usersPresenceControlExchange ?= 'users-presence-control'
    @reroutingExchange            ?= 'routing-control'
    @notificationExchange         ?= 'notification'
    @authExchange                 ?= 'auth'
    @authAllExchange              ?= 'authAll'
    @presenceTimeoutAmount        ?= 1000 * 60 * 2 # 2 min

    # instance state
    @services         = {}
    @clients          =
      bySocketId      : {}
      byExchange      : {}
      byRoutingKey    : {}
    @usersBySocketId  = {}
    @counts           = {}
    @waitingAuthWhos  = {}

  bound: require 'koding-bound'

  authenticate: (messageData, routingKey, socketId, callback) ->
    {clientId, channel, event} = messageData
    @requireSession clientId, routingKey, socketId, callback

  requireSession: (clientId, routingKey, socketId, callback) ->
    {JSession} = @bongo.models
    JSession.fetchSession clientId, (err, session) =>
      if err? or not session? then @rejectClient routingKey
      else
        @addUserSocket session.username, socketId  if session.username?
        tokenHasChanged = session.clientId isnt clientId
        @updateSessionToken session.clientId, routingKey  if tokenHasChanged
        callback session

  addUserSocket: (username, socketId) ->
    @usersBySocketId[socketId] = username
    @fetchUserPresenceControlExchange (exchange) ->
      exchange.publish 'auth.join', { username, socketId }

  removeUserSocket: (socketId) ->
    username = @usersBySocketId[socketId]
    return  unless username
    delete @usersBySocketId[username]
    @fetchUserPresenceControlExchange (exchange) ->
      exchange.publish 'auth.leave', { username, socketId }

  updateSessionToken: (clientId, routingKey) ->
    @bongo.respondToClient routingKey,
      method      : 'updateSessionToken'
      arguments   : [clientId]
      callbacks   : {}

  getNextServiceInfo: (serviceType) ->
    count = @counts[serviceType] ?= 0
    servicesOfType = @services[serviceType]
    return  unless servicesOfType?.length
    serviceInfo = servicesOfType[count % servicesOfType.length]
    @counts[serviceType] += 1
    return serviceInfo

  addService: ({serviceGenericName, serviceUniqueName, loadBalancing}) ->
    servicesOfType = @services[serviceGenericName] ?= []
    servicesOfType.push {serviceUniqueName, serviceGenericName, loadBalancing}

  removeService: ({serviceGenericName, serviceUniqueName}) ->
    servicesOfType = @services[serviceGenericName]
    [index] = (i for s, i in servicesOfType \
                 when s.serviceUniqueName is serviceUniqueName)
    servicesOfType.splice index, 1
    clientsByExchange = @clients.byExchange[serviceUniqueName]
    clientsByExchange?.forEach @bound 'cycleClient'

  cycleClient: (client) ->
    {routingKey} = client
    @bongo.respondToClient routingKey,
      method      : 'cycleChannel'
      arguments   : []
      callbacks   : {}

  removeClient: (rest...) ->
    if rest.length is 1
      [client] = rest
      return @removeClient client.socketId, client.exchange, client.routingKey
    [socketId, exchange, routingKey] = rest
    delete @clients.bySocketId[socketId]
    delete @clients.byExchange[exchange]
    delete @clients.byRoutingKey[routingKey]

  addClient: (socketId, exchange, routingKey, sendOk=yes) ->
    if sendOk
      @bongo.respondToClient routingKey,
        method    : 'auth.authOk'
        arguments : []
        callbacks : {}
    clientsBySocketId   = @clients.bySocketId[socketId]     ?= []
    clientsByExchange   = @clients.byExchange[exchange]     ?= []
    clientsByRoutingKey = @clients.byRoutingKey[routingKey] ?= []
    client = new AuthedClient { routingKey, socketId, exchange }
    clientsBySocketId.push client
    clientsByRoutingKey.push client
    clientsByExchange.push client

  rejectClient:(routingKey, message)->
    console.log 'rejecting', routingKey
    return console.trace()  unless routingKey?
    @bongo.respondToClient routingKey,
      method    : 'error'
      arguments : [message: message ? 'Access denied']
      callbacks : {}

  setSecretNames:(routingKey, publishingName, subscribingName)->
    setSecretNamesEvent = "#{routingKey}.setSecretNames"
    message = JSON.stringify { publishingName, subscribingName }
    @bongo.respondToClient setSecretNamesEvent, message

  publishToService: (exchangeName, routingKey, payload, callback) ->
    { connection } = @bongo.mq
    connection.exchange exchangeName, AUTH_EXCHANGE_OPTIONS,
      (exchange) =>
        exchange.publish routingKey, payload
        exchange.close() # don't leak a channel
        callback? null

  sendAuthMessage: (options) ->
    { serviceUniqueName, serviceGenericName, routingKey, method, callback
    username, correlationName, socketId, deadService } = options

    params = { routingKey, username, correlationName
               serviceGenericName, deadService }

    @publishToService serviceUniqueName, method, params, callback

  sendAuthJoin: (options) ->
    { socketId, serviceUniqueName, routingKey } = options
    options.callback = =>
      key = getWaitingAuthWhoKey options
      socketId ?= @waitingAuthWhos[key]
      delete @waitingAuthWhos[key]
      @addClient socketId, serviceUniqueName, routingKey
    options.method = 'auth.join'
    @sendAuthMessage options

  getWaitingAuthWhoKey = (o) ->
    "#{o.username}!#{o.correlationName}!#{o.serviceGenericName}"

  sendAuthWho: (options) ->
    options.method = 'auth.who'
    @waitingAuthWhos[getWaitingAuthWhoKey options] = options.socketId
    @sendAuthMessage options

  fetchReroutingExchange:(callback)->
    @bongo.mq.connection.exchange(
      @reroutingExchange
      REROUTING_EXCHANGE_OPTIONS
      callback<commit_msg>auth worker: add replyExchange param<commit_after>{EventEmitter} = require 'microemitter'

module.exports = class AuthWorker extends EventEmitter

  AuthedClient = require './authedclient'

  AUTH_EXCHANGE_OPTIONS =
    type        : 'fanout'
    autoDelete  : yes

  REROUTING_EXCHANGE_OPTIONS =
  USERS_PRESENCE_CONTROL_EXCHANGE_OPTIONS =
    type        : 'fanout'
    autoDelete  : yes

  NOTIFICATION_EXCHANGE_OPTIONS =
    type        : 'topic'
    autoDelete  : yes

  constructor: (@bongo, options = {}) ->

    # instance options
    { @servicesPresenceExchange, @reroutingExchange
      @notificationExchange, @usersPresenceControlExchange
      @presenceTimeoutAmount, @authExchange, @authAllExchange } = options

    # initialize defaults:
    @servicesPresenceExchange     ?= 'services-presence'
    @usersPresenceControlExchange ?= 'users-presence-control'
    @reroutingExchange            ?= 'routing-control'
    @notificationExchange         ?= 'notification'
    @authExchange                 ?= 'auth'
    @authAllExchange              ?= 'authAll'
    @presenceTimeoutAmount        ?= 1000 * 60 * 2 # 2 min

    # instance state
    @services         = {}
    @clients          =
      bySocketId      : {}
      byExchange      : {}
      byRoutingKey    : {}
    @usersBySocketId  = {}
    @counts           = {}
    @waitingAuthWhos  = {}

  bound: require 'koding-bound'

  authenticate: (messageData, routingKey, socketId, callback) ->
    {clientId, channel, event} = messageData
    @requireSession clientId, routingKey, socketId, callback

  requireSession: (clientId, routingKey, socketId, callback) ->
    {JSession} = @bongo.models
    JSession.fetchSession clientId, (err, session) =>
      if err? or not session? then @rejectClient routingKey
      else
        @addUserSocket session.username, socketId  if session.username?
        tokenHasChanged = session.clientId isnt clientId
        @updateSessionToken session.clientId, routingKey  if tokenHasChanged
        callback session

  addUserSocket: (username, socketId) ->
    @usersBySocketId[socketId] = username
    @fetchUserPresenceControlExchange (exchange) ->
      exchange.publish 'auth.join', { username, socketId }

  removeUserSocket: (socketId) ->
    username = @usersBySocketId[socketId]
    return  unless username
    delete @usersBySocketId[username]
    @fetchUserPresenceControlExchange (exchange) ->
      exchange.publish 'auth.leave', { username, socketId }

  updateSessionToken: (clientId, routingKey) ->
    @bongo.respondToClient routingKey,
      method      : 'updateSessionToken'
      arguments   : [clientId]
      callbacks   : {}

  getNextServiceInfo: (serviceType) ->
    count = @counts[serviceType] ?= 0
    servicesOfType = @services[serviceType]
    return  unless servicesOfType?.length
    serviceInfo = servicesOfType[count % servicesOfType.length]
    @counts[serviceType] += 1
    return serviceInfo

  addService: ({serviceGenericName, serviceUniqueName, loadBalancing}) ->
    servicesOfType = @services[serviceGenericName] ?= []
    servicesOfType.push {serviceUniqueName, serviceGenericName, loadBalancing}

  removeService: ({serviceGenericName, serviceUniqueName}) ->
    servicesOfType = @services[serviceGenericName]
    [index] = (i for s, i in servicesOfType \
                 when s.serviceUniqueName is serviceUniqueName)
    servicesOfType.splice index, 1
    clientsByExchange = @clients.byExchange[serviceUniqueName]
    clientsByExchange?.forEach @bound 'cycleClient'

  cycleClient: (client) ->
    {routingKey} = client
    @bongo.respondToClient routingKey,
      method      : 'cycleChannel'
      arguments   : []
      callbacks   : {}

  removeClient: (rest...) ->
    if rest.length is 1
      [client] = rest
      return @removeClient client.socketId, client.exchange, client.routingKey
    [socketId, exchange, routingKey] = rest
    delete @clients.bySocketId[socketId]
    delete @clients.byExchange[exchange]
    delete @clients.byRoutingKey[routingKey]

  addClient: (socketId, exchange, routingKey, sendOk=yes) ->
    if sendOk
      @bongo.respondToClient routingKey,
        method    : 'auth.authOk'
        arguments : []
        callbacks : {}
    clientsBySocketId   = @clients.bySocketId[socketId]     ?= []
    clientsByExchange   = @clients.byExchange[exchange]     ?= []
    clientsByRoutingKey = @clients.byRoutingKey[routingKey] ?= []
    client = new AuthedClient { routingKey, socketId, exchange }
    clientsBySocketId.push client
    clientsByRoutingKey.push client
    clientsByExchange.push client

  rejectClient:(routingKey, message)->
    console.log 'rejecting', routingKey
    return console.trace()  unless routingKey?
    @bongo.respondToClient routingKey,
      method    : 'error'
      arguments : [message: message ? 'Access denied']
      callbacks : {}

  setSecretNames:(routingKey, publishingName, subscribingName)->
    setSecretNamesEvent = "#{routingKey}.setSecretNames"
    message = JSON.stringify { publishingName, subscribingName }
    @bongo.respondToClient setSecretNamesEvent, message

  publishToService: (exchangeName, routingKey, payload, callback) ->
    { connection } = @bongo.mq
    connection.exchange exchangeName, AUTH_EXCHANGE_OPTIONS,
      (exchange) =>
        exchange.publish routingKey, payload
        exchange.close() # don't leak a channel
        callback? null

  sendAuthMessage: (options) ->
    { serviceUniqueName, serviceGenericName, routingKey, method, callback
    username, correlationName, socketId, deadService } = options

    params = { routingKey, username, correlationName
               serviceGenericName, deadService, replyExchange: @authExchange }

    @publishToService serviceUniqueName, method, params, callback

  sendAuthJoin: (options) ->
    { socketId, serviceUniqueName, routingKey } = options
    options.callback = =>
      key = getWaitingAuthWhoKey options
      socketId ?= @waitingAuthWhos[key]
      delete @waitingAuthWhos[key]
      @addClient socketId, serviceUniqueName, routingKey
    options.method = 'auth.join'
    @sendAuthMessage options

  getWaitingAuthWhoKey = (o) ->
    "#{o.username}!#{o.correlationName}!#{o.serviceGenericName}"

  sendAuthWho: (options) ->
    options.method = 'auth.who'
    @waitingAuthWhos[getWaitingAuthWhoKey options] = options.socketId
    @sendAuthMessage options

  fetchReroutingExchange:(callback)->
    @bongo.mq.connection.exchange(
      @reroutingExchange
      REROUTING_EXCHANGE_OPTIONS
      callback
{'fraction_non_alphanumeric': 0.09643379804359548, 'fraction_numerical': 0.0018485712085034277, 'mean_word_length': 3.352665102246061, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 53, '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': '2964904', 'n_tokens_mistral': 3827, 'n_tokens_neox': 3675, 'n_words': 781}