chore: Update heartbeat cron job to run every second for more accurate updates

Author: Timeraa

apps/master/src/index.ts

@@ -53,4 +53,4 @@ await Promise.all([
 
 new CronJob("* * * * *", updateScience).start();
 new CronJob("* * * * *", calculatePresenceUsage).start();
-new CronJob("* * * * *", updateHeartbeats).start();
+new CronJob("*/1 * * * * *", updateHeartbeats).start();

apps/master/src/util/updateHeartbeats.ts

@@ -1,64 +1,67 @@
 import pLimit from "p-limit";
 import { mainLog, mongo, redis } from "../index.js";
 
-const limit = pLimit(Number.parseInt(process.env.LIMIT || "5"));
+const BATCH_LIMIT = pLimit(Number.parseInt(process.env.BATCH_LIMIT || "5"));
 const BATCH_SIZE = Number.parseInt(process.env.BATCH_SIZE || "1000");
 
+const limit = pLimit(1);
 export default async function () {
-	const log = mainLog.extend("updateHeartbeats");
-	let count = 0;
+	return limit(async () => {
+		const log = mainLog.extend("updateHeartbeats");
+		let count = 0;
 
-	const processBatch = async (cursor: string) => {
-		const result = await redis.hscan(
-			"pmd-api.heartbeatUpdates",
-			cursor,
-			"COUNT",
-			BATCH_SIZE
-		);
-		const newCursor = result[0];
+		const processBatch = async (cursor: string) => {
+			const result = await redis.hscan(
+				"pmd-api.heartbeatUpdates",
+				cursor,
+				"COUNT",
+				BATCH_SIZE
+			);
+			const newCursor = result[0];
 
-		let batchToDelete = [];
-		let data = [];
-		for (let i = 0; i < result[1].length; i += 2) {
-			const identifier = result[1][i];
-			data.push(JSON.parse(result[1][i + 1]));
+			let batchToDelete = [];
+			let data = [];
+			for (let i = 0; i < result[1].length; i += 2) {
+				const identifier = result[1][i];
+				data.push(JSON.parse(result[1][i + 1]));
 
-			batchToDelete.push(identifier);
-			count++;
-		}
+				batchToDelete.push(identifier);
+				count++;
+			}
 
-		if (batchToDelete.length) {
-			await redis.hdel("pmd-api.heartbeatUpdates", ...batchToDelete);
+			if (batchToDelete.length) {
+				await redis.hdel("pmd-api.heartbeatUpdates", ...batchToDelete);
 
-			const res = await mongo
-				.db("PreMiD")
-				.collection("heartbeats")
-				.bulkWrite(
-					data.map(d => ({
-						updateOne: {
-							filter: { identifier: d.identifier },
-							update: {
-								$set: { ...d, updated: new Date(d.updated) }
-							},
-							upsert: true
-						}
-					}))
+				const res = await mongo
+					.db("PreMiD")
+					.collection("heartbeats")
+					.bulkWrite(
+						data.map(d => ({
+							updateOne: {
+								filter: { identifier: d.identifier },
+								update: {
+									$set: { ...d, updated: new Date(d.updated) }
+								},
+								upsert: true
+							}
+						}))
+					);
+				log(
+					"Batch %s: Inserted %s entries, Updated %s entries",
+					Math.floor(count / BATCH_SIZE) + 1,
+					res.upsertedCount,
+					res.modifiedCount
 				);
-			log(
-				"Batch %s: Inserted %s entries, Updated %s entries",
-				Math.floor(count / BATCH_SIZE) + 1,
-				res.upsertedCount,
-				res.modifiedCount
-			);
-		}
+			}
 
-		return newCursor;
-	};
+			return newCursor;
+		};
 
-	let cursor = "0";
-	do {
-		cursor = await limit(() => processBatch(cursor));
-	} while (cursor !== "0");
+		let cursor = "0";
+		do {
+			cursor = await BATCH_LIMIT(() => processBatch(cursor));
+		} while (cursor !== "0");
 
-	if (count > 0) log("Updated %s entries", count);
+		if (count > 0) log("Updated %s entries", count);
+	});
 }