chore: redis cluster

Co-authored-by: Bas van Zanten <me@bas950.com>

apps/master/package.json

@@ -12,15 +12,13 @@
 		"@sentry/tracing": "^7.46.0",
 		"cron": "^2.3.0",
 		"debug": "^4.3.4",
-		"ioredis": "^5.0.1",
 		"mongodb": "^5.1.0",
-		"p-event": "^6.0.0",
+		"redis": "^4.6.12",
 		"source-map-support": "^0.5.21"
 	},
 	"devDependencies": {
 		"@types/cron": "^2.0.1",
 		"@types/debug": "^4.1.7",
-		"@types/ioredis": "^4.28.10",
 		"dotenv": "^16.0.3",
 		"ts-devscript": "^3.0.7",
 		"typescript": "^5.0.3"

apps/master/src/index.ts

@@ -6,11 +6,10 @@ import * as Sentry from "@sentry/node";
 import { Integrations } from "@sentry/tracing";
 import debug from "debug";
 import { MongoClient } from "mongodb";
-import { Redis } from "ioredis";
+import { createCluster } from "redis";
 
 import calculatePresenceUsage from "./util/calculatePresenceUsage.js";
 import updateScience from "./util/updateScience.js";
-import { pEvent } from "p-event";
 
 if (process.env.NODE_ENV !== "production")
 	(await import("dotenv")).config({ path: "../../../.env" });
@@ -24,14 +23,12 @@ Sentry.init({
 	integrations: [new Integrations.Mongo()]
 });
 
-export const redis = new Redis({
-		sentinels: [
+export const redis = createCluster({
+		rootNodes: [
 			{
-				host: process.env.REDIS_HOST || "localhost",
-				port: parseInt(process.env.REDIS_PORT || "26379")
+				url: process.env.REDIS_URL || "redis://localhost:6379"
 			}
-		],
-		name: "mymaster"
+		]
 	}),
 	mongo = new MongoClient(process.env.MONGO_URL!, {
 		appName: "PreMiD-API-Master"
@@ -40,8 +37,10 @@ export const redis = new Redis({
 
 debug.enable("API-Master*");
 
-mainLog("Connecting to MongoDB and Redis...");
-await Promise.all([mongo.connect(), pEvent(redis, "connect")]);
+mainLog("Connecting to MongoDB...");
+await mongo.connect();
+mainLog("Connecting to Redis...");
+await redis.connect();
 mainLog("Connected!");
 
 await Promise.all([updateScience(), calculatePresenceUsage()]);

apps/master/src/util/updateScience.ts

@@ -5,7 +5,7 @@ export default async function () {
 		key = "pmd-api.scienceUpdates";
 
 	log("Updating...");
-	const scienceUpdates = await redis.hvals(key);
+	const scienceUpdates = await redis.hVals(key);
 
 	let invalidEntries: string[] = [],
 		entries: {
@@ -35,7 +35,7 @@ export default async function () {
 	}
 
 	if (invalidEntries.length) {
-		await redis.hdel(key, ...invalidEntries);
+		await redis.hDel(key, invalidEntries);
 		log("Deleted %n invalid entries", invalidEntries.length);
 	}
 
@@ -53,7 +53,10 @@ export default async function () {
 				}))
 			);
 
-		await redis.hdel(key, ...entries.map(e => e.identifier));
+		await redis.hDel(
+			key,
+			entries.map(e => e.identifier)
+		);
 
 		log(
 			"Inserted %s entries, Updated %s entries",
@@ -62,7 +65,7 @@ export default async function () {
 		);
 	} else log("No entries to update");
 
-	const delRedis = await redis.hvals("pmd-api.scienceDeletes");
+	const delRedis = await redis.hVals("pmd-api.scienceDeletes");
 
 	let orMatch: any[] = [
 		{
@@ -76,7 +79,7 @@ export default async function () {
 		$or: orMatch
 	});
 
-	if (delRedis.length) await redis.hdel("pmd-api.scienceDeletes", ...delRedis);
+	if (delRedis.length) await redis.hDel("pmd-api.scienceDeletes", delRedis);
 
 	if (delRes.deletedCount) log("Deleted %s entries", delRes.deletedCount);
 }

apps/worker/package.json

@@ -7,10 +7,8 @@
 		"build": "tsc && devScript --copyOnly"
 	},
 	"dependencies": {
-		"@keyv/redis": "^2.5.1",
 		"@sentry/node": "^6.19.2",
 		"@sentry/tracing": "^6.19.2",
-		"apollo-datasource": "^3.3.1",
 		"apollo-server-cache-redis": "^3.3.1",
 		"apollo-server-caching": "^3.3.0",
 		"apollo-server-core": "^3.6.6",
@@ -21,18 +19,19 @@
 		"axios": "^0.27.0",
 		"fastify": "^3.27.4",
 		"graphql": "^16.3.0",
-		"ioredis": "^5.0.1",
+		"keyv": "^4.5.4",
+		"keyv-anyredis": "^3.3.0",
 		"language-flag-colors": "^2.1.1",
 		"lodash": "^4.17.21",
 		"mongodb": "^4.4.1",
 		"mongodb-caching": "^1.1.0",
 		"p-event": "^4.2.0",
+		"redis": "^4.6.12",
 		"source-map-support": "^0.5.21",
 		"validator": "^13.7.0"
 	},
 	"devDependencies": {
 		"@types/archiver": "^5.3.1",
-		"@types/ioredis": "^4.28.10",
 		"@types/lodash": "^4.14.180",
 		"@types/node": "^18.16.0",
 		"@types/validator": "^13.7.2",

apps/worker/src/index.ts

@@ -12,17 +12,15 @@ import {
 import { ApolloServer } from "apollo-server-fastify";
 import responseCachePlugin from "apollo-server-plugin-response-cache";
 import fastify, { FastifyContext, FastifyReply, FastifyRequest } from "fastify";
-import Redis from "ioredis";
+import { createCluster } from "redis";
 import { MongoClient } from "mongodb";
-import pEvent from "p-event";
 
 import appUpdate from "./generic/appUpdate";
 import ffUpdates from "./generic/ffUpdates";
 import zippedPresences from "./generic/zippedPresences";
 import fastifyAppClosePlugin from "./plugins/fastifyAppClosePlugin";
 import sentryPlugin from "./plugins/sentryPlugin";
 import dataSources from "./util/dataSources";
-import zipPresences from "./util/functions/zipPresences";
 import deleteScience from "./v2/deleteScience";
 import versions from "./v2/versions";
 import { resolvers as v3Resolvers } from "./v3/resolvers";
@@ -45,25 +43,25 @@ if (process.env.SENTRY_DSN)
 export const mongodb = new MongoClient(process.env.MONGO_URL!, {
 		appName: "PreMiD-API-Worker"
 	}),
-	redis = new Redis({
-		sentinels: [
+	redis = createCluster({
+		rootNodes: [
 			{
-				host: process.env.REDIS_HOST || "localhost",
-				port: parseInt(process.env.REDIS_PORT || "26379")
+				url: process.env.REDIS_URL || "redis://localhost:6379"
 			}
 		],
-		name: "mymaster"
+		useReplicas: true
 	}),
 	baseRedisCache = new BaseRedisCache({
 		//@ts-ignore
 		client: redis
 	}),
-	dSources = dataSources(),
 	app = fastify({
 		connectionTimeout: 10_000,
 		keepAliveTimeout: 10_000
 	});
 
+export let dSources: ReturnType<typeof dataSources>;
+
 export let v3Server: ApolloServer<FastifyContext>,
 	v4Server: ApolloServer<FastifyContext>;
 
@@ -72,6 +70,14 @@ export interface Context {
 }
 
 async function run() {
+	redis.setMaxListeners(12);
+	redis.on("error", error => {
+		console.log(error);
+	});
+	await redis.connect();
+
+	dSources = dataSources();
+
 	const apolloGenericSettings = {
 		dataSources: () => dSources,
 		context: (req: FastifyRequest, res: FastifyReply) => {
@@ -111,17 +117,7 @@ async function run() {
 		resolvers: await v4Resolvers
 	});
 
-	await Promise.all([
-		mongodb.connect(),
-		pEvent(redis, "connect"),
-		v3Server.start(),
-		v4Server.start(),
-		zipPresences()
-	]);
-
-	setInterval(() => {
-		zipPresences();
-	}, 60_000);
+	await Promise.all([mongodb.connect(), v3Server.start(), v4Server.start()]);
 
 	app.addHook("onError", (_, _1, error, done) => {
 		Sentry.captureException(error);

apps/worker/src/util/dataSources.ts

@@ -15,79 +15,84 @@ import { Sponsors } from "../v3/fields/sponsors";
 import { Usage } from "../v3/fields/usage";
 import { Versions } from "../v3/fields/versions";
 import { Strings } from "../v4/fields/strings";
-import KeyvRedis from "@keyv/redis";
+import { KeyvRedis } from "./keyv";
 
 export let pmdDb: Db, dSources: ReturnType<typeof dataSources>;
 
+let store: KeyvRedis;
+
 export default function dataSources() {
+	if (!store) {
+		store = new KeyvRedis(redis);
+	}
 	pmdDb = mongodb.db("PreMiD");
 
 	const keyvDefaultOptions = {
-		store: new KeyvRedis(redis),
+		store: store,
 		ttl: 60 * 1000
 	};
 
 	const sources = {
 		usage: new Usage(pmdDb.collection("science"), {
 			...keyvDefaultOptions,
 			namespace: "science"
-		}),
+		} as any),
 		presences: new Presences(pmdDb.collection("presences"), {
 			...keyvDefaultOptions,
 			namespace: "presences"
-		}),
+		} as any),
 		versions: new Versions(pmdDb.collection("versions"), {
 			...keyvDefaultOptions,
 			namespace: "versions"
-		}),
+		} as any),
 		discordUsers: new DiscordUsers(pmdDb.collection("discordUsers"), {
 			...keyvDefaultOptions,
 			namespace: "discordUsers"
-		}),
+		} as any),
 		langFiles: new LangFiles(pmdDb.collection("langFiles"), {
 			...keyvDefaultOptions,
 			namespace: "langFiles"
-		}),
+		} as any),
 		strings: new Strings(pmdDb.collection("strings"), {
 			...keyvDefaultOptions,
 			namespace: "strings"
-		}),
+		} as any),
 		sponsors: new Sponsors(pmdDb.collection("sponsors"), {
 			...keyvDefaultOptions,
 			namespace: "sponsors"
-		}),
+		} as any),
 		partners: new Partners(pmdDb.collection("partners"), {
 			...keyvDefaultOptions,
 			namespace: "partners"
-		}),
+		} as any),
 		jobs: new Jobs(pmdDb.collection("jobs"), {
 			...keyvDefaultOptions,
 			namespace: "jobs"
-		}),
+		} as any),
 		downloads: new Downloads(pmdDb.collection("downloads"), {
 			...keyvDefaultOptions,
 			namespace: "downloads"
-		}),
+		} as any),
 		alphaUsers: new AlphaUsers(pmdDb.collection("alphaUsers"), {
 			...keyvDefaultOptions,
 			namespace: "alphaUsers"
-		}),
+		} as any),
 		betaUsers: new BetaUsers(pmdDb.collection("betaUsers"), {
 			...keyvDefaultOptions,
 			namespace: "betaUsers"
-		}),
+		} as any),
 		credits: new Credits(pmdDb.collection("credits"), {
 			...keyvDefaultOptions,
 			namespace: "credits"
-		}),
+		} as any),
 		benefits: new Benefits(pmdDb.collection("benefits"), {
 			...keyvDefaultOptions,
 			namespace: "benefits"
-		}),
+		} as any),
 		ffUpdates: new FFUpdates(pmdDb.collection("ffUpdates"), {
 			...keyvDefaultOptions,
 			namespace: "ffUpdates"
-		})
+		} as any)
 	};
 
 	dSources = sources;