<commit_before>/*!
 * Copyright 2013 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 events = require('events');
var util = require('util');

var Cassandra = require('oae-util/lib/cassandra');
var Locking = require('oae-util/lib/locking');
var log = require('oae-logger').logger('oae-messagebox-api');
var OaeUtil = require('oae-util/lib/util');
var Validator = require('oae-util/lib/validator').Validator;

var MessageBoxModel = require('./model');
var MessageBoxConstants = require('./constants').MessageBoxConstants;

// A contribution will be considered "recent" for 30 days after it occurs
var DURATION_RECENT_CONTRIBUTIONS_SECONDS = 30 * 24 * 60 * 60;

/**
 * ### Events
 *
 * The `MessageBoxAPI`, as enumerated in `MessageboxConstants.events`, emits the following events:
 *
 * * `createdMessage(message)`: A new message was created.
 * * `updatedMessage(messageId, newBody)`: The body of a message has been updated.
 * * `deletedMessage(messageId, deleteType)`: A message has been deleted, deleteType will indicate whether it was a hard or a soft delete.
 */
var MessageBoxAPI = module.exports = new events.EventEmitter();

/**
 * Create a new message in a message box. The message can be either its own "top-level" message, or a reply to
 * another message in this message box. To specify a message is a reply of another, you must supply the
 * `opts.replyToCreated` parameter to indicate the timestamp of the parent message which establishes its
 * hierarchy and ordering in the message box.
 *
 * @param  {String}     messageBoxId            The id of the message box that holds the message, ex: d:cam:dfjOSdfij#discussion would be an example of the messagebox for a discussion thread with id d:cam:dfjOSdfij.
 * @param  {String}     createdBy               The id of the user who sent the message
 * @param  {String}     body                    The body of the message
 * @param  {Object}     [opts]                  Additional optional message attributes
 * @param  {String}     [opts.replyToCreated]   The timestamp (millis since the epoch) that the message to which this is a reply (if applicable) was created
 * @param  {Function}   [callback]              Invoked when the process completes
 * @param  {Object}     [callback.err]          An error that occurred, if any
 * @param  {Message}    [callback.message]      The message model object that was persisted
 */
var createMessage = module.exports.createMessage = function(messageBoxId, createdBy, body, opts, callback) {
    opts = opts || {};

    var validator = new Validator();
    validator.check(messageBoxId, {'code': 400, 'msg': 'A messageBoxId must be specified.'}).notNull();
    validator.check(createdBy, {'code': 400, 'msg': 'The createdBy parameter must be a valid user id.'}).isUserId();
    validator.check(body, {'code': 400, 'msg': 'The body of the message must be specified.'}).notNull();
    if (opts.replyToCreated) {
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it should not be null.'}).notNull();
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it should be an integer.'}).isInt();
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it cannot be in the future.'}).max(Date.now());
    }
    if (validator.hasErrors()) {
        return callback(validator.getFirstError());
    }

    var replyToCreated = OaeUtil.getNumberParam(opts.replyToCreated);
    var replyToMessageId = (replyToCreated) ? _createMessageId(messageBoxId, replyToCreated) : null;

    // Fetch the threadKey of the parent so we can nest under it
    _getMessageThreadKey(replyToMessageId, function(err, replyToThreadKey) {
        if (err) {
            return callback(err);
        }

        // Make sure we have a unique create timestamp
        // Pass in a meaningfull ID on which we can lock
        var id = (replyToThreadKey) ? replyToThreadKey : messageBoxId;
        _lockUniqueTimestamp(id, Date.now(), function(created, lockKey, lockToken) {

            // Data that will be output in diagnostic error messages
            var diagnosticData = {
                'messageBoxId': messageBoxId,
                'createdBy': createdBy,
                'created': created,
                'replyToMessageId': replyToMessageId
            };

            if (replyToMessageId && !replyToThreadKey) {
                // We specified a message that doesn't actually exist in our message box, don't let that happen
                log().error(diagnosticData, 'Reply-to message does not exist');
                return callback({'code': 400, 'msg': 'Reply-to message does not exist'});
            }

            // Derive this message's thread key by appending it to the parent, if applicable. Otherwise, it is a top-level key
            var threadKey = (replyToThreadKey) ? _appendToThreadKey(replyToThreadKey, created) : created + '|';

            // A low-level storage hash that represents this item stored in Cassandra or Redis
            var messageId = _createMessageId(messageBoxId, created);
            var messageStorageHash = {
                'createdBy': createdBy,
                'body': body,
                'threadKey': threadKey
            };

            // Create the query that creates the message object
            var createMessageQuery = Cassandra.constructUpsertCQL('Messages', 'id', messageId, messageStorageHash, 'QUORUM');
            if (!createMessageQuery) {
                log().error(diagnosticData, 'Failed to create a new message query.');
                return callback({'code': 500, 'msg': 'Failed to create a new message'});
            }

            // Create the query that adds the message object to the messagebox
            var indexMessageQuery = {
                'query': 'UPDATE MessageBoxMessages SET ? = ? WHERE messageBoxId = ?',
                'parameters': [threadKey, '1', messageBoxId]
            };

            // Create the query that updates the "recent contributors" to a message box
            var recentContributionsQuery = {
                'query': 'UPDATE MessageBoxRecentContributions USING TTL ' + DURATION_RECENT_CONTRIBUTIONS_SECONDS + ' SET ? = \'1\' WHERE messageBoxId = ?',
                'parameters': [createdBy, messageBoxId]
            };

            // First insert the new message object, if this fails we do not want to update the messagebox index
            Cassandra.runQuery(createMessageQuery.query, createMessageQuery.parameters, function(err) {
                if (err) {
                    return callback(err);
                }

                // Update the messagebox index, so this message will turn up in queries for all messages in the messagebox
                Cassandra.runQuery(indexMessageQuery.query, indexMessageQuery.parameters, function(err) {
                    if (err) {
                        return callback(err);
                    }

                    // Asynchronously update the recent contributions
                    Cassandra.runQuery(recentContributionsQuery.query, recentContributionsQuery.parameters);

                    // Get the expanded Message object, emit it so it can be indexed, and return it to the caller.
                    var message = _storageHashToMessage(messageId, messageStorageHash);
                    MessageBoxAPI.emit(MessageBoxConstants.events.CREATED_MESSAGE, message);
                    return callback(null, message);
                });
            });
            Locking.release(lockKey, lockToken, function(){});
        });
    });
};

