Merge pull request #589 from TaloDev/develop

Release 0.85.0

Author: tudddorrr

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "game-services",
-  "version": "0.84.1",
+  "version": "0.85.0",
   "description": "",
   "main": "src/index.ts",
   "scripts": {

package.json

@@ -1,6 +1,5 @@
 import 'dotenv/config'
 import entities from '../entities'
-import subscribers from '../entities/subscribers'
 import migrationsList from '../migrations'
 import { TsMorphMetadataProvider } from '@mikro-orm/reflection'
 import { Migrator } from '@mikro-orm/migrations'
@@ -19,7 +18,6 @@ const ormConfig = defineConfig({
     migrationsList,
     path: 'src/migrations' // for generating migrations via the cli
   },
-  subscribers,
   metadataProvider: TsMorphMetadataProvider,
   extensions: [Migrator],
   pool: {

src/config/mikro-orm.config.ts

@@ -122,6 +122,7 @@ export default class PlayerGroup {
   async checkMembership(em: EntityManager) {
     const players = await this.getQuery(em).getResultList()
     this.members.set(players)
+    await em.flush()
   }
 
   async isPlayerEligible(em: EntityManager, player: Player): Promise<boolean> {

src/entities/player-group.ts

@@ -14,6 +14,7 @@ import createClickHouseClient from '../lib/clickhouse/createClient'
 import { captureException } from '@sentry/node'
 import { ClickHouseClient } from '@clickhouse/client'
 import GameChannel, { GameChannelLeavingReason } from './game-channel'
+import checkGroupMemberships from '../lib/groups/checkGroupMemberships'
 
 @Entity()
 export default class Player {
@@ -71,7 +72,9 @@ export default class Player {
   }
 
   setProps(props: { key: string, value: string }[]) {
-    this.props.set(props.map(({ key, value }) => new PlayerProp(this, key, value)))
+    this.props.set(
+      props.map(({ key, value }) => new PlayerProp(this, key, value))
+    )
   }
 
   async insertSession(clickhouse: ClickHouseClient, session: PlayerSession) {
@@ -149,9 +152,11 @@ export default class Player {
     await em.flush()
 
     const conns = await socket.findConnectionsAsync(async (conn) => {
-      return conn.hasScope(APIKeyScope.READ_PLAYERS) &&
+      return (
+        conn.hasScope(APIKeyScope.READ_PLAYERS) &&
         !!conn.playerAliasId &&
         this.game.id === (await conn.getPlayerAlias())?.player.game.id
+      )
     })
     await sendMessages(conns, 'v1.players.presence.updated', {
       presence: this.presence,
@@ -194,8 +199,14 @@ export default class Player {
     this.addProp('META_DEV_BUILD', '1')
   }
 
+  async checkGroupMemberships(em: EntityManager) {
+    await checkGroupMemberships(em, this)
+  }
+
   toJSON() {
-    const presence = this.presence ? { ...this.presence.toJSON(), playerAlias: undefined } : null
+    const presence = this.presence
+      ? { ...this.presence.toJSON(), playerAlias: undefined }
+      : null
 
     return {
       id: this.id,
@@ -204,7 +215,9 @@ export default class Player {
       devBuild: this.devBuild,
       createdAt: this.createdAt,
       lastSeenAt: this.lastSeenAt,
-      groups: this.groups.getItems().map((group) => ({ id: group.id, name: group.name })),
+      groups: this.groups
+        .getItems()
+        .map((group) => ({ id: group.id, name: group.name })),
       auth: this.auth ?? undefined,
       presence
     }

src/entities/player.ts

@@ -2,54 +2,38 @@ import { EntityManager } from '@mikro-orm/mysql'
 import Player from '../../entities/player'
 import PlayerGroup from '../../entities/player-group'
 import { getResultCacheOptions } from '../perf/getResultCacheOptions'
+import { captureException } from '@sentry/node'
+import { createRedisConnection } from '../../config/redis.config'
 
-const enableLogging = process.env.NODE_ENV !== 'test'
-
-class PlayerGroupMember {
-  player: Player
-  group: PlayerGroup
-
-  constructor(player: Player, group: PlayerGroup) {
-    this.player = player
-    this.group = group
-  }
-
-  toJSON() {
-    return {
-      player_id: this.player.id,
-      player_group_id: this.group.id
-    }
-  }
-}
-
-export default async function checkGroupMemberships(em: EntityManager, player: Player): Promise<boolean> {
-  const groups = await em.repo(PlayerGroup).find({
-    game: player.game
-  }, getResultCacheOptions(PlayerGroup.getCacheKey(player.game)))
+let redis: ReturnType<typeof createRedisConnection>
 
-  if (groups.length === 0) {
-    return false
-  }
+const enableLogging = process.env.NODE_ENV !== 'test'
 
+async function runMembershipChecksForGroups(em: EntityManager, player: Player, groups: PlayerGroup[]) {
   const label = `Checking group memberships for ${player.id}`
 
   /* v8 ignore next 3 */
   if (enableLogging) {
     console.time(label)
   }
 
+  let shouldFlush = false
+
   for (const group of groups) {
     const eligible = await group.isPlayerEligible(em, player)
-    const groupMember = new PlayerGroupMember(player, group).toJSON()
-
-    if (eligible) {
-      await em.qb('player_group_members')
-        .insert(groupMember)
-        .onConflict()
-        .ignore()
-        .execute()
-    } else {
-      await em.nativeDelete('player_group_members', groupMember)
+    const isInGroup = player.groups.contains(group)
+
+    const eligibleButNotInGroup = eligible && !isInGroup
+    const inGroupButNotEligible = !eligible && isInGroup
+
+    if (eligibleButNotInGroup) {
+      player.groups.add(group)
+    } else if (inGroupButNotEligible) {
+      player.groups.remove(group)
+    }
+
+    if (eligibleButNotInGroup || inGroupButNotEligible) {
+      shouldFlush = true
     }
   }
 
@@ -58,5 +42,40 @@ export default async function checkGroupMemberships(em: EntityManager, player: P
     console.timeEnd(label)
   }
 
-  return true
+  return shouldFlush
+}
+
+export default async function checkGroupMemberships(em: EntityManager, player: Player) {
+  if (!redis) {
+    redis = createRedisConnection()
+  }
+
+  const groups = await em.repo(PlayerGroup).find({
+    game: player.game
+  }, getResultCacheOptions(PlayerGroup.getCacheKey(player.game)))
+
+  if (groups.length === 0) {
+    return
+  }
+
+  const redisKey = `checkMembership:${player.id}`
+  let lockCreated: 'OK' | null = null
+  let shouldFlush = false
+
+  try {
+    lockCreated = await redis.set(redisKey, '1', 'EX', 30, 'NX')
+    if (lockCreated) {
+      shouldFlush = await runMembershipChecksForGroups(em, player, groups)
+    }
+  } catch (err) {
+    console.error(`Failed checking memberships: ${(err as Error).message}`)
+    captureException(err)
+  } finally {
+    if (lockCreated) {
+      if (shouldFlush) {
+        await em.flush()
+      }
+      await redis.del(redisKey)
+    }
+  }
 }

src/entities/subscribers/index.ts

@@ -1,5 +1,6 @@
 import Event from '../../../entities/event'
-import { FlushMetricsQueueHandler } from './flush-metrics-queue-handler'
+import Player from '../../../entities/player'
+import { FlushMetricsQueueHandler, postFlushCheckMemberships } from './flush-metrics-queue-handler'
 
 export class FlushEventsQueueHandler extends FlushMetricsQueueHandler<Event> {
   constructor() {
@@ -14,6 +15,26 @@ export class FlushEventsQueueHandler extends FlushMetricsQueueHandler<Event> {
         values: values.flatMap((event) => event.getInsertableProps()),
         format: 'JSONEachRow'
       })
+    }, {
+      postFlush: async (values) => {
+        const playerSet = this.buildPlayerSet(values)
+
+        if (playerSet.size > 0) {
+          /* v8 ignore next 3 */
+          if (process.env.NODE_ENV !== 'test') {
+            console.info(`FlushEventsQueueHandler checking groups for ${playerSet.size} players`)
+          }
+          await postFlushCheckMemberships(Array.from(playerSet))
+        }
+      }
     })
   }
+
+  buildPlayerSet(values: Event[]) {
+    const playerMap = new Map<string, Player>()
+    for (const event of values) {
+      playerMap.set(event.playerAlias.player.id, event.playerAlias.player)
+    }
+    return new Set(playerMap.values())
+  }
 }