perf(stats): threads, better & parallel queries

src/lib/stats.js

@@ -1,27 +1,22 @@
 /* eslint-disable no-underscore-dangle */
 
-const {
-	spawn,
-	Pool,
-	Worker,
-} = require('threads');
-const { cpus } = require('node:os');
 const { version } = require('../../package.json');
 const { md5 } = require('./misc');
+const {
+	quick,
+	relativePool,
+} = require('./threads');
 
-// ! ceiL: at least 1
-const poolSize = Math.ceil(cpus().length / 4);
-const pool = Pool(() => spawn(new Worker('./workers/stats.js')), { size: poolSize });
-
-module.exports.getAvgResolutionTime = tickets => (tickets.reduce((total, ticket) => total + (ticket.closedAt - ticket.createdAt), 0) || 1) / Math.max(tickets.length, 1);
-
-module.exports.getAvgResponseTime = tickets => (tickets.reduce((total, ticket) => total + (ticket.firstResponseAt - ticket.createdAt), 0) || 1) / Math.max(tickets.length, 1);
+const getAverageTimes = closedTickets => quick('stats', async w => ({
+	avgResolutionTime: await w.getAvgResolutionTime(closedTickets),
+	avgResponseTime: await w.getAvgResponseTime(closedTickets),
+}));
 
 /**
- *
+ * Report stats to Houston
  * @param {import("../client")} client
  */
-module.exports.sendToHouston = async client => {
+async function sendToHouston(client) {
 	const guilds = await client.prisma.guild.findMany({
 		include: {
 			categories: { include: { _count: { select: { questions: true } } } },
@@ -35,30 +30,32 @@ module.exports.sendToHouston = async client => {
 			},
 		},
 	});
-	const users = (await client.prisma.user.aggregate({
+	const users = await client.prisma.user.aggregate({
 		_count: true,
 		_sum: { messageCount: true },
-	}));
-	const messages = users._sum.messageCount ?? 0;
+	});
+	const messages = users._sum.messageCount;
 	const stats = {
 		activated_users: users._count,
 		arch: process.arch,
 		database: process.env.DB_PROVIDER,
 		guilds: await Promise.all(
-			guilds
-				.filter(guild => client.guilds.cache.has(guild.id))
-				.map(async guild => {
-					guild.members = client.guilds.cache.get(guild.id).memberCount;
-					return pool.queue(worker => worker.aggregateGuildForHouston(guild, messages));
-				}),
+			await relativePool(0.25, 'stats', pool =>
+				guilds
+					.filter(guild => client.guilds.cache.has(guild.id))
+					.map(async guild => {
+						guild.members = client.guilds.cache.get(guild.id).memberCount;
+						return pool.queue(w => w.aggregateGuildForHouston(guild, messages));
+					}),
+			),
 		),
 		id: md5(client.user.id),
 		node: process.version,
 		os: process.platform,
 		version,
 	};
-
 	const delta = guilds.length - stats.guilds.length;
+
 	if (delta !== 0) {
 		client.log.warn('%d guilds are not cached and were excluded from the stats report', delta);
 	}
@@ -84,3 +81,8 @@ module.exports.sendToHouston = async client => {
 		client.log.debug(res);
 	}
 };
+
+module.exports = {
+	getAverageTimes,
+	sendToHouston,
+};

src/lib/threads.js

@@ -0,0 +1,59 @@
+const {
+	spawn,
+	Pool,
+	Thread,
+	Worker,
+} = require('threads');
+const { cpus } = require('node:os');
+
+/**
+ * Use a thread pool of a fixed size
+ * @param {number} size number of threads
+ * @param {string} name name of file in workers directory
+ * @param {function} fun async function
+ * @returns {Promise<any>}
+ */
+async function pool(size, name, fun) {
+	const pool = Pool(() => spawn(new Worker(`./workers/${name}.js`)), { size });
+	try {
+		return await fun(pool);
+	} finally {
+		await pool.settled();
+		await pool.terminate();
+	}
+};
+
+/**
+ * Spawn one thread, do something, and terminate it
+ * @param {string} name name of file in workers directory
+ * @param {function} fun async function
+ * @returns {Promise<any}
+ */
+async function quick(name, fun) {
+	const thread = await spawn(new Worker(`./workers/${name}.js`));
+	try {
+		// ! this await is extremely important
+		return await fun(thread);
+	} finally {
+		await Thread.terminate(thread);
+	}
+};
+
+/**
+ * Use a thread pool of a variable size
+ * @param {number} size fraction of available CPU cores to use (ceil'd)
+ * @param {string} name name of file in workers directory
+ * @param {function} fun async function
+ * @returns {Promise<any>}
+ */
+function relativePool(fraction, ...args) {
+	// ! ceiL: at least 1
+	const poolSize = Math.ceil(fraction * cpus().length);
+	return pool(poolSize, ...args);
+}
+
+module.exports = {
+	pool,
+	quick,
+	relativePool,
+};

src/lib/tickets/manager.js

@@ -20,15 +20,12 @@ const { isStaff } = require('../users');
 const { Collection } = require('discord.js');
 const spacetime = require('spacetime');
 const Cryptr = require('cryptr');
-const {
-	getAvgResolutionTime,
-	getAvgResponseTime,
-} = require('../stats');
 const {
 	decrypt,
 	encrypt,
 } = new Cryptr(process.env.ENCRYPTION_KEY);
 const { getSUID } = require('../logging');
+const { getAverageTimes } = require('../stats');
 
 /**
  * @typedef {import('@prisma/client').Category &
@@ -434,9 +431,13 @@ module.exports = class TicketManager {
 					open: false,
 				},
 			});
+			const {
+				avgResolutionTime,
+				avgResponseTime,
+			} = await getAverageTimes(closedTickets);
 			stats = {
-				avgResolutionTime: ms(getAvgResolutionTime(closedTickets), { long: true }),
-				avgResponseTime: ms(getAvgResponseTime(closedTickets), { long: true }),
+				avgResolutionTime: ms(avgResolutionTime, { long: true }),
+				avgResponseTime: ms(avgResponseTime, { long: true }),
 			};
 			this.client.keyv.set(statsCacheKey, stats, ms('1h'));
 		}

src/lib/workers/stats.js

@@ -7,9 +7,13 @@ const md5 = str => createHash('md5').update(str).digest('hex');
 
 const msToMins = ms => Number((ms / 1000 / 60).toFixed(2));
 
-const getAvgResolutionTime = tickets => (tickets.reduce((total, ticket) => total + (ticket.closedAt - ticket.createdAt), 0) || 1) / Math.max(tickets.length, 1);
+const reduce = (closedTickets, prop) => closedTickets.reduce((total, ticket) => total + (ticket[prop] - ticket.createdAt), 0) || 1;
 
-const getAvgResponseTime = tickets => (tickets.reduce((total, ticket) => total + (ticket.firstResponseAt - ticket.createdAt), 0) || 1) / Math.max(tickets.length, 1);
+const getAvgResolutionTime = closedTickets => reduce(closedTickets, 'closedAt') / Math.max(closedTickets.length, 1);
+
+const getAvgResponseTime = closedTickets => reduce(closedTickets, 'firstResponseAt') / Math.max(closedTickets.length, 1);
+
+const sum = numbers => numbers.reduce((t, n) => t + n, 0);
 
 expose({
 	aggregateGuildForHouston(guild, messages) {
@@ -37,4 +41,5 @@ expose({
 	},
 	getAvgResolutionTime,
 	getAvgResponseTime,
+	sum,
 });