feat: track user cids (#14114)

* feat: start tracking which cids a user has posted to, update account pages' topics/posts view to call this new sorted set

re: #14113

* feat: upgrade script for #14113

* fix: cids unavailable in getPostsFields, duh

* fix: update sortedSetIncrByBulk in mongo/psql to return early on empty data

* fix: remove unused lodash require

* test: sortedSetIncrBy and sortedSetIncrByBulk tests

* test: who needs null checks anyway

* fix: sortedSetIncrByBulk null response

* test: aggregate zincrbulk data

if there are alot of identical key/value pairs they will be combined into a single row

* fix: key name

* test: fix test name

* lint: fix lint issues

* test: negative values should work too

* fix: add e11000 handler for incrByBulk

* refactor: fix variable name

* merge tests with existing zset test, remove dupes

* test: return topicData for failing test

* delete uid:<uid>:cids on user delete

---------

Co-authored-by: Barış Soner Uşaklı <barisusakli@gmail.com>

src/database/helpers.js

@@ -42,3 +42,21 @@ helpers.globToRegex = function (match) {
 	}
 	return _match;
 };
+
+helpers.aggregateIncrByBulk = function (data) {
+	const buckets = Object.create(null);
+
+	for (const [key, incr, val] of data) {
+		buckets[key] = buckets[key] || {};
+		buckets[key][val] = (buckets[key][val] || 0) + incr;
+	}
+
+	const result = [];
+	for (const [key, vals] of Object.entries(buckets)) {
+		for (const [val, incr] of Object.entries(vals)) {
+			result.push([key, incr, val]);
+		}
+	}
+
+	return result;
+};

src/database/mongo/sorted.js

@@ -459,16 +459,35 @@ module.exports = function (module) {
 	};
 
 	module.sortedSetIncrByBulk = async function (data) {
+		if (!Array.isArray(data) || !data.length) {
+			return [];
+		}
+		const aggregated = dbHelpers.aggregateIncrByBulk(data);
 		const bulk = module.client.collection('objects').initializeUnorderedBulkOp();
-		data.forEach((item) => {
+		aggregated.forEach((item) => {
 			bulk.find({ _key: item[0], value: helpers.valueToString(item[2]) })
 				.upsert()
 				.update({ $inc: { score: parseFloat(item[1]) } });
 		});
-		await bulk.execute();
+
+		try {
+			await bulk.execute();
+		} catch (err) {
+			// retry failed e11000 operations
+			if (err.code === 11000 || (err.writeErrors && err.writeErrors.some(e => e.code === 11000))) {
+				const failedIndices = err.writeErrors.map(e => e.index);
+				const retryData = failedIndices.map(idx => aggregated[idx]);
+				await Promise.all(retryData.map(
+					item => module.sortedSetIncrBy(item[0], item[1], item[2])
+				));
+			} else {
+				throw err;
+			}
+		}
+
 		const result = await module.client.collection('objects').find({
-			_key: { $in: _.uniq(data.map(i => i[0])) },
-			value: { $in: _.uniq(data.map(i => i[2])) },
+			_key: { $in: _.uniq(aggregated.map(i => i[0])) },
+			value: { $in: _.uniq(aggregated.map(i => i[2])) },
 		}, {
 			projection: { _id: 0, _key: 1, value: 1, score: 1 },
 		}).toArray();

src/database/postgres/sorted.js

@@ -1,8 +1,10 @@
 'use strict';
 
 module.exports = function (module) {
-	const helpers = require('./helpers');
 	const util = require('util');
+
+	const helpers = require('./helpers');
+	const dbHelpers = require('../helpers');
 	const Cursor = require('pg-cursor');
 	Cursor.prototype.readAsync = util.promisify(Cursor.prototype.read);
 	const sleep = util.promisify(setTimeout);
@@ -547,18 +549,19 @@ RETURNING "score" s`,
 	};
 
 	module.sortedSetIncrByBulk = async function (data) {
-		if (!data.length) {
+		if (!Array.isArray(data) || !data.length) {
 			return [];
 		}
 
+		const aggregated = dbHelpers.aggregateIncrByBulk(data);
 		return await module.transaction(async (client) => {
-			await helpers.ensureLegacyObjectsType(client, data.map(item => item[0]), 'zset');
+			await helpers.ensureLegacyObjectsType(client, aggregated.map(item => item[0]), 'zset');
 
 			const values = [];
 			const queryParams = [];
 			let paramIndex = 1;
 
-			data.forEach(([key, increment, value]) => {
+			aggregated.forEach(([key, increment, value]) => {
 				value = helpers.valueToString(value);
 				increment = parseFloat(increment);
 				values.push(key, value, increment);

src/database/redis/sorted.js

@@ -261,8 +261,12 @@ module.exports = function (module) {
 	};
 
 	module.sortedSetIncrByBulk = async function (data) {
+		if (!Array.isArray(data) || !data.length) {
+			return [];
+		}
+		const aggregated = dbHelpers.aggregateIncrByBulk(data);
 		const multi = module.client.multi();
-		data.forEach((item) => {
+		aggregated.forEach((item) => {
 			multi.zIncrBy(item[0], item[1], String(item[2]));
 		});
 		const result = await multi.exec();