From 6022fd984a4169a2c259facc4645d2af9da683ee Mon Sep 17 00:00:00 2001
From: barisusakli <barisusakli@gmail.com>
Date: Mon, 15 Aug 2016 19:23:10 +0300
Subject: [PATCH] closes #4945

---
 src/database/mongo/sorted.js |  80 ++++++++++++++++++++++++++
 src/database/redis/sorted.js |  63 +++++++++++++++++++--
 tests/database/sorted.js     | 105 +++++++++++++++++++++++++++++++++++
 3 files changed, 243 insertions(+), 5 deletions(-)

diff --git a/src/database/mongo/sorted.js b/src/database/mongo/sorted.js
index 9e9f72969e..a262ff9984 100644
--- a/src/database/mongo/sorted.js
+++ b/src/database/mongo/sorted.js
@@ -595,4 +595,84 @@ module.exports = function(db, module) {
 			callback
 		);
 	};
+
+
+	module.getSortedSetIntersect = function(params, callback) {
+		params.sort = 1;
+		getSortedSetRevIntersect(params, callback);
+	};
+
+	module.getSortedSetRevIntersect = function(params, callback) {
+		params.sort = -1;
+		getSortedSetRevIntersect(params, callback);
+	};
+
+	function getSortedSetRevIntersect (params, callback) {
+		var sets = params.sets;
+		var start = params.hasOwnProperty('start') ? params.start : 0;
+		var stop = params.hasOwnProperty('stop') ? params.stop : -1;
+		var weights = params.weights || [];
+		var aggregate = {};
+
+		if (params.aggregate) {
+			aggregate['$' + params.aggregate.toLowerCase()] = '$score';
+		} else {
+			aggregate.$sum = '$score';
+		}
+
+		var limit = stop - start + 1;
+		if (limit <= 0) {
+			limit = 0;
+		}
+
+		var pipeline = [];
+
+		pipeline.push({ $match: { _key: {$in: sets}} });
+
+		weights.forEach(function(weight, index) {
+			if (weight !== 1) {
+				pipeline.push({
+					$project: {
+						value: 1,
+						score: {
+							$cond: { if: { $eq: [ "$_key", sets[index] ] }, then: { $multiply: [ '$score', weight ] }, else: '$score' }
+						}
+					}
+				});
+			}
+		});
+
+		pipeline.push({ $group: { _id: {value: '$value'}, totalScore: aggregate, count: {$sum: 1}} });
+		pipeline.push({ $match: { count: sets.length} });
+		pipeline.push({ $sort: { totalScore: params.sort} });
+
+		if (start) {
+			pipeline.push({ $skip: start });
+		}
+
+		if (limit > 0) {
+			pipeline.push({ $limit: limit });
+		}
+
+		var project = { _id: 0, value: '$_id.value'};
+		if (params.withScores) {
+			project.score = '$totalScore';
+		}
+		pipeline.push({ $project: project });
+
+		db.collection('objects').aggregate(pipeline, function(err, data) {
+			if (err || !data) {
+				return callback(err);
+			}
+
+			if (!params.withScores) {
+				data = data.map(function(item) {
+					return item.value;
+				});
+			}
+
+			callback(null, data);
+		});
+	}
+
 };
diff --git a/src/database/redis/sorted.js b/src/database/redis/sorted.js
index cd1587f494..d97b6b60cc 100644
--- a/src/database/redis/sorted.js
+++ b/src/database/redis/sorted.js
@@ -29,7 +29,7 @@ module.exports = function(redisClient, module) {
 			args.push(scores[i], values[i]);
 		}
 