/**
 * Locks a timestamp within a message thread key
 *
 * @param  {String}        id                  An id to provide context for the lock
 * @param  {Integer}       timestamp           The timestamp you'd like to try to lock (ms since unix epoch)
 * @param  {Function}      callback            Standard callback function
 * @param  {Object}        callback.timestamp  The timestamp we actually got
 * @param  {Object}        callback.key        The key we used to acquire the lock
 * @param  {String}        callback.lockToken  The lockToken that can be used to release this lock
 * @api private
 */
var _lockUniqueTimestamp = function(id, timestamp, callback) {
    var key = 'oae-messagebox:' + id + ':' + timestamp;
    Locking.acquire(key, 1, function(err, lockToken) {
        if (err) {
            // This should only occur if Redis is down, just return the requested ts
            return callback(timestamp, lockToken);
        }
        if (!lockToken) {
            // Someone else has the requested ts, try to lock one ms later
            return _lockUniqueTimestamp(id, timestamp + 1, callback);
        }
        // Successful lock, return the details
        return callback(timestamp, key, lockToken);
    });
};

/**
 * Update the content body of a message.
 *
 * @param  {String}         messageBoxId    The id message whose body we will update
 * @param  {String|Number}  created         The created timestamp (in millis since epoch) that we wish to update
 * @param  {String}         newBody         The new message content to push to the body
 * @param  {Function}       callback        Invoked when the process completes
 * @param  {Object}         callback.err    An error that occurred, if any
 */
