chore: release v0.0.13

apps/api-master/package.json

@@ -1,7 +1,7 @@
 {
 	"name": "@premid/api-master",
 	"type": "module",
-	"version": "0.0.4",
+	"version": "0.0.24",
 	"private": true,
 	"description": "PreMiD's api master",
 	"license": "MPL-2.0",
@@ -14,12 +14,16 @@
 		"dev": "node --watch --env-file .env --enable-source-maps ."
 	},
 	"dependencies": {
-		"@discordjs/rest": "^2.3.0",
 		"@envelop/sentry": "^9.0.0",
+		"@opentelemetry/api": "^1.9.0",
+		"@opentelemetry/exporter-prometheus": "^0.52.1",
+		"@opentelemetry/node": "^0.24.0",
 		"@sentry/node": "^8.17.0",
 		"cron": "^3.1.7",
 		"debug": "^4.3.6",
-		"ioredis": "^5.3.2"
+		"ioredis": "^5.3.2",
+		"ky": "^1.7.2",
+		"p-limit": "^6.1.0"
 	},
 	"devDependencies": {
 		"@types/debug": "^4.1.12"

apps/api-master/src/functions/clearOldSessions.ts

@@ -1,48 +1,105 @@
-import { REST } from "@discordjs/rest";
+import pLimit from "p-limit";
+import ky, { HTTPError, TimeoutError } from "ky";
 import { mainLog, redis } from "../index.js";
 
+let inProgress = false;
 export async function clearOldSessions() {
-	const sessionKeys = await redis.keys("pmd:session:*");
-	const now = Math.floor(Date.now() / 1000);
-
-	if (sessionKeys.length === 0) {
-		mainLog("No sessions to clear");
+	if (inProgress) {
+		mainLog("Session cleanup already in progress");
 		return;
 	}
 
-	mainLog(`Checking ${sessionKeys.length} sessions`);
-
+	inProgress = true;
+	const now = Date.now();
+	let cursor = "0";
+	let totalSessions = 0;
 	let cleared = 0;
-	for (const key of sessionKeys) {
-		const session = await redis.hgetall(key);
+	const batchSize = 100;
+	let keysToDelete: string[] = [];
 
-		if (!session.t || !session.u || !session.s) {
-			await redis.del(key);
-			cleared++;
-			continue;
+	mainLog("Starting session cleanup");
+
+	const limit = pLimit(100); // Create a limit of 100 concurrent operations
+
+	do {
+		const [nextCursor, result] = await redis.hscan("pmd-api.sessions", cursor, "COUNT", batchSize);
+		cursor = nextCursor;
+		totalSessions += result.length / 2;
+
+		const deletePromises = [];
+
+		for (let i = 0; i < result.length; i += 2) {
+			const key = result[i];
+			const value = result[i + 1];
+
+			if (!key || !value) {
+				continue;
+			}
+
+			const session = JSON.parse(value) as {
+				token: string;
+				session: string;
+				lastUpdated: number;
+			};
+
+			if (now - session.lastUpdated < 30000)
+				continue;
+
+			deletePromises.push(limit(() => deleteSession(session, key)));
 		}
 
-		//* If the session is younger than 30 seconds, skip it
-		if (now - Number(session.u) < 30)
-			continue;
+		const results = await Promise.allSettled(deletePromises);
+		results.forEach((result) => {
+			if (result.status === "fulfilled" && result.value) {
+				keysToDelete.push(result.value);
+				cleared++;
+			}
+		});
 
-		//* Delete the session
-		try {
-			const discord = new REST({ version: "10", authPrefix: "Bearer" });
-			discord.setToken(session.t);
-			await discord.post("/users/@me/headless-sessions/delete", {
-				body: {
-					token: session.s,
-				},
-			});
-		}
-		catch (error) {
-			mainLog(`Failed to delete session: %O`, error);
+		if (keysToDelete.length >= batchSize) {
+			await redis.hdel("pmd-api.sessions", ...keysToDelete);
+			keysToDelete = [];
 		}
+	} while (cursor !== "0");
 
-		cleared++;
-		await redis.del(key);
+	if (keysToDelete.length > 0) {
+		await redis.hdel("pmd-api.sessions", ...keysToDelete);
 	}
 
-	mainLog(`Cleared ${cleared} sessions`);
+	if (totalSessions === 0) {
+		mainLog("No sessions to clear");
+	}
+	else {
+		mainLog(`Checked ${totalSessions} sessions, cleared ${cleared}`);
+	}
+
+	inProgress = false;
+}
+
+async function deleteSession(session: { token: string; session: string }, key: string): Promise<string> {
+	try {
+		await ky.post("https://discord.com/api/v10/users/@me/headless-sessions/delete", {
+			json: {
+				token: session.session,
+			},
+			headers: {
+				Authorization: `Bearer ${session.token}`,
+			},
+			retry: 3,
+			timeout: 5000,
+		});
+	}
+	catch (error) {
+		if (error instanceof TimeoutError) {
+			mainLog(`Session deletion aborted due to timeout for key ${key}`);
+		}
+		else if (error instanceof HTTPError) {
+			mainLog(`Failed to delete session for key ${key}: [${error.name}] ${error.message} ${JSON.stringify(await error.response.json())}`);
+		}
+		else {
+			mainLog(`Failed to delete session for key ${key}: Unknown error`);
+		}
+	}
+
+	return key;
 }

apps/api-master/src/functions/setCounter.ts

@@ -0,0 +1,13 @@
+import { redis } from "../index.js";
+import { counter } from "../tracing.js";
+
+let activeActivities = 0;
+counter.add(0);
+export async function setCounter() {
+	const length = await redis.hlen("pmd-api.sessions");
+	if (length === activeActivities)
+		return;
+	const diff = length - activeActivities;
+	activeActivities = length;
+	counter.add(diff);
+}

apps/api-master/src/functions/updateActivePresenceGauge.ts

@@ -0,0 +1,39 @@
+import { redis } from "../index.js";
+import { activePresenceGauge } from "../tracing.js";
+
+//* Track previously recorded services
+const previousServices = new Set<string>();
+
+//* Function to update the gauge with per-service counts
+export async function updateActivePresenceGauge() {
+	const pattern = "pmd-api.heartbeatUpdates.*";
+	let cursor: string = "0";
+	const serviceCounts = new Map<string, number>();
+
+	do {
+		const [newCursor, keys] = await redis.scan(cursor, "MATCH", pattern, "COUNT", 1000); //* Use SCAN with COUNT for memory efficiency
+		cursor = newCursor;
+		for (const key of keys) {
+			const hash = await redis.hgetall(key);
+			const service = hash.service;
+			const version = hash.version; //* Get version from hash
+			serviceCounts.set(`${service}:${version}`, (serviceCounts.get(`${service}:${version}`) || 0) + 1);
+		}
+	} while (cursor !== "0");
+
+	// Set current counts and remove from previousServices
+	serviceCounts.forEach((count, serviceVersion) => {
+		const [presence_name, version] = serviceVersion.split(":");
+		activePresenceGauge.record(count, { presence_name, version });
+		previousServices.delete(serviceVersion);
+	});
+
+	// Set gauge to 0 for services that are no longer active
+	previousServices.forEach((serviceVersion) => {
+		const [presence_name, version] = serviceVersion.split(":");
+		activePresenceGauge.record(0, { presence_name, version });
+	});
+
+	// Update the set of previous services
+	serviceCounts.forEach((_, serviceVersion) => previousServices.add(serviceVersion));
+}

apps/api-master/src/index.ts

@@ -3,19 +3,42 @@ import { CronJob } from "cron";
 import debug from "debug";
 import { clearOldSessions } from "./functions/clearOldSessions.js";
 import createRedis from "./functions/createRedis.js";
+import { setCounter } from "./functions/setCounter.js";
+import "./tracing.js";
+import { updateActivePresenceGauge } from "./functions/updateActivePresenceGauge.js"; //* Added import
 
 export const redis = createRedis();
 
 export const mainLog = debug("api-master");
 
-debug("Starting cron job to clear old sessions");
+debug("Starting cron jobs");
 
 void new CronJob(
 	// Every 5 seconds
 	"*/5 * * * * *",
-	async () => {
+	() => {
 		clearOldSessions();
 	},
 	undefined,
 	true,
 );
+
+void new CronJob(
+	// Every second
+	"* * * * * *",
+	() => {
+		setCounter();
+	},
+	undefined,
+	true,
+);
+
+void new CronJob(
+	// Every 5 seconds
+	"*/5 * * * * *",
+	() => {
+		updateActivePresenceGauge();
+	},
+	undefined,
+	true,
+);

apps/api-master/src/tracing.ts

@@ -0,0 +1,24 @@
+import { ValueType } from "@opentelemetry/api";
+import { PrometheusExporter } from "@opentelemetry/exporter-prometheus";
+import { MeterProvider } from "@opentelemetry/sdk-metrics";
+
+const prometheusExporter = new PrometheusExporter();
+
+const provider = new MeterProvider({
+	readers: [prometheusExporter],
+});
+
+const meter = provider.getMeter("nice");
+
+export const counter = meter.createUpDownCounter("active_activites", {
+	description: "Number of active activities",
+	valueType: ValueType.INT,
+});
+
+// * Replace Observable Gauge with regular Gauge
+export const activePresenceGauge = meter.createGauge("active_presence_names", {
+	description: "Number of active presence names per service",
+	valueType: ValueType.INT,
+});
+
+prometheusExporter.startServer();

apps/api-worker/package.json

@@ -1,7 +1,7 @@
 {
 	"name": "@premid/api-worker",
 	"type": "module",
-	"version": "0.0.7",
+	"version": "0.0.13",
 	"private": true,
 	"description": "PreMiD's api",
 	"license": "MPL-2.0",

apps/api-worker/src/classes/Socket.ts

@@ -5,7 +5,6 @@ import WebSocket from "ws";
 import type { FastifyRequest } from "fastify";
 import type { RawData } from "ws";
 import { redis } from "../functions/createServer.js";
-import { shortHash } from "../functions/shortHash.js";
 import { counter } from "../tracing.js";
 
 const schema = scope({
@@ -55,7 +54,7 @@ export class Socket {
 					break;
 				}
 				case "session": {
-					await redis.del(`pmd:session:${out.token}`);
+					await redis.hdel("pmd-api.sessions", out.token);
 					this.currentSession = out;
 					break;
 				}
@@ -73,18 +72,15 @@ export class Socket {
 		if (!this.currentToken || !this.currentSession)
 			return;
 
-		const now = Math.floor(Date.now() / 1000);
-
-		await redis.hmset(
-			`pmd:session:${shortHash(this.currentSession.token, this.currentToken.token)}`,
-			{
-				t: this.currentToken.token,
-				u: now,
-				s: this.currentSession.token,
-			},
+		await redis.hset(
+			"pmd-api.sessions",
+			this.currentSession.token,
+			JSON.stringify({
+				session: this.currentSession.token,
+				token: this.currentToken.token,
+				lastUpdated: Date.now(),
+			}),
 		);
-
-		await redis.expire(`pmd:session:${shortHash(this.currentSession.token, this.currentToken.token)}`, 60); // Expire after 1 minute
 	}
 
 	async isTokenValid(token: typeof schema.token.infer) {

apps/api-worker/src/constants.ts

@@ -0,0 +1,10 @@
+import process from "node:process";
+import { defu } from "defu";
+
+const disabledFlags = process.env.DISABLED_FEATURE_FLAGS?.split(",") ?? [];
+const flags = Object.fromEntries(disabledFlags.map(flag => [flag, false]));
+
+export const featureFlags = defu(flags, {
+	WebSocketManager: true,
+	SessionKeepAlive: true,
+});

apps/api-worker/src/functions/createServer.ts

@@ -1,13 +1,11 @@
 import { readFile } from "node:fs/promises";
 import { resolve } from "node:path";
-import process from "node:process";
 import { useSentry } from "@envelop/sentry";
 import { maxAliasesPlugin } from "@escape.tech/graphql-armor-max-aliases";
 import { maxDepthPlugin } from "@escape.tech/graphql-armor-max-depth";
 import { maxDirectivesPlugin } from "@escape.tech/graphql-armor-max-directives";
 import { maxTokensPlugin } from "@escape.tech/graphql-armor-max-tokens";
 import fastifyWebsocket from "@fastify/websocket";
-import { defu } from "defu";
 import fastify from "fastify";
 
 import { createSchema, createYoga } from "graphql-yoga";
@@ -15,6 +13,7 @@ import type { FastifyReply, FastifyRequest } from "fastify";
 import { Socket } from "../classes/Socket.js";
 import { resolvers } from "../graphql/resolvers/v5/index.js";
 import { sessionKeepAlive } from "../routes/sessionKeepAlive.js";
+import { featureFlags } from "../constants.js";
 import createRedis from "./createRedis.js";
 
 export interface FastifyContext {
@@ -48,7 +47,7 @@ export default async function createServer() {
 			maxDepthPlugin(),
 			maxDirectivesPlugin(),
 			maxTokensPlugin(),
-			useSentry(),
+			/* useSentry(), */
 		],
 		schema: createSchema<FastifyContext>({
 			resolvers,
@@ -87,15 +86,7 @@ export default async function createServer() {
 	});
 
 	app.get("/v5/feature-flags", async (request, reply) => {
-		const disabledFlags = process.env.DISABLED_FEATURE_FLAGS?.split(",") ?? [];
-		const flags = Object.fromEntries(disabledFlags.map(flag => [flag, false]));
-
-		const test = defu(flags, {
-			WebSocketManager: true,
-			SessionKeepAlive: true,
-		});
-
-		void reply.send(test);
+		void reply.send(featureFlags);
 	});
 
 	app.post("/v5/session-keep-alive", sessionKeepAlive);

apps/api-worker/src/functions/shortHash.ts

@@ -1,5 +0,0 @@
-import { createHash } from "node:crypto";
-
-export function shortHash(...input: string[]) {
-	return createHash("sha256").update(input.join("")).digest("hex").slice(0, 16);
-}

apps/api-worker/src/graphql/resolvers/v5/Mutation/heartbeat.ts

@@ -1,9 +1,10 @@
 import { type } from "arktype";
 import type { MutationResolvers } from "../../../../generated/graphql-v5.js";
+import { redis } from "../../../../functions/createServer.js";
 
 const heartbeatSchema = type({
 	identifier: "string.uuid & string.lower",
-	presences: {
+	presence: {
 		service: "string.trim",
 		version: "string.semver",
 		language: "string.trim",
@@ -25,13 +26,19 @@ const mutation: MutationResolvers["heartbeat"] = async (_parent, input) => {
 	if (out instanceof type.errors)
 		throw new Error(out.summary);
 
-	// ! Disabled for now
-	/* await redis.setex(
-		`pmd-api.heartbeatUpdates.${data.identifier}`,
-		// 5 minutes
-		300,
-		JSON.stringify(data)
-	); */
+	// * Use Redis Hash with 'service' in the key to store heartbeat data
+	const redisKey = `pmd-api.heartbeatUpdates.${out.identifier}`;
+	await redis.hset(redisKey, {
+		service: out.presence.service,
+		version: out.presence.version,
+		language: out.presence.language,
+		since: out.presence.since.toString(),
+		extension_version: out.extension.version,
+		extension_language: out.extension.language,
+		extension_connected_app: out.extension.connected?.app?.toString(),
+		extension_connected_discord: out.extension.connected?.discord?.toString(),
+	});
+	await redis.expire(redisKey, 300);
 
 	return {
 		__typename: "HeartbeatResult",

apps/api-worker/src/graphql/resolvers/v5/Mutation/index.ts

@@ -1,6 +1,6 @@
+import type { MutationResolvers } from "../../../../generated/graphql-v5.js";
 import addScience from "./addScience.js";
 import heartbeat from "./heartbeat.js";
-import type { MutationResolvers } from "../../../../generated/graphql-v5.js";
 
 export const Mutation: MutationResolvers = {
 	addScience,

apps/api-worker/src/graphql/resolvers/v5/Query/index.ts

@@ -1,5 +1,5 @@
-import presences from "./presences.js";
 import type { QueryResolvers } from "../../../../generated/graphql-v5.js";
+import presences from "./presences.js";
 
 export const Query: QueryResolvers = {
 	presences,

apps/api-worker/src/graphql/resolvers/v5/index.ts

@@ -1,6 +1,6 @@
+import type { Resolvers } from "../../../generated/graphql-v5.js";
 import { Mutation } from "./Mutation/index.js";
 import { Query } from "./Query/index.js";
-import type { Resolvers } from "../../../generated/graphql-v5.js";
 
 export const resolvers: Resolvers = {
 	Query,

apps/api-worker/src/index.ts

@@ -3,8 +3,6 @@ import process from "node:process";
 import * as Sentry from "@sentry/node";
 import { connect } from "mongoose";
 import "./tracing.js";
-
-// eslint-disable-next-line perfectionist/sort-imports
 import createServer from "./functions/createServer.js";
 
 // TODO SETUP SENTRY

apps/api-worker/src/routes/sessionKeepAlive.ts

@@ -4,18 +4,25 @@ import { type } from "arktype";
 import { Routes } from "discord-api-types/v10";
 import type { FastifyReply, FastifyRequest } from "fastify";
 import { redis } from "../functions/createServer.js";
-import { shortHash } from "../functions/shortHash.js";
+import { featureFlags } from "../constants.js";
 
 const schema = type({
 	token: "string.trim",
 	session: "string.trim",
+	version: "string.semver & string.trim",
+	scienceId: "string.trim",
 });
 
 export async function sessionKeepAlive(request: FastifyRequest, reply: FastifyReply) {
-	//* Get the 2 headers
+	if (!featureFlags.SessionKeepAlive)
+		return reply.status(202).send();
+
+	//* Get the headers
 	const out = schema({
 		token: request.headers["x-token"],
 		session: request.headers["x-session"],
+		version: request.headers["x-version"] ?? "2.6.8",
+		scienceId: request.headers["x-science-id"] ?? request.headers["x-token"],
 	});
 
 	if (out instanceof type.errors)
@@ -24,20 +31,16 @@ export async function sessionKeepAlive(request: FastifyRequest, reply: FastifyRe
 	if (!await isTokenValid(out.token))
 		return reply.status(400).send({ code: "INVALID_TOKEN", message: "The token is invalid" });
 
-	const now = Math.floor(Date.now() / 1000); // Unix timestamp in seconds
-
-	await redis.hmset(
-		`pmd:session:${shortHash(out.session, out.token)}`,
-		{
-			t: out.token,
-			u: now,
-			s: out.session,
-		},
+	await redis.hset(
+		"pmd-api.sessions",
+		out.scienceId,
+		JSON.stringify({
+			session: out.session,
+			token: out.token,
+			lastUpdated: Date.now(),
+		}),
 	);
 
-	// Set expiration for the hash
-	await redis.expire(`pmd:session:${shortHash(out.session, out.token)}`, 60); // Expire after 1 minute
-
 	const interval = Number.parseInt(process.env.SESSION_KEEP_ALIVE_INTERVAL ?? "5000");
 
 	return reply.status(200).send({

apps/pd/src/functions/createKeyv.ts

@@ -1,12 +1,11 @@
 import process from "node:process";
 import KeyvRedis from "@keyv/redis";
 import Keyv from "keyv";
-import type { KeyvOptions } from "keyv";
 
 import redis from "../redis.js";
 
 export default function createKeyv() {
-	let options: KeyvOptions | undefined;
+	let options: Keyv.Options<string> | undefined;
 
 	/* c8 ignore next 8 */
 	if (process.env.REDIS_SENTINELS) {
@@ -16,7 +15,7 @@ export default function createKeyv() {
 		};
 	}
 
-	const keyv = new Keyv(
+	const keyv = new Keyv<string>(
 		options,
 	);
 

apps/pd/src/routes/createFromImage.test.ts

@@ -1,9 +1,9 @@
 import { Buffer } from "node:buffer";
 import { readFile } from "node:fs/promises";
-import { afterAll, beforeAll, describe, it } from "vitest";
 
 import type { RequestOptions } from "node:http";
 import type { AddressInfo } from "node:net";
+import { afterAll, beforeAll, describe, it } from "vitest";
 
 import { createServer } from "../functions/createServer.js";
 

apps/pd/src/routes/createShortenedLink.ts

@@ -20,7 +20,7 @@ const handler: RouteHandlerMethod = async (request, reply) => {
 		return reply.status(400).send("Invalid URL");
 
 	const hash = crypto.createHash("sha256").update(url).digest("hex");
-	const existingShortenedUrl = await keyv.get<string>(hash);
+	const existingShortenedUrl = await keyv.get(hash);
 
 	void reply.header("Cache-control", "public, max-age=1800");
 

apps/pd/src/routes/getFullLink.ts

@@ -25,7 +25,7 @@ const handler: RouteHandlerMethod = async (request, reply) => {
 	if (id.split(".")[0]?.length !== 10)
 		return reply.code(404).send("Invalid ID");
 
-	const url = await keyv.get<string>(id);
+	const url = await keyv.get(id);
 	if (!url)
 		return reply.code(404).send("Unknown ID");
 

apps/website/server/api/getStaffData.ts

@@ -1,4 +1,4 @@
-import { ActivityType, flagsToBadges, PresenceUpdateStatus } from "@discord-user-card/vue";
+import { ActivityType, PresenceUpdateStatus, flagsToBadges } from "@discord-user-card/vue";
 import { REST } from "@discordjs/rest";
 import { Routes } from "discord-api-types/v10";
 import type { DiscordUserCardActivity, DiscordUserCardUser } from "@discord-user-card/vue";