-		redisClient.zadd(args, function(err, res) {
+		redisClient.zadd(args, function(err) {
 			callback(err);
 		});
 	}
@@ -42,7 +42,7 @@ module.exports = function(redisClient, module) {
 			multi.zadd(keys[i], score, value);
 		}
 
-		multi.exec(function(err, res) {
+		multi.exec(function(err) {
 			callback(err);
 		});
 	};
@@ -53,13 +53,13 @@ module.exports = function(redisClient, module) {
 			value = [value];
 		}
 
-		helpers.multiKeyValues(redisClient, 'zrem', key, value, function(err, result) {
+		helpers.multiKeyValues(redisClient, 'zrem', key, value, function(err) {
 			callback(err);
 		});
 	};
 
 	module.sortedSetsRemove = function(keys, value, callback) {
-		helpers.multiKeysValue(redisClient, 'zrem', keys, value, function(err, result) {
+		helpers.multiKeysValue(redisClient, 'zrem', keys, value, function(err) {
 			callback(err);
 		});
 	};
@@ -70,7 +70,7 @@ module.exports = function(redisClient, module) {
 		for(var i=0; i<keys.length; ++i) {
 			multi.zremrangebyscore(keys[i], min, max);
 		}
-		multi.exec(function(err, result) {
+		multi.exec(function(err) {
 			callback(err);
 		});
 	};
@@ -282,4 +282,57 @@ module.exports = function(redisClient, module) {
 		}
 		redisClient.zrangebylex([key, min, max, 'LIMIT', start, count], callback);
 	};
+
+	module.getSortedSetIntersect = function(params, callback) {
+		params.method = 'zrange';
+		getSortedSetRevIntersect(params, callback);
+	};
+
+	module.getSortedSetRevIntersect = function(params, callback) {
+		params.method = 'zrevrange';
+		getSortedSetRevIntersect(params, callback);
+	};
+
+	function getSortedSetRevIntersect (params, callback) {
+		var sets = params.sets;
+		var start = params.hasOwnProperty('start') ? params.start : 0;
+		var stop = params.hasOwnProperty('stop') ? params.stop : -1;
+		var weights = params.weights || [];
+
+		var tempSetName = 'temp_' + Date.now();
+
+		var interParams = [tempSetName, sets.length].concat(sets);
+		if (weights.length) {
+			interParams = interParams.concat(['WEIGHTS'].concat(weights));
+		}
+
+		if (params.aggregate) {
+			interParams = interParams.concat(['AGGREGATE', params.aggregate]);
+		}
+
+		var rangeParams = [tempSetName, start, stop];
+		if (params.withScores) {
+			rangeParams.push('WITHSCORES');
+		}
+
+		var	multi = redisClient.multi();
+		multi.zinterstore(interParams);
+		multi[params.method](rangeParams);
+		multi.del(tempSetName);
+		multi.exec(function(err, results) {
+			if (err) {
+				return callback(err);
+			}
+
+			if (!params.withScores) {
+				return callback(null, results ? results[1] : null);
+			}
+			results = results[1] || [];
+			var objects = [];
+			for(var i=0; i<results.length; i+=2) {
+				objects.push({value: results[i], score: parseFloat(results[i + 1])});
+			}
+			callback(null, objects);
+		});
+	}
 };
\ No newline at end of file
diff --git a/tests/database/sorted.js b/tests/database/sorted.js
index 64e8891a30..8f23038dae 100644
--- a/tests/database/sorted.js
+++ b/tests/database/sorted.js
@@ -532,6 +532,111 @@ describe('Sorted Set methods', function() {
 		});
 	});
 
+	describe('getSortedSetIntersect', function() {
+		before(function(done) {
+			async.parallel([
+				function(next) {
+					db.sortedSetAdd('interSet1', [1,2,3], ['value1', 'value2', 'value3'], next);
+				},
+				function(next) {
+					db.sortedSetAdd('interSet2', [4,5,6], ['value2', 'value3', 'value5'], next);
+				}
+			], done);
+		});
+
+		it('should return the intersection of two sets', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet2'],
+				start: 0,
+				stop: -1
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.deepEqual(['value2', 'value3'], data);
+				done();
+			});
+		});
+
+		it('should return the intersection of two sets with scores', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet2'],
+				start: 0,
+				stop: -1,
+				withScores: true
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.deepEqual([{value: 'value2', score: 6}, {value: 'value3', score: 8}], data);
+				done();
+			});
+		});
+
+		it('should return the intersection of two sets with scores aggregate MIN', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet2'],
+				start: 0,
+				stop: -1,
+				withScores: true,
+				aggregate: 'MIN'
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.deepEqual([{value: 'value2', score: 2}, {value: 'value3', score: 3}], data);
+				done();
+			});
+		});
+
+		it('should return the intersection of two sets with scores aggregate MAX', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet2'],
+				start: 0,
+				stop: -1,
+				withScores: true,
+				aggregate: 'MAX'
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.deepEqual([{value: 'value2', score: 4}, {value: 'value3', score: 5}], data);
+				done();
+			});
+		});
+
+		it('should return the intersection with scores modified by weights', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet2'],
+				start: 0,
+				stop: -1,
+				withScores: true,
+				weights: [1, 0.5]
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.deepEqual([{value: 'value2', score: 4}, {value: 'value3', score: 5.5}], data);
+				done();
+			});
+		});
+
+		it('should return empty array if sets do not exist', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet10', 'interSet12'],
+				start: 0,
+				stop: -1
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.equal(data.length, 0);
+				done();
+			});
+		});
+
+		it('should return empty array if one set does not exist', function(done) {
+			db.getSortedSetIntersect({
+				sets: ['interSet1', 'interSet12'],
+				start: 0,
+				stop: -1
+			}, function(err, data) {
+				assert.ifError(err);
+				assert.equal(data.length, 0);
+				done();
+			});
+		});
+
+	});
+
 
 	after(function(done) {
 		db.flushdb(done);