+
@@ -95,7 +205,7 @@ export function StreamCardClient() {
{/* Win Rate */}
-
Win Rate:
+
Win Rate:
{playerData.winRate}%
@@ -117,10 +227,17 @@ export function StreamCardClient() {
{/* Streak */}
-
+
Streak:
{playerData.streak}
+
+ {children}
)
}
diff --git a/src/lib/events.ts b/src/lib/events.ts
new file mode 100644
index 0000000..3e6b31d
--- /dev/null
+++ b/src/lib/events.ts
@@ -0,0 +1,56 @@
+import { EventEmitter } from 'node:events'
+export const globalEmitter = new EventEmitter()
+export function createEventIterator
(
+ 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((resolve) => {
+ resolvePromise = resolve
+ })
+ return this.next()
+ },
+ async return() {
+ cleanup()
+ return { value: undefined, done: true }
+ },
+ async throw(error) {
+ cleanup()
+ throw error
+ },
+ }
+}
diff --git a/src/server/api/root.ts b/src/server/api/root.ts
index 41765fb..aa5f044 100644
--- a/src/server/api/root.ts
+++ b/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
diff --git a/src/server/api/routers/player-state.ts b/src/server/api/routers/player-state.ts
new file mode 100644
index 0000000..25e7dbc
--- /dev/null
+++ b/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(
+ 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)
+ }
+ }),
+})
diff --git a/src/server/redis.ts b/src/server/redis.ts
index ec0024a..ddc8972 100644
--- a/src/server/redis.ts
+++ b/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`
diff --git a/src/server/services/leaderboard.ts b/src/server/services/leaderboard.ts
index e54e546..d15c7d0 100644
--- a/src/server/services/leaderboard.ts
+++ b/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()
diff --git a/src/trpc/react.tsx b/src/trpc/react.tsx
index dabe85a..8aa199a 100644
--- a/src/trpc/react.tsx
+++ b/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,14 +54,24 @@ export function TRPCReactProvider(props: { children: React.ReactNode }) {
process.env.NODE_ENV === 'development' ||
(op.direction === 'down' && op.result instanceof Error),
}),
- httpBatchLink({
- transformer: SuperJSON,
- url: `${getBaseUrl()}/api/trpc`,
- headers: () => {
- const headers = new Headers()
- headers.set('x-trpc-source', 'nextjs-react')
- return headers
+
+ splitLink({
+ condition(op) {
+ return op.type === 'subscription'
},
+ true: httpSubscriptionLink({
+ url: `${getBaseUrl()}/api/trpc`,
+ transformer: SuperJSON,
+ }),
+ false: httpBatchLink({
+ transformer: SuperJSON,
+ url: `${getBaseUrl()}/api/trpc`,
+ headers: () => {
+ const headers = new Headers()
+ headers.set('x-trpc-source', 'nextjs-react')
+ return headers
+ },
+ }),
}),
],
})