//
//   Copyright 2014 Ilkka Oksanen <iao@iki.fi>
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing,
//   software distributed under the License is distributed on an "AS
//   IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//   express or implied.  See the License for the specific language
//   governing permissions and limitations under the License.
//

'use strict';

const util = require('util'),
      _ = require('lodash'),
      redisModule = require('./redis'),
      redis = redisModule.createClient();

exports.send = async function(userId, sessionId, ntfs) {
    await queueNotifications(userId, sessionId, null, ntfs);
};

exports.broadcast = async function(userId, ntfs, excludeSessionId) {
    await queueNotifications(userId, null, excludeSessionId, ntfs);
};

exports.receive = async function(userId, sessionId, timeout) {
    let command;
    let commands = [];

    let client = redisModule.createClient({ autoClose: false });
    let result = await client.brpop(`outbox:${userId}:${sessionId}`, timeout);

    if (result) {
        commands.push(result[1]);
    }

    // Retrieve other ntfs if there are any
    while ((command = await client.rpop(`outbox:${userId}:${sessionId}`)) !== null) {
        commands.push(command);
    }

    await client.quit();

    return commands.map(function(value) {
        return JSON.parse(value);
    });
};

exports.requeue = async function(userId, sessionId, ntfs) {
    ntfs = ntfs.map(function(ntf) {
        return JSON.stringify(ntf);
    });

    await redis.rpush.apply(redis, [ `outbox:${userId}:${sessionId}` ].concat(ntfs.reverse()));
};

exports.communicateNewUserIds = async function(userId, sessionId, msg) {
    // TBD: Refactor to a separate module
    await handleNewUserIds(userId, sessionId, null, [ msg ]);
};

async function queueNotifications(userId, sessionId, excludeSessionId, ntfs) {
    if (!util.isArray(ntfs)) {
        ntfs = [ ntfs ];
    }

    await handleNewUserIds(userId, sessionId, excludeSessionId, ntfs);

    ntfs = ntfs.map(function(value) {
        return typeof(value) === 'string' ? value : JSON.stringify(value);
    });

    await redis.run.apply(null,
        [ 'queueOutbox', userId, sessionId, excludeSessionId ].concat(ntfs));
}

async function handleNewUserIds(userId, sessionId, excludeSessionId, ntfs) {
    let allUserIds = [];

    ntfs.forEach(function(command) {
        allUserIds = allUserIds.concat(scanUserIds(command));
    });

    allUserIds = _.uniq(allUserIds);
    await redis.run.apply(null,
        [ 'introduceNewUserIds', userId, sessionId, excludeSessionId, false ].concat(allUserIds));
}

function scanUserIds(obj) {
    let res = [];

    if (typeof(obj) === 'string') {
        obj = JSON.parse(obj);
    }

    for (let key in obj) {
        let value = obj[key];

        if (typeof value === 'object') {
            res = res.concat(scanUserIds(value));
        } else if (key === 'userId' && value) {
            res.push(value);
        }
    }

    return res;
}

@@ -636,16 +636,49 @@
 trict';%0A
+const assert = require('assert');
 %0Aconst u
@@ -838,34 +838,32 @@
 nc function(user
-Id
 , sessionId, ntf
@@ -892,34 +892,32 @@
 tifications(user
-Id
 , sessionId, nul
@@ -969,18 +969,16 @@
 ion(user
-Id
 , ntfs, 
@@ -1030,18 +1030,16 @@
 ons(user
-Id
 , null, 
@@ -1096,34 +1096,32 @@
 nc function(user
-Id
 , sessionId, tim
@@ -1276,32 +1276,43 @@
 p(%60outbox:$%7Buser
+.globalUser
 Id%7D:$%7BsessionId%7D
@@ -1474,32 +1474,43 @@
 p(%60outbox:$%7Buser
+.globalUser
 Id%7D:$%7BsessionId%7D
@@ -1618,305 +1618,33 @@
 map(
-function(value) %7B%0A        return JSON.parse(value);%0A    %7D);%0A%7D;%0A%0Aexports.requeue = async function(userId, sessionId, ntfs) %7B%0A    ntfs = ntfs.map(function(ntf) %7B%0A        return JSON.stringify(ntf);%0A    %7D);%0A%0A    await redis.rpush.apply(redis, %5B %60outbox:$%7BuserId%7D:$%7BsessionId%7D%60 %5D.concat(ntfs.reverse()
+value =%3E JSON.parse(value
 ));%0A
@@ -1694,26 +1694,24 @@
 unction(user
-Id
 , sessionId,
@@ -1783,34 +1783,32 @@
 eNewUserIds(user
-Id
 , sessionId, nul
@@ -1858,26 +1858,24 @@
 cations(user
-Id
 , sessionId,
@@ -1897,24 +1897,43 @@
 Id, ntfs) %7B%0A
+    assert(ntfs);%0A%0A
     if (!uti
@@ -2006,34 +2006,32 @@
 eNewUserIds(user
-Id
 , sessionId, exc
@@ -2194,39 +2194,17 @@
 edis.run
-.apply(null,%0A        %5B 
+(
 'queueOu
@@ -2210,24 +2210,35 @@
 utbox', user
+.globalUser
 Id, sessionI
@@ -2260,24 +2260,18 @@
 onId
- %5D.concat(
+, ...
 ntfs)
-)
 ;%0A%7D%0A
@@ -2303,26 +2303,24 @@
 UserIds(user
-Id
 , sessionId,
@@ -2540,28 +2540,17 @@
 .run
-.apply(null,
+(
 %0A       
  %5B '
@@ -2545,18 +2545,16 @@
 %0A       
- %5B
  'introd
@@ -2573,16 +2573,27 @@
 s', user
+.globalUser
 Id, sess
@@ -2626,18 +2626,13 @@
 alse
- %5D.concat(
+, ...
 allU
@@ -2638,17 +2638,16 @@
 UserIds)
-)
 ;%0A%7D%0A%0Afun

Use user object instead of userId