feat: heartbeat

Author: Timeraa

apps/master/src/util/updateHeartbeats.ts

@@ -1,35 +1,33 @@
 import pLimit from "p-limit";
 import { mainLog, mongo, redis } from "../index.js";
 
-const limit = pLimit(1);
+const limit = pLimit(Number.parseInt(process.env.LIMIT || "5"));
+const BATCH_SIZE = Number.parseInt(process.env.BATCH_SIZE || "1000");
 
-const BATCH_SIZE = Number.parseInt(process.env.BATCH_SIZE || "10000");
 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",
-				BATCH_SIZE
-			);
-			cursor = result[0];
+	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];
 
-			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) continue;
+		if (batchToDelete.length) {
 			await redis.hdel("pmd-api.heartbeatUpdates", ...batchToDelete);
 
 			const res = await mongo
@@ -40,7 +38,7 @@ export default async function () {
 						updateOne: {
 							filter: { identifier: d.identifier },
 							update: {
-								$set: { ...d, updated: new Date() }
+								$set: { ...d, updated: new Date(d.updated) }
 							},
 							upsert: true
 						}
@@ -52,8 +50,15 @@ export default async function () {
 				res.upsertedCount,
 				res.modifiedCount
 			);
-		} while (cursor !== "0");
+		}
+
+		return newCursor;
+	};
+
+	let cursor = "0";
+	do {
+		cursor = await limit(() => processBatch(cursor));
+	} while (cursor !== "0");
 
-		if (count > 0) log("Updated %s entries", count);
-	});
+	if (count > 0) log("Updated %s entries", count);
 }

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

@@ -81,7 +81,8 @@ export async function resolver(
 	const data = {
 		identifier: params.identifier,
 		presence: params.presence,
-		extension: params.extension
+		extension: params.extension,
+		updated: new Date()
 	};
 
 	await redis.hset(