Enable map counting workers with two grouped SQL statements

apps/scheduler/src/main.ts

@@ -1,6 +1,7 @@
 import { masterServerScheduler } from './schedulers/masterServerScheduler';
 import { gameServerScheduler } from './schedulers/gameServerScheduler';
 import { gameTypeScheduler } from './schedulers/gameTypeScheduler';
+import { mapScheduler } from './schedulers/mapScheduler';
 import { addDefaultGameTypes } from './addDefaultGameTypes';
 import { addDefaultMasterServers } from './addDefaultMasterServers';
 import { fillClanActivePlayerCountScheduler } from './schedulers/fillClanActivePlayerCountScheduler';
@@ -20,6 +21,7 @@ async function main() {
   masterServerScheduler();
   gameServerScheduler();
   gameTypeScheduler();
+  mapScheduler();
   fillClanActivePlayerCountScheduler();
   updateGlobalCountsScheduler();
   archiveSnapshotsScheduler();

apps/scheduler/src/schedulers/mapScheduler.ts

@@ -1,42 +1,13 @@
 import { scheduleMapCount } from "@teerank/teerank";
-import { hoursToMilliseconds, minutesToMilliseconds } from "date-fns";
-import { prisma } from "../prisma";
-import { schedule, scheduleWithSpread } from "../utils";
-
-let lastId = 0;
+import { hoursToMilliseconds } from "date-fns";
+import { schedule } from "../utils";
 
 export async function mapScheduler() {
-  schedule(minutesToMilliseconds(5), async () => {
-    const maps = await prisma.map.findMany({
-      where: {
-        id: {
-          gt: lastId,
-        },
-      },
-      select: {
-        gameTypeName: true,
-        name: true,
-        id: true,
-      },
-      orderBy: {
-        id: 'asc',
-      },
-    });
-
-    for (const map of maps) {
-      scheduleWithSpread(hoursToMilliseconds(24), async () => {
-        await scheduleMapCount({
-          gameTypeName: map.gameTypeName,
-          mapName: map.name,
-          mapId: map.id,
-        });
-      });
-    }
-
-    console.log(`Scheduled ${maps.length} new maps`);
+  schedule(hoursToMilliseconds(24), async () => {
+    await scheduleMapCount({ mode: 'full' });
+  });
 
-    if (maps.length > 0) {
-      lastId = maps[maps.length - 1].id;
-    }
+  schedule(hoursToMilliseconds(1), async () => {
+    await scheduleMapCount({ mode: 'gameServers' });
   });
 }

apps/worker/src/workers/updateMapsCounts.ts

@@ -1,46 +1,13 @@
+import { updateMapsCounts, updateMapsGameServerCounts } from "@prisma/client/sql";
 import { prisma } from "../prisma";
 import { MapCountJobData, processMapCountJobs } from "@teerank/teerank"
 
 export async function updateMapsCount(data: MapCountJobData) {
-  const map = await prisma.map.findUniqueOrThrow({
-    select: {
-      _count: {
-        select: {
-          playerInfoMaps: true,
-          clanInfoMaps: true,
-        },
-      },
-    },
-    where: {
-      name_gameTypeName: {
-        name: data.mapName,
-        gameTypeName: data.gameTypeName,
-      },
-    },
-  });
-
-  const gameServerCount = await prisma.gameServerState.count({
-    where: {
-      map: {
-        name: data.mapName,
-        gameTypeName: data.gameTypeName,
-      },
-    },
-  });
-
-  await prisma.map.update({
-    where: {
-      name_gameTypeName: {
-        name: data.mapName,
-        gameTypeName: data.gameTypeName,
-      },
-    },
-    data: {
-      playerCount: map._count.playerInfoMaps,
-      clanCount: map._count.clanInfoMaps,
-      gameServerCount,
-    },
-  });
+  if (data.mode === 'full') {
+    await prisma.$queryRawTyped(updateMapsCounts());
+  } else {
+    await prisma.$queryRawTyped(updateMapsGameServerCounts());
+  }
 }
 
 export async function startUpdateMapsCountsWorker() {

libs/prisma/prisma/sql/updateMapsCounts.sql

@@ -0,0 +1,18 @@
+UPDATE "Map" SET
+  "playerCount" = counts."playerCount",
+  "clanCount" = counts."clanCount",
+  "gameServerCount" = counts."gameServerCount"
+FROM (
+  SELECT
+    m.id,
+    COALESCE(p.count, 0)::int4 AS "playerCount",
+    COALESCE(c.count, 0)::int4 AS "clanCount",
+    COALESCE(g.count, 0)::int4 AS "gameServerCount"
+  FROM "Map" m
+  LEFT JOIN (SELECT "mapId", count(*) AS count FROM "PlayerInfoMap" GROUP BY "mapId") p ON p."mapId" = m.id
+  LEFT JOIN (SELECT "mapId", count(*) AS count FROM "ClanInfoMap" GROUP BY "mapId") c ON c."mapId" = m.id
+  LEFT JOIN (SELECT "mapId", count(*) AS count FROM "GameServerState" GROUP BY "mapId") g ON g."mapId" = m.id
+) counts
+WHERE "Map".id = counts.id
+  AND ("Map"."playerCount", "Map"."clanCount", "Map"."gameServerCount")
+      IS DISTINCT FROM (counts."playerCount", counts."clanCount", counts."gameServerCount");

libs/prisma/prisma/sql/updateMapsGameServerCounts.sql

@@ -0,0 +1,11 @@
+UPDATE "Map" SET
+  "gameServerCount" = counts."gameServerCount"
+FROM (
+  SELECT
+    m.id,
+    COALESCE(g.count, 0)::int4 AS "gameServerCount"
+  FROM "Map" m
+  LEFT JOIN (SELECT "mapId", count(*) AS count FROM "GameServerState" GROUP BY "mapId") g ON g."mapId" = m.id
+) counts
+WHERE "Map".id = counts.id
+  AND "Map"."gameServerCount" IS DISTINCT FROM counts."gameServerCount";

libs/teerank/src/lib/bullmq/queueMapCount.ts

@@ -2,31 +2,27 @@ import { Job, Queue, Worker } from "bullmq";
 import { bullmqConnection, lastCompletedJobDate } from "./config";
 import { z } from "zod";
 import { minutesToSeconds } from "date-fns";
-import { getEnvInt } from "../utils";
 
 let mapCountQueue: Queue | null = null;
 
 const QUEUE_NAME_MAP_COUNT = 'map-count';
-const UPDATE_MAPS_COUNTS_CONCURRENCY = getEnvInt('UPDATE_MAPS_COUNTS_CONCURRENCY', 5);
 
 function getQueueMapCount() {
   mapCountQueue ??= new Queue(QUEUE_NAME_MAP_COUNT, { connection: bullmqConnection });
   return mapCountQueue;
 }
 
 const schema = z.object({
-  gameTypeName: z.string(),
-  mapName: z.string(),
-  mapId: z.number(),
+  mode: z.enum(['full', 'gameServers']),
 });
 
 export type MapCountJobData = z.infer<typeof schema>;
 
 export async function scheduleMapCount(data: MapCountJobData) {
   const queue = getQueueMapCount();
-  await queue.add(`${data.gameTypeName} - ${data.mapName}`, data, {
+  await queue.add(data.mode, data, {
     deduplication: {
-      id: data.mapId.toString(),
+      id: data.mode,
     }
   });
 }
@@ -39,7 +35,7 @@ export async function processMapCountJobs(processor: (data: MapCountJobData) =>
 
   return new Worker(QUEUE_NAME_MAP_COUNT, jobProcessor, {
     connection: bullmqConnection,
-    concurrency: UPDATE_MAPS_COUNTS_CONCURRENCY,
+    concurrency: 1,
     removeOnComplete: {
       age: minutesToSeconds(10),
     },