update stream card, update leaderboards on match finished webhook

src/app/api/neatqueue-webhook/route.ts

@@ -1,4 +1,10 @@
 import crypto from 'node:crypto'
+import { globalEmitter } from '@/lib/events'
+import { syncHistory } from '@/server/api/routers/history'
+import type { PlayerState } from '@/server/api/routers/player-state'
+import { PLAYER_STATE_KEY, redis } from '@/server/redis'
+import { leaderboardService } from '@/server/services/leaderboard'
+import { RANKED_CHANNEL, VANILLA_CHANNEL } from '@/shared/constants'
 import { type NextRequest, NextResponse } from 'next/server'
 
 const EXPECTED_QUERY_SECRET = process.env.WEBHOOK_QUERY_SECRET
@@ -42,8 +48,6 @@ function verifyQuerySecret(req: NextRequest): boolean {
  * Verifies query secret, logs payload, and handles actions.
  */
 export async function POST(req: NextRequest) {
-  let payload: any
-
   try {
     const isVerified = verifyQuerySecret(req)
     if (!isVerified) {
@@ -54,7 +58,66 @@ export async function POST(req: NextRequest) {
       )
     }
 
-    payload = await req.json()
+    const payload = await req.json()
+
+    switch (payload.action) {
+      case 'JOIN_QUEUE': {
+        const state: PlayerState = {
+          status: 'queuing',
+          queueStartTime: Date.now(),
+        }
+        const userId = payload.new_players[0].id
+        console.log('-----JOIN QUEUE-----')
+        console.dir(payload, { depth: null })
+        console.log(userId)
+        await redis.set(PLAYER_STATE_KEY(userId), JSON.stringify(state))
+        globalEmitter.emit(`state-change:${userId}`, state)
+        break
+      }
+
+      case 'MATCH_STARTED': {
+        const playerIds = payload.players.map((p: any) => p.id) as string[]
+
+        await Promise.all(
+          playerIds.map(async (id) => {
+            const state = {
+              status: 'in_game',
+              currentMatch: {
+                opponentId: playerIds.find((p) => p !== id),
+                startTime: Date.now(),
+              },
+            }
+            await redis.set(PLAYER_STATE_KEY(id), JSON.stringify(state))
+            globalEmitter.emit(`state-change:${id}`, state)
+          })
+        )
+        break
+      }
+
+      case 'MATCH_COMPLETED': {
+        const playerIds = payload.teams.map((p: any) => p[0].id) as string[]
+        console.log({ playerIds })
+        await syncHistory()
+        if ([RANKED_CHANNEL, VANILLA_CHANNEL].includes(payload.channel)) {
+          await leaderboardService.refreshLeaderboard(payload.channel)
+        }
+        await Promise.all(
+          playerIds.map(async (id) => {
+            await redis.del(PLAYER_STATE_KEY(id))
+            globalEmitter.emit(`state-change:${id}`, { status: 'idle' })
+          })
+        ).catch(console.error)
+
+        break
+      }
+
+      case 'LEAVE_QUEUE': {
+        const userId = payload.players_removed[0].id
+        await redis.del(PLAYER_STATE_KEY(userId))
+        globalEmitter.emit(`state-change:${userId}`, { status: 'idle' })
+        break
+      }
+    }
 
     console.log(
       '--- Verified Webhook Received (Query Auth) ---',

src/app/api/refresh-leaderboard/route.ts

@@ -1,5 +1,5 @@
 import { env } from '@/env'
-import { LeaderboardService } from '@/server/services/leaderboard'
+import { leaderboardService } from '@/server/services/leaderboard'
 import { RANKED_CHANNEL, VANILLA_CHANNEL } from '@/shared/constants'
 import { headers } from 'next/headers'
 
@@ -14,12 +14,10 @@ export async function POST() {
   }
 
   try {
-    const service = new LeaderboardService()
-
     for (const channelId of CHANNEL_IDS) {
       try {
         console.log(`refreshing leaderboard for ${channelId}...`)
-        await service.refreshLeaderboard(channelId)
+        await leaderboardService.refreshLeaderboard(channelId)
       } catch (err) {
         console.error('refresh failed:', err)
         return new Response('internal error', { status: 500 })

src/app/stream-card/[id]/_components/stream-card-client.tsx

@@ -1,37 +1,18 @@
 'use client'
 
+import { cn } from '@/lib/utils'
+import type { SelectGames } from '@/server/db/types'
+import type { LeaderboardEntry } from '@/server/services/neatqueue.service'
 import { RANKED_CHANNEL } from '@/shared/constants'
 import { api } from '@/trpc/react'
+import { Swords } from 'lucide-react'
 import { useParams } from 'next/navigation'
-import { useEffect } from 'react'
+import { type ComponentPropsWithoutRef, useEffect, useState } from 'react'
 
-export function StreamCardClient() {
+function getPlayerData(
-  const { id } = useParams()
+  playerLeaderboardEntry: LeaderboardEntry,
-  if (!id || typeof id !== 'string') {
+  games: SelectGames[]
-    return null
+) {
-  }
-
-  const [gamesQueryResult, gamesQuery] =
-    api.history.user_games.useSuspenseQuery({ user_id: id })
-  const games = gamesQueryResult || [] // Ensure games is always an array
-
-  const [rankedUserRank, rankedUserQuery] =
-    api.leaderboard.get_user_rank.useSuspenseQuery({
-      channel_id: RANKED_CHANNEL,
-      user_id: id,
-    })
-
-  useEffect(() => {
-    const interval = setInterval(async () => {
-      await Promise.all([gamesQuery.refetch(), rankedUserQuery.refetch()])
-    }, 1000 * 60)
-
-    return () => clearInterval(interval)
-  }, [])
-
-  if (!rankedUserRank || !games?.length) {
-    return null
-  }
   const filteredGamesByLeaderboard = games.filter(
     (game) => game.gameType === 'ranked'
   )
@@ -53,10 +34,10 @@ export function StreamCardClient() {
   }
 
   const lastGame = filteredGamesByLeaderboard.at(0)
-
   const currentName = lastGame?.playerName
   const meaningful_games = games_played - ties
-  const playerData = {
+
+  return {
     username: currentName,
     games: games_played,
     meaningful_games,
@@ -67,22 +48,151 @@ export function StreamCardClient() {
       meaningful_games > 0 ? Math.ceil((wins / meaningful_games) * 100) : 0,
     lossRate:
       meaningful_games > 0 ? Math.floor((losses / meaningful_games) * 100) : 0,
-    rank: rankedUserRank.rank,
+    rank: playerLeaderboardEntry.rank,
-    mmr: Math.round(rankedUserRank.mmr),
+    mmr: Math.round(playerLeaderboardEntry.mmr),
-    mmrChange: `${(lastGame?.mmrChange ?? 0) >= 0 ? '+' : ''}${Math.round(lastGame?.mmrChange ?? 0)}`,
+    mmrChange: `${(lastGame?.mmrChange ?? 0) >= 0 ? '+' : ''}${Math.round(
-    streak: rankedUserRank?.streak,
+      lastGame?.mmrChange ?? 0
+    )}`,
+    streak: playerLeaderboardEntry?.streak,
   }
+}
+
+export function StreamCardClient() {
+  const { id } = useParams()
+  if (!id || typeof id !== 'string') {
+    return null
+  }
+
+  const [gamesQueryResult, gamesQuery] =
+    api.history.user_games.useSuspenseQuery({ user_id: id })
+  const games = gamesQueryResult || []
+
+  const [rankedUserRank, rankedUserQuery] =
+    api.leaderboard.get_user_rank.useSuspenseQuery({
+      channel_id: RANKED_CHANNEL,
+      user_id: id,
+    })
+
+  const result = api.playerState.onStateChange.useSubscription(
+    { userId: id },
+    {
+      onData: async () => {
+        await Promise.all([gamesQuery.refetch(), rankedUserQuery.refetch()])
+      },
+    }
+  )
+
+  const playerState = result.data?.data
+
+  if (!rankedUserRank || !games?.length) {
+    return null
+  }
+
+  const playerData = getPlayerData(rankedUserRank, games)
+
+  const isQueuing = playerState?.status === 'queuing'
+  const opponentId = playerState?.currentMatch?.opponentId
+  return (
+    <div className={'flex items-center gap-2'} style={{ zoom: '200%' }}>
+      <PlayerInfo playerData={playerData}>
+        {isQueuing && playerState.queueStartTime && (
+          <QueueTimer startTime={playerState.queueStartTime} />
+        )}
+      </PlayerInfo>
+      {opponentId && (
+        <>
+          <span>
+            <Swords />
+          </span>{' '}
+          <Opponent id={opponentId} />
+        </>
+      )}
+    </div>
+  )
+}
+
+export function formatDuration(ms: number): string {
+  const seconds = Math.floor(ms / 1000)
+  const minutes = Math.floor(seconds / 60)
+  const hours = Math.floor(minutes / 60)
+
+  if (hours > 0) {
+    return `${hours}h ${minutes % 60}m`
+  }
+  if (minutes > 0) {
+    return `${minutes}m ${seconds % 60}s`
+  }
+  return `${seconds}s`
+}
+
+function QueueTimer({ startTime }: { startTime: number }) {
+  const [queueTime, setQueueTime] = useState(Date.now() - startTime)
+
+  useEffect(() => {
+    const interval = setInterval(() => {
+      setQueueTime(Date.now() - startTime)
+    }, 1000)
+    return () => clearInterval(interval)
+  })
+
+  return (
+    <div className='flex animate-pulse items-center gap-1.5 border-slate-700 px-2'>
+      <div>Queueing for </div>
+      <div className='font-bold text-emerald-400'>
+        {formatDuration(queueTime)}
+      </div>
+    </div>
+  )
+}
+
+function Opponent({ id }: { id: string }) {
+  const [gamesQueryResult] = api.history.user_games.useSuspenseQuery({
+    user_id: id,
+  })
+  const games = gamesQueryResult || []
+
+  const [rankedUserRank] = api.leaderboard.get_user_rank.useSuspenseQuery({
+    channel_id: RANKED_CHANNEL,
+    user_id: id,
+  })
+  if (!rankedUserRank || !games?.length) {
+    return null
+  }
+
+  const playerData = getPlayerData(rankedUserRank, games)
+  return <PlayerInfo playerData={playerData} isReverse />
+}
+
+function PlayerInfo({
+  playerData,
+  className,
+  children,
+  isReverse = false,
+  ...rest
+}: {
+  playerData: ReturnType<typeof getPlayerData>
+  isReverse?: boolean
+} & ComponentPropsWithoutRef<'div'>) {
   return (
     <div
-      style={{ zoom: '200%' }}
+      className={cn(
-      className='flex h-10 w-fit max-w-[800px] items-center overflow-hidden rounded-md border-2 border-slate-800 bg-slate-900/90 text-white shadow-lg backdrop-blur-sm'
+        'flex h-10 w-fit max-w-[800px] items-center overflow-hidden border-2 border-slate-800 bg-slate-900/90 text-white shadow-lg backdrop-blur-sm',
+        isReverse ? 'flex-row-reverse' : 'flex-row',
+        className
+      )}
+      {...rest}
     >
-      <div className='flex h-full items-center gap-1 border-slate-700 border-r bg-gradient-to-r from-indigo-600 to-purple-600 px-2'>
+      <div className='flex aspect-square h-full items-center justify-center gap-1 border-slate-700 border-r bg-gradient-to-r from-indigo-600 to-purple-600 px-2'>
         <span className='font-bold text-sm'>{playerData.rank}</span>
       </div>
 
       {/* Player Name */}
-      <div className='max-w-[180px] flex-shrink-0 border-slate-700 border-r px-2'>
+      <div
+        className={cn(
+          'max-w-[180px] flex-shrink-0 border-slate-700 px-2',
+          !isReverse && 'border-r'
+        )}
+      >
         <div className='truncate font-medium'>{playerData.username}</div>
       </div>
 
@@ -95,7 +205,7 @@ export function StreamCardClient() {
 
       {/* Win Rate */}
       <div className='flex items-center gap-1.5 text-nowrap border-slate-700 border-r px-2'>
-        <div className={'text-nowrap'}>Win Rate:</div>
+        <div className='text-nowrap'>Win Rate:</div>
         <div className='text font-bold text-emerald-400'>
           {playerData.winRate}%
         </div>
@@ -117,10 +227,17 @@ export function StreamCardClient() {
       </div>
 
       {/* Streak */}
-      <div className='flex items-center gap-1.5 border-slate-700 px-2'>
+      <div
+        className={cn(
+          'flex items-center gap-1.5 border-slate-700 px-2',
+          (children || isReverse) && 'border-r'
+        )}
+      >
         <div>Streak:</div>
         <div className='font-bold text-emerald-400'>{playerData.streak}</div>
       </div>
+
+      {children}
     </div>
   )
 }

src/lib/events.ts

@@ -0,0 +1,56 @@
+import { EventEmitter } from 'node:events'
+export const globalEmitter = new EventEmitter()
+export function createEventIterator<T>(
+  emitter: EventEmitter,
+  eventName: string,
+  opts?: { signal?: AbortSignal }
+): AsyncIterableIterator<[T]> {
+  const events: [T][] = []
+  let resolvePromise: (() => void) | null = null
+  let done = false
+
+  const push = (data: T) => {
+    events.push([data])
+    if (resolvePromise) {
+      resolvePromise()
+      resolvePromise = null
+    }
+  }
+
+  const cleanup = () => {
+    done = true
+    emitter.off(eventName, push)
+    if (resolvePromise) {
+      resolvePromise()
+    }
+  }
+
+  emitter.on(eventName, push)
+  opts?.signal?.addEventListener('abort', cleanup)
+
+  return {
+    [Symbol.asyncIterator]() {
+      return this
+    },
+    async next() {
+      if (events.length > 0) {
+        return { value: events.shift()!, done: false }
+      }
+      if (done) {
+        return { value: undefined, done: true }
+      }
+      await new Promise<void>((resolve) => {
+        resolvePromise = resolve
+      })
+      return this.next()
+    },
+    async return() {
+      cleanup()
+      return { value: undefined, done: true }
+    },
+    async throw(error) {
+      cleanup()
+      throw error
+    },
+  }
+}

src/server/api/root.ts

@@ -1,6 +1,7 @@
 import { discord_router } from '@/server/api/routers/discord'
 import { history_router } from '@/server/api/routers/history'
 import { leaderboard_router } from '@/server/api/routers/leaderboard'
+import { playerStateRouter } from '@/server/api/routers/player-state'
 import { createCallerFactory, createTRPCRouter } from '@/server/api/trpc'
 
 /**
@@ -12,6 +13,7 @@ export const appRouter = createTRPCRouter({
   history: history_router,
   discord: discord_router,
   leaderboard: leaderboard_router,
+  playerState: playerStateRouter,
 })
 
 // export type definition of API

src/server/api/routers/player-state.ts

@@ -0,0 +1,50 @@
+import { createEventIterator, globalEmitter } from '@/lib/events'
+import { redis } from '@/server/redis'
+import { tracked } from '@trpc/server'
+import { z } from 'zod'
+import { createTRPCRouter, publicProcedure } from '../trpc'
+
+export type PlayerState = {
+  status: 'idle' | 'queuing' | 'in_game'
+  queueStartTime?: number
+  currentMatch?: {
+    opponentId: string
+    startTime: number
+  }
+}
+
+const PLAYER_STATE_KEY = (userId: string) => `player:${userId}:state`
+
+export const playerStateRouter = createTRPCRouter({
+  getState: publicProcedure
+    .input(z.string())
+    .query(async ({ input: userId }) => {
+      const state = await redis.get(PLAYER_STATE_KEY(userId))
+      return state ? (JSON.parse(state) as PlayerState) : null
+    }),
+  onStateChange: publicProcedure
+    .input(
+      z.object({
+        userId: z.string(),
+        lastEventId: z.string().optional(),
+      })
+    )
+    .subscription(async function* ({ input, ctx, signal }) {
+      const iterator = createEventIterator<PlayerState>(
+        globalEmitter,
+        `state-change:${input.userId}`,
+        { signal: signal }
+      )
+
+      // get initial state
+      const initialState = await redis.get(PLAYER_STATE_KEY(input.userId))
+      if (initialState) {
+        yield tracked('initial', JSON.parse(initialState) as PlayerState)
+      }
+
+      // listen for updates
+      for await (const [state] of iterator) {
+        yield tracked(Date.now().toString(), state)
+      }
+    }),
+})

src/server/redis.ts

@@ -2,3 +2,5 @@ import { env } from '@/env'
 import { Redis } from 'ioredis'
 
 export const redis = new Redis(env.REDIS_URL)
+
+export const PLAYER_STATE_KEY = (userId: string) => `player:${userId}:state`

src/server/services/leaderboard.ts

@@ -1,5 +1,5 @@
 import { redis } from '../redis'
-import { neatqueue_service } from './neatqueue.service'
+import { type LeaderboardEntry, neatqueue_service } from './neatqueue.service'
 
 export class LeaderboardService {
   private getZSetKey(channel_id: string) {
@@ -56,23 +56,19 @@ export class LeaderboardService {
 
   async getUserRank(channel_id: string, user_id: string) {
     try {
-      const zsetKey = this.getZSetKey(channel_id)
-      const rank = await redis.zrevrank(zsetKey, user_id)
-
-      if (rank === null) return null
-
       const userData = await redis.hgetall(this.getUserKey(user_id, channel_id))
       if (!userData) return null
 
       return {
-        rank: rank + 1,
         ...userData,
         mmr: Number(userData.mmr),
         streak: userData.streak,
-      }
+      } as unknown as LeaderboardEntry
     } catch (error) {
       console.error('Error getting user rank:', error)
       throw error
     }
   }
 }
+
+export const leaderboardService = new LeaderboardService()

src/trpc/react.tsx

@@ -1,7 +1,12 @@
 'use client'
 
 import { type QueryClient, QueryClientProvider } from '@tanstack/react-query'
-import { httpBatchLink, httpBatchStreamLink, loggerLink } from '@trpc/client'
+import {
+  httpBatchLink,
+  httpSubscriptionLink,
+  loggerLink,
+  splitLink,
+} from '@trpc/client'
 import { createTRPCReact } from '@trpc/react-query'
 import type { inferRouterInputs, inferRouterOutputs } from '@trpc/server'
 import { useState } from 'react'
@@ -49,7 +54,16 @@ export function TRPCReactProvider(props: { children: React.ReactNode }) {
             process.env.NODE_ENV === 'development' ||
             (op.direction === 'down' && op.result instanceof Error),
         }),
-        httpBatchLink({
+
+        splitLink({
+          condition(op) {
+            return op.type === 'subscription'
+          },
+          true: httpSubscriptionLink({
+            url: `${getBaseUrl()}/api/trpc`,
+            transformer: SuperJSON,
+          }),
+          false: httpBatchLink({
             transformer: SuperJSON,
             url: `${getBaseUrl()}/api/trpc`,
             headers: () => {
@@ -58,6 +72,7 @@ export function TRPCReactProvider(props: { children: React.ReactNode }) {
               return headers
             },
           }),
+        }),
       ],
     })
   )