var updateMessageBody = module.exports.updateMessageBody = function(messageBoxId, created, newBody, callback) {
    var validator = new Validator();
    validator.check(messageBoxId, {'code': 400, 'msg': 'A messageBoxId must be specified.'}).notNull();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be specified.'}).notNull();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be a valid timestamp (integer).'}).isInt();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be a valid timestamp (integer) that is not in the future.'}).max(Date.now());
    validator.check(newBody, {'code': 400, 'msg': 'The new body of the message must be specified.'}).notNull();
    if (validator.hasErrors()) {<commit_msg>Followed up on message timestamp locking<commit_after>/*!
 * Copyright 2013 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 events = require('events');
var util = require('util');

var Cassandra = require('oae-util/lib/cassandra');
var Locking = require('oae-util/lib/locking');
var log = require('oae-logger').logger('oae-messagebox-api');
var OaeUtil = require('oae-util/lib/util');
var Validator = require('oae-util/lib/validator').Validator;

var MessageBoxModel = require('./model');
var MessageBoxConstants = require('./constants').MessageBoxConstants;

// A contribution will be considered "recent" for 30 days after it occurs
var DURATION_RECENT_CONTRIBUTIONS_SECONDS = 30 * 24 * 60 * 60;

/**
 * ### Events
 *
 * The `MessageBoxAPI`, as enumerated in `MessageboxConstants.events`, emits the following events:
 *
 * * `createdMessage(message)`: A new message was created.
 * * `updatedMessage(messageId, newBody)`: The body of a message has been updated.
 * * `deletedMessage(messageId, deleteType)`: A message has been deleted, deleteType will indicate whether it was a hard or a soft delete.
 */
var MessageBoxAPI = module.exports = new events.EventEmitter();

/**
 * Create a new message in a message box. The message can be either its own "top-level" message, or a reply to
 * another message in this message box. To specify a message is a reply of another, you must supply the
 * `opts.replyToCreated` parameter to indicate the timestamp of the parent message which establishes its
 * hierarchy and ordering in the message box.
 *
 * @param  {String}     messageBoxId            The id of the message box that holds the message, ex: d:cam:dfjOSdfij#discussion would be an example of the messagebox for a discussion thread with id d:cam:dfjOSdfij.
 * @param  {String}     createdBy               The id of the user who sent the message
 * @param  {String}     body                    The body of the message
 * @param  {Object}     [opts]                  Additional optional message attributes
 * @param  {String}     [opts.replyToCreated]   The timestamp (millis since the epoch) that the message to which this is a reply (if applicable) was created
 * @param  {Function}   [callback]              Invoked when the process completes
 * @param  {Object}     [callback.err]          An error that occurred, if any
 * @param  {Message}    [callback.message]      The message model object that was persisted
 */
var createMessage = module.exports.createMessage = function(messageBoxId, createdBy, body, opts, callback) {
    opts = opts || {};

    var validator = new Validator();
    validator.check(messageBoxId, {'code': 400, 'msg': 'A messageBoxId must be specified.'}).notNull();
    validator.check(createdBy, {'code': 400, 'msg': 'The createdBy parameter must be a valid user id.'}).isUserId();
    validator.check(body, {'code': 400, 'msg': 'The body of the message must be specified.'}).notNull();
    if (opts.replyToCreated) {
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it should not be null.'}).notNull();
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it should be an integer.'}).isInt();
        validator.check(opts.replyToCreated, {'code': 400, 'msg': 'If the replyToCreated optional parameter is specified, it cannot be in the future.'}).max(Date.now());
    }
    if (validator.hasErrors()) {
        return callback(validator.getFirstError());
    }

    var replyToCreated = OaeUtil.getNumberParam(opts.replyToCreated);
    var replyToMessageId = (replyToCreated) ? _createMessageId(messageBoxId, replyToCreated) : null;

    // Fetch the threadKey of the parent so we can nest under it
    _getMessageThreadKey(replyToMessageId, function(err, replyToThreadKey) {
        if (err) {
            return callback(err);
        }

        // Generate an ID that can be used for locking and is as specific as possible.
        // Locking is required to make sure we don't end up with 2 messages that were
        // created at exactly the same time
        var id = (replyToThreadKey) ? replyToThreadKey : messageBoxId;
        _lockUniqueTimestamp(id, Date.now(), function(created, lockKey, lockToken) {

            // Data that will be output in diagnostic error messages
            var diagnosticData = {
                'messageBoxId': messageBoxId,
                'createdBy': createdBy,
                'created': created,
                'replyToMessageId': replyToMessageId
            };

            if (replyToMessageId && !replyToThreadKey) {
                // We specified a message that doesn't actually exist in our message box, don't let that happen
                log().error(diagnosticData, 'Reply-to message does not exist');
                return callback({'code': 400, 'msg': 'Reply-to message does not exist'});
            }

            // Derive this message's thread key by appending it to the parent, if applicable. Otherwise, it is a top-level key
            var threadKey = (replyToThreadKey) ? _appendToThreadKey(replyToThreadKey, created) : created + '|';

            // A low-level storage hash that represents this item stored in Cassandra or Redis
            var messageId = _createMessageId(messageBoxId, created);
            var messageStorageHash = {
                'createdBy': createdBy,
                'body': body,
                'threadKey': threadKey
            };

            // Create the query that creates the message object
            var createMessageQuery = Cassandra.constructUpsertCQL('Messages', 'id', messageId, messageStorageHash, 'QUORUM');
            if (!createMessageQuery) {
                log().error(diagnosticData, 'Failed to create a new message query.');
                return callback({'code': 500, 'msg': 'Failed to create a new message'});
            }

            // Create the query that adds the message object to the messagebox
            var indexMessageQuery = {
                'query': 'UPDATE MessageBoxMessages SET ? = ? WHERE messageBoxId = ?',
                'parameters': [threadKey, '1', messageBoxId]
            };

            // Create the query that updates the "recent contributors" to a message box
            var recentContributionsQuery = {
                'query': 'UPDATE MessageBoxRecentContributions USING TTL ' + DURATION_RECENT_CONTRIBUTIONS_SECONDS + ' SET ? = \'1\' WHERE messageBoxId = ?',
                'parameters': [createdBy, messageBoxId]
            };

            // First insert the new message object, if this fails we do not want to update the messagebox index
            Cassandra.runQuery(createMessageQuery.query, createMessageQuery.parameters, function(err) {
                if (err) {
                    return callback(err);
                }

                // Update the messagebox index, so this message will turn up in queries for all messages in the messagebox
                Cassandra.runQuery(indexMessageQuery.query, indexMessageQuery.parameters, function(err) {
                    if (err) {
                        return callback(err);
                    }

                    // Asynchronously update the recent contributions
                    Cassandra.runQuery(recentContributionsQuery.query, recentContributionsQuery.parameters);

                    // Get the expanded Message object, emit it so it can be indexed, and return it to the caller.
                    var message = _storageHashToMessage(messageId, messageStorageHash);
                    MessageBoxAPI.emit(MessageBoxConstants.events.CREATED_MESSAGE, message);
                    return callback(null, message);
                });
            });
            Locking.release(lockKey, lockToken, function(){});
        });
    });
};

/**
 * Lock a timestamp within a message thread key to prevent 2 messages within that message thread
 * key to be created at exactly the same time
 *
 * @param  {String}        id                  Identifier that provides context for the lock
 * @param  {Number}        timestamp           Timestamp on which to lock (ms since unix epoch)
 * @param  {Function}      callback            Standard callback function
 * @param  {Object}        callback.timestamp  The timestamp on which the lock was eventually acquired
 * @param  {Object}        callback.key        The key used to acquire the lock
 * @param  {String}        callback.lockToken  The lockToken that can be used to release the acquired lock
 * @api private
 */
var _lockUniqueTimestamp = function(id, timestamp, callback) {
    var key = 'oae-messagebox:' + id + ':' + timestamp;
    Locking.acquire(key, 1, function(err, lockToken) {
        if (err) {
            // This should only occur if Redis is down, just return the requested ts
            return callback(timestamp, lockToken);
        } else if (!lockToken) {
            // Someone else has the requested ts, try to lock one ms later
            return _lockUniqueTimestamp(id, timestamp + 1, callback);
        }
        // Successful lock, return the details
        return callback(timestamp, key, lockToken);
    });
};

/**
 * Update the content body of a message.
 *
 * @param  {String}         messageBoxId    The id message whose body we will update
 * @param  {String|Number}  created         The created timestamp (in millis since epoch) that we wish to update
 * @param  {String}         newBody         The new message content to push to the body
 * @param  {Function}       callback        Invoked when the process completes
 * @param  {Object}         callback.err    An error that occurred, if any
 */
var updateMessageBody = module.exports.updateMessageBody = function(messageBoxId, created, newBody, callback) {
    var validator = new Validator();
    validator.check(messageBoxId, {'code': 400, 'msg': 'A messageBoxId must be specified.'}).notNull();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be specified.'}).notNull();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be a valid timestamp (integer).'}).isInt();
    validator.check(created, {'code': 400, 'msg': 'The created parameter must be a valid timestamp (integer) that is not in the future.'}).max(Date.now());
    validator.check(newBody, {'code': 400, 'msg': 'The new body of the message must be specified.'}).notNull();
    if (validator.hasErrors()) {
{'fraction_non_alphanumeric': 0.09641860677857568, 'fraction_numerical': 0.005671682751680922, 'mean_word_length': 2.722799250808786, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1772342', 'n_tokens_mistral': 5732, 'n_tokens_neox': 5479, 'n_words': 2153}