feat: heartbeat

apps/master/src/index.ts

@@ -10,6 +10,7 @@ import { Redis } from "ioredis";
 
 import calculatePresenceUsage from "./util/calculatePresenceUsage.js";
 import updateScience from "./util/updateScience.js";
+import updateHeartbeats from "./util/updateHeartbeats.js";
 
 if (process.env.NODE_ENV !== "production")
 	(await import("dotenv")).config({ path: "../../../.env" });
@@ -44,7 +45,12 @@ mainLog("Connecting to Redis...");
 await redis.connect();
 mainLog("Connected!");
 
-await Promise.all([updateScience(), calculatePresenceUsage()]);
+await Promise.all([
+	updateScience(),
+	calculatePresenceUsage(),
+	updateHeartbeats()
+]);
 
 new CronJob("* * * * *", updateScience).start();
 new CronJob("* * * * *", calculatePresenceUsage).start();
+new CronJob("* * * * *", updateHeartbeats).start();

apps/master/src/util/updateHeartbeats.ts

@@ -0,0 +1,57 @@
+import pLimit from "p-limit";
+import { mainLog, mongo, redis } from "../index.js";
+
+const limit = pLimit(1);
+export default async function () {
+	limit(async () => {
+		let count = 0;
+		const log = mainLog.extend("updateHeartbeats");
+
+		let cursor = "";
+		do {
+			const result = await redis.hscan(
+				"pmd-api.heartbeatUpdates",
+				cursor,
+				"COUNT",
+				1000
+			);
+			cursor = 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]));
+
+				batchToDelete.push(identifier);
+				count++;
+			}
+
+			if (!batchToDelete.length) continue;
+			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() }
+							},
+							upsert: true
+						}
+					}))
+				);
+			log(
+				"Batch %s: Inserted %s entries, Updated %s entries",
+				Math.floor(count / 1000) + 1,
+				res.upsertedCount,
+				res.modifiedCount
+			);
+		} while (cursor !== "0");
+
+		if (count > 0) log("Updated %s entries", count);
+	});
+}

apps/worker/src/v4/fields/heartbeat.ts

@@ -84,13 +84,11 @@ export async function resolver(
 		extension: params.extension
 	};
 
-	//! Disabled for now
-	/* await redis.setex(
-		`pmd-api.heartbeatUpdates.${data.identifier}`,
-		// 5 minutes
-		300,
+	await redis.hset(
+		"pmd-api.heartbeatUpdates",
+		data.identifier,
 		JSON.stringify(data)
-	); */
+	);
 
 	return data;
 }