/*
 * Copyright 2014 Apereo Foundation (AF) Licensed under the
 * Educational Community 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://opensource.org/licenses/ECL-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.
 */

var _ = require('underscore');
var assert = require('assert');
var ShortId = require('shortid');
var util = require('util');

var MQ = require('oae-util/lib/mq');
var TaskQueue = require('oae-util/lib/taskqueue');

describe('MQ', function() {

    /**
     * Some options that can be used to bind to a message queue.
     */
    var purgeQueueOptions = {
        'subscribe': {
            'prefetchCount': 1
        },
        'queue': {
            'durable': false
        }
    };

    /**
     * Verify that re-initializing the MQ doesn't invoke an error
     */
    it('verify re-initialization is safe', function(callback) {
        // Ensure processing continues, and that MQ is still stable with the tests that follow
        MQ.init({}, function(err) {
            assert.ok(!err);
            return callback();
        });
    });

    describe('#purge()', function() {

        /**
         * Test that verifies the parameters
         */
        it('verify parameter validation', function(callback) {
            var name = util.format('testQueue-%s', ShortId.generate());
            MQ.purge(name, function(err) {
                assert.equal(err.code, 400);
                return callback();
            });
        });

        /**
         * Verify that a queue can be purged of its tasks.
         */
        it ('verify a queue can be purged', function(callback) {
            var called = 0;
            var taskHandler = function(data, taskCallback) {
                called++;
                setTimeout(taskCallback, 2000);
            };

            var testQueue = 'testQueue-' + new Date().getTime();
            TaskQueue.bind(testQueue, taskHandler, purgeQueueOptions, function() {
                // Submit a couple of tasks.
                for (var i = 0; i < 10; i++) {
                    TaskQueue.submit(testQueue, { 'foo': 'bar' });
                }

                // Purge the queue.
                MQ.purge(testQueue, function() {
                    // Because of the asynchronous nature of node/rabbitmq it's possible that a task gets delivered
                    // before the purge command is processed.
                    // That means we should have only handled at most 1 task.
                    assert.ok(called <= 1);
                    callback();
                });
            });
        });
    });

    describe('#purgeAll()', function() {

        /**
         * Verify that all known queues can be purged of its tasks.
         */
        it ('verify all queues can be purged', function(callback) {
            var called = {'a': 0, 'b': 0};
            var taskHandler = function(data, taskCallback) {
                called[data.queue]++;
                setTimeout(taskCallback, 2000);
            };

            var testQueueA = 'testQueueA-' + new Date().getTime();
            var testQueueB = 'testQueueB-' + new Date().getTime();
            TaskQueue.bind(testQueueA, taskHandler, purgeQueueOptions, function() {
                TaskQueue.bind(testQueueB, taskHandler, purgeQueueOptions, function() {
                    // Submit a couple of tasks.
                    for (var i = 0; i < 10; i++) {
                        TaskQueue.submit(testQueueA, { 'queue': 'a' });
                        TaskQueue.submit(testQueueB, { 'queue': 'b' });
                    }

                    // Purge all the queues.
                    MQ.purgeAll(function() {
                        // Because of the asynchronous nature of node/rabbitmq it's possible that a task gets delivered
                        // before the purge command is processed.
                        // That means we should have only handled at most 1 task.
                        assert.ok(called['a'] <= 1);
                        assert.ok(called['b'] <= 1);
                        callback();
                    });
                });
            });
        });
    });

    describe('#declareExchange()', function() {

        /**
         * Test that verifies that the parameters are validated
         */
        it('verify parameter validation', function(callback) {
            MQ.declareExchange(null, {'durable': false, 'autoDelete': true}, function(err) {
                assert.equal(err.code, 400);

                // Sanity check
                var exchangeName = util.format('testExchange-%s', ShortId.generate());
                MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    return callback();
                });
            });
        });
        /**
         * Test that verifies that exchanges cannot be declared twice
         */
        it('verify exchanges cannot be declared twice', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.equal(err.code, 400);
                    return callback();
                });
            });
        });
    });

    describe('#declareQueue()', function() {

        /**
         * Test that verifies that the parameters are validated
         */
        it('verify parameter validation', function(callback) {
            MQ.declareQueue(null, {'durable': false, 'autoDelete': true}, function(err) {
                assert.equal(err.code, 400);

                // Sanity check
                var queueName = util.format('testQueue-%s', ShortId.generate());
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    return callback();
                });
            });
        });

        /**
         * Test that verifies that queues cannot be declared twice
         */
        it('verify queues cannot be declared twice', function(callback) {
            var queueName = util.format('testQueue-%s', ShortId.generate());
            MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.equal(err.code, 400);
                    return callback();
                });
            });
        });
    });

    describe('#isQueueDeclared()', function() {

        /**
         * Test that verifies that it can be retrieved whether or not queues are declared
         */
        it('verify isQueueDeclared works', function(callback) {
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var exchangeName = util.format('testExchange-%s', ShortId.generate());

            var isDeclared = MQ.isQueueDeclared(queueName);
            assert.strictEqual(isDeclared, false);

            MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);

                isDeclared = MQ.isQueueDeclared(queueName);
                assert.strictEqual(isDeclared, true);
                return callback();
            });
        });
    });

    describe('#bindQueueToExchange()', function() {

        /**
         * Test that verifies that the parameters are validated
         */
        it('verify parameter validation', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    MQ.bindQueueToExchange(null, exchangeName, routingKey, function(err) {
                        assert.equal(err.code, 400);
                        MQ.bindQueueToExchange(queueName, null, routingKey, function(err) {
                            assert.equal(err.code, 400);
                            MQ.bindQueueToExchange(queueName, exchangeName, null, function(err) {
                                assert.equal(err.code, 400);

                                // Sanity check that the queue can be bound
                                MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                                    assert.ok(!err);

                                    // Tidy up after ourselves and remove the binding
                                    MQ.unbindQueueFromExchange(queueName, exchangeName, routingKey, function(err) {
                                        assert.ok(!err);
                                        return callback();
                                    });
                                });
                            });
                        });
                    });
                });
            });
        });

        /**
         * Test that verifies a queue can be bound to an exchange
         */
        it('verify functionality', function() {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());
            var data = {'text': 'The truth is out there'};

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    var listener = function(msg) {
                        // Verify the message we receive is correct
                        assert.strictEqual(msg.text, data.text);

                        // Unbind the queue so both the queue and exchange will go away when we restart rabbitmq-server
                        MQ.unbindQueueFromExchange(queueName, exchangeName, routingKey, function(err) {
                            assert.ok(!err);
                            return callback();
                        });
                    };
                    MQ.subscribeQueue(queueName, {}, listener, function(err) {
                        assert.ok(!err);

                        MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                            assert.ok(!err);

                            MQ.submit(exchangeName, routingKey, data);
                        });
                    });
                });
            });
        });

        /**
         * Test that verifies you can bind queues to exchanges in parallel
         */
        it('verify you can bind queues to exchanges in parallel', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var data = {'text': 'The truth is out there'};
            var routingKeys = [];
            for (var i = 0; i < 100; i++) {
                routingKeys.push('key-' + i);
            }

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);

                    // This test passes if we receive a message from RabbitMQ for each of our routing keys
                    var receivedMessage = _.after(routingKeys.length, function(message) {
                        return callback();
                    });

                    // Subscribe for incoming messages
                    MQ.subscribeQueue(queueName, {}, receivedMessage, function(err) {
                        assert.ok(!err);

                        // When our queue is bound for all routing keys, we will submit a message for each one
                        var queueBound = _.after(routingKeys.length, function() {
                            _.each(routingKeys, function(routingKey) {
                                MQ.submit(exchangeName, routingKey, data);
                            });
                        });

                        // Bind our queue for all routing keys
                        _.each(routingKeys, function(routingKey) {
                            MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                                assert.ok(!err);
                                queueBound();
                            });
                        });
                    });
                });
            });
        });
    });

    describe('#unbindQueueFromExchange()', function() {

        /**
         * Test that verifies that the parameters are validated
         */
        it('verify parameter validation', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                        assert.ok(!err);
                        MQ.unbindQueueFromExchange(null, exchangeName, routingKey, function(err) {
                            assert.equal(err.code, 400);
                            MQ.unbindQueueFromExchange(queueName, null, routingKey, function(err) {
                                assert.equal(err.code, 400);
                                MQ.unbindQueueFromExchange(queueName, exchangeName, null, function(err) {
                                    assert.equal(err.code, 400);

                                    // Sanity-check and tidy up
                                    MQ.unbindQueueFromExchange(queueName, exchangeName, routingKey, function(err) {
                                        assert.ok(!err);
                                        return callback();
                                    });
                                });
                            });
                        });
                    });
                });
            });
        });

        /**
         * Test that verifies a queue can be unbound from an exchange
         */
        it('verify functionality', function() {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());
            var data = {'text': 'The truth is out there'};

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);
                MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    var handledMessages = 0;
                    var listener = function(msg) {
                        handledMessages++;

                        // We should only receive one message
                        assert.strictEqual(handledMessages, 1);

                        // Verify the message we receive is correct
                        assert.strictEqual(msg.text, data.text);
                    };
                    MQ.subscribeQueue(queueName, {}, listener, function(err) {
                        assert.ok(!err);

                        MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                            assert.ok(!err);

                            MQ.submit(exchangeName, routingKey, data, function() {

                                // Unbind the queue from the exchange, we should no longer receive any messages
                                MQ.unbindQueueFromExchange(queueName, exchangeName, routingKey, function(err) {
                                    assert.ok(!err);

                                    // Submit one more message. If it ends up at our listener the test will fail
                                    MQ.submit(exchangeName, routingKey, data, function() {
                                        return callback();
                                    });
                                });
                            });
                        });
                    });
                });
            });
        });
    });

    describe('#submit()', function() {

        /**
         * Test that verifies the passed in parameters
         */
        it('verify parameter validation', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());
            var data = {'text': 'The truth is out there'};

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);

                // An exchange must be provided
                MQ.submit(null, routingKey, data, null, function(err) {
                    assert.equal(err.code, 400);

                    // A routing-key must be provided
                    MQ.submit(exchangeName, null, data, null, function(err) {
                        assert.equal(err.code, 400);

                        // Sanity check
                        MQ.submit(exchangeName, routingKey, data, null, function(err) {
                            assert.ok(!err);
                            return callback();
                        });
                    });
                });
            });
        });

        /**
         * Test that verifies that the callback function in the submit handler is properly executed
         */
        it('verify callback', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());
            var data = {'text': 'The truth is out there'};

            MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                assert.ok(!err);

                var noConfirmCalled = 0;
                MQ.submit(exchangeName, routingKey, data, null, function(err) {
                    assert.ok(!err);

                    // This should only be executed once
                    noConfirmCalled++;
                    assert.equal(noConfirmCalled, 1);

                    // Declare an exchange that acknowledges the message
                    exchangeName = util.format('testExchange-%s', ShortId.generate());
                    MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true, 'confirm': true}, function(err) {
                        assert.ok(!err);

                        var confirmCalled = 0;
                        MQ.submit(exchangeName, routingKey, data, null, function(err) {
                            assert.ok(!err);

                            // This should only be executed once
                            confirmCalled++;
                            assert.equal(confirmCalled, 1);
                            return callback();
                        });
                    });
                });
            });
        });

        /**
         * Test that verifies when an amqp message is redelivered (rejected or failed), it gets sent into a
         * redelivery queue for manual intervention, rather than refiring the listener
         */
        it('verify redelivered messages are not re-executed', function(callback) {
            var exchangeName = util.format('testExchange-%s', ShortId.generate());
            var queueName = util.format('testQueue-%s', ShortId.generate());
            var routingKey = util.format('testRoutingKey-%s', ShortId.generate());

            // Make sure the redeliver queue is empty to start
            MQ.purge('oae-util-mq-redeliverqueue', function(err) {
                assert.ok(!err);

                // Create the exchange and queue on which we'll deliver a message and reject it
                MQ.declareExchange(exchangeName, {'durable': false, 'autoDelete': true}, function(err) {
                    assert.ok(!err);
                    MQ.declareQueue(queueName, {'durable': false, 'autoDelete': true}, function(err) {
                        assert.ok(!err);

                        // A listener that ensures it only handles the rejected message once
                        var handledMessages = 0;
                        var listener = function(msg, callback) {
                            handledMessages++;
                            if (handledMessages > 1) {
                                // Throw in a new tick to ensure it doesn't get caught by MQ for automatic acknowledgement
                                process.nextTick(function() {
                                    assert.fail('Should only have handled the message at most once');
                                });
                            }
                        };

                        // Subscribe to the queue and allow it to start accepting messages on the exchange
                        MQ.subscribeQueue(queueName, {'ack': true}, listener, function(err) {
                            assert.ok(!err);
                            MQ.bindQueueToExchange(queueName, exchangeName, routingKey, function(err) {
                                assert.ok(!err);

                                // Submit a message that we can handle
                                MQ.submit(exchangeName, routingKey, {'data': 'test'}, null, function(err) {
                                    assert.ok(!err);
                                });

                                // When the raw message comes in, reject it so it gets redelivered
                                _bindPreHandleOnce(queueName, function(_queueName, data, headers, deliveryInfo, message) {

                                    // Reject the message, indicating that we want it requeued and redelivered
                                    message.reject(true);

                                    // Ensure that rabbitmq intercepts the redelivery of the rejected message and stuffs it in the redelivery queue
                                    // for manual intervention
                                    MQ.once('storedRedelivery', function(_queueName) {
                                        // Here we make sure that the listener received the message the first time. But this does not
                                        // ensure it doesn't receive it the second time. That is what the `assert.fail` is for in the
                                        // listener
                                        assert.equal(handledMessages, 1);
                                        assert.equal(queueName, _queueName);

                                        // Make sure we can take the item off the redelivery queue
                                        MQ.subscribeQueue('oae-util-mq-redeliverqueue', {'prefetchCount': 1}, function(data, listenerCallback) {
                                            assert.ok(data);
                                            assert.ok(data.headers);
                                            assert.equal(data.deliveryInfo.queue, queueName);
                                            assert.equal(data.deliveryInfo.exchange, exchangeName);
                                            assert.equal(data.deliveryInfo.routingKey, routingKey);
                                            assert.equal(data.data.data, 'test');

                                            // Don't accept any more messages on this queue
                                            MQ.unsubscribeQueue('oae-util-mq-redeliverqueue', function(err) {
                                                assert.ok(!err);

                                                // Acknowledge the redelivered message so it doesn't go in an infinite redelivery loop
                                                listenerCallback();

                                                return callback();
                                            });
                                        });
                                    });
                                });
                            });
                        });
                    });
                });
            });
        });
    });
});

/**
 * Bind a listener to the MQ preHandle event for a particular queue name. The bound function
 * will be unbound immediately after the first message on the queue is received.
 *
 * @param  {String}     handlingQueueName   The name of the queue on which to listen to a message
 * @param  {Function}   handler             The listener to invoke when a message comes. Same as the MQ event `preHandle`
 * @api private
 */
var _bindPreHandleOnce = function(handlingQueueName, handler) {

    /*!
     * Filters tasks by those on the expected queue, and immediately unbinds the
     * handler so it only gets invoked once. The parameters are the MQ preHandle
     * event parameters.
     */
    var _handler = function(queueName, data, headers, deliveryInfo, message) {
        if (queueName !== handlingQueueName) {
            return;
        }

        MQ.removeListener('preHandle', _handler);
        return handler(queueName, data, headers, deliveryInfo, message);
    };

    MQ.on('preHandle', _handler);
};
