Implement lobby feature at server side

packages/server/src/do-lobby/do-lobby.ts

@@ -1,30 +1,134 @@
 import { DurableObject } from "cloudflare:workers"
-import { z } from "zod"
+import { eq } from "drizzle-orm"
+import { drizzle } from "drizzle-orm/d1"
+import {
+    UpdateFromClient,
+    UpdateFromServer,
+    updateFromClient,
+} from "shared/src/websocket-api/lobby-api"
+import { Env } from "../worker/env"
+import { users } from "../worker/framework/db-schema"
+import { userFromAuthorizationHeader } from "../worker/framework/helper/user-from-authorization-header"
+import { UserPositionTracker } from "./user-positions-tracker"
+
+interface WebsocketClientContext {
+    userId: number
+    username: string
+}
 
 export class DurableObjectLobby extends DurableObject {
+    private db: ReturnType<typeof drizzle>
+
+    private positionsTracker: UserPositionTracker
+    private connections: Set<WebSocket> = new Set()
+
+    private usersJoined: { username: string }[]
+    private userLeft: { username: string }[]
+
+    constructor(state: DurableObjectState, env: Env) {
+        super(state, env)
+
+        this.db = drizzle(env.DB)
+        this.positionsTracker = new UserPositionTracker()
+
+        this.usersJoined = []
+        this.userLeft = []
+
+        setInterval(() => {
+            const message: UpdateFromServer = {
+                type: "serverUpdate",
+                positionPackets: this.positionsTracker.retrievePackets(),
+                usersConnected: this.usersJoined,
+                usersDisconnected: this.userLeft,
+            }
+
+            const messageString = JSON.stringify(message)
+
+            for (const connection of this.connections) {
+                connection.send(messageString)
+            }
+        }, 1000)
+    }
+
     async fetch(request: Request): Promise<Response> {
         const { 0: client, 1: server } = new WebSocketPair()
 
+        const context = await this.contextFromRequest(request)
+
+        if (!context) {
+            console.warn("Failed to load context for user")
+            return new Response(null, {
+                status: 401,
+            })
+        }
+
         // why does server not have accept when all examples use it???
         ;(server as any).accept()
 
+        this.connections.add(server)
+        this.usersJoined.push({ username: context.username })
+
+        console.log(`Client ${context.username} connected (${this.connections.size} total)`)
+
         server.addEventListener("message", event => {
-            const parsed = z
-                .object({
-                    type: z.string(),
-                })
-                .safeParse(event.data)
-
-            if (parsed.success) {
-                const t = parsed.data.type
+            const message = updateFromClient.safeParse(JSON.parse(event.data))
+
+            if (message.success === false) {
+                server.close(1008, "Invalid message")
+                return
+            }
+
+            switch (message.data.type) {
+                case "clientUpdate":
+                    this.onClientUpdate(message.data, context)
+                    break
             }
 
             return undefined
         })
 
+        server.addEventListener("close", () => {
+            this.connections.delete(client)
+            this.userLeft.push({ username: context.username })
+
+            console.log(
+                `Client ${context.username} disconnected (${this.connections.size} remaining)`,
+            )
+        })
+
         return new Response(null, {
             status: 101,
             webSocket: client,
         })
     }
+
+    private async contextFromRequest(
+        request: Request,
+    ): Promise<WebsocketClientContext | undefined> {
+        const user = userFromAuthorizationHeader(
+            this.env as Env,
+            new URL(request.url).searchParams.get("authorization"),
+        )
+
+        if (!user) {
+            console.log("No user")
+            return undefined
+        }
+
+        const [dbUser] = await this.db.select().from(users).where(eq(users.id, user.id))
+
+        if (!dbUser) {
+            console.log("No db user")
+            return undefined
+        }
+
+        return {
+            userId: user.id,
+            username: dbUser.username,
+        }
+    }
+
+    private onClientUpdate(message: UpdateFromClient, context: WebsocketClientContext) {
+        this.positionsTracker.addPositions(context.username, message.positions)
+    }
 }

packages/server/src/do-lobby/user-positions-tracker.ts

@@ -0,0 +1,47 @@
+import { TRPCError } from "@trpc/server"
+import { Point } from "runtime/src/model/point"
+import { PositionsPacket, UPDATE_POSITIONS_EVERY_MS } from "shared/src/websocket-api/lobby-api"
+
+const RECEIVE_POS_TOLERANCE_MS = UPDATE_POSITIONS_EVERY_MS * 0.9
+
+interface TrackerForUser {
+    receivedLastAt: number
+}
+
+export class UserPositionTracker {
+    private trackers: Map<string, TrackerForUser> = new Map()
+    private packets: PositionsPacket[] = []
+
+    addPositions(username: string, positions: Point[]) {
+        let tracker = this.trackers.get(username)
+
+        if (!tracker) {
+            tracker = {
+                receivedLastAt: Date.now(),
+            }
+            this.trackers.set(username, tracker)
+        } else {
+            if (Date.now() - tracker.receivedLastAt < RECEIVE_POS_TOLERANCE_MS) {
+                throw new TRPCError({
+                    code: "TOO_MANY_REQUESTS",
+                    message: "Too many updates",
+                })
+            }
+        }
+
+        tracker.receivedLastAt = Date.now()
+
+        const packet: PositionsPacket = {
+            username,
+            positions,
+        }
+
+        this.packets.push(packet)
+    }
+
+    retrievePackets(): PositionsPacket[] {
+        const temp = this.packets
+        this.packets = []
+        return temp
+    }
+}

packages/server/src/worker/env.ts

@@ -1,8 +1,11 @@
+import { DurableObjectLobby } from "../do-lobby/do-lobby"
+
 export interface Env {
     CLIENT_URL: string
     API_URL: string
 
     DB: D1Database
+    LOBBY_DO: DurableObjectNamespace<DurableObjectLobby>
 
     AUTH_DISCORD_CLIENT_ID: string
     AUTH_DISCORD_CLIENT_SECRET: string

packages/server/src/worker/fetch.ts

@@ -25,6 +25,24 @@ export async function fetch(request: Request, env: Env): Promise<Response> {
         })
     }
 
+    const url = new URL(request.url)
+
+    if (url.pathname.startsWith("/lobby")) {
+        const upgradeHeader = request.headers.get("Upgrade")
+
+        if (upgradeHeader !== "websocket") {
+            console.log("Invalid upgrade header")
+            return new Response("Invalid upgrade header", {
+                status: 426,
+            })
+        }
+
+        const id = env.LOBBY_DO.idFromName("lobby")
+        const lobby = env.LOBBY_DO.get(id)
+
+        return lobby.fetch(request)
+    }
+
     const response = await fetchRequestHandler({
         endpoint: "/trpc",
         req: request,

packages/server/src/worker/framework/trpc.ts

@@ -26,7 +26,7 @@ export const createContext =
         }
     }
 
-type Context = inferAsyncReturnType<inferAsyncReturnType<typeof createContext>>
+export type Context = inferAsyncReturnType<inferAsyncReturnType<typeof createContext>>
 
 const t = initTRPC.context<Context>().create({
     transformer: superjson,

packages/server/wrangler.toml

@@ -19,3 +19,11 @@ migrations_dir = "drizzle"
 binding = "DB"
 database_name = "polyburn"
 database_id = "432d6db4-6e41-4a3f-98bc-9d98bd3f115d"
+
+[[durable_objects.bindings]]
+name = "LOBBY_DO"
+class_name = "DurableObjectLobby"
+
+[[migrations]]
+tag = "v1"
+new_classes = ["DurableObjectLobby"]

packages/shared/package.json

@@ -7,11 +7,12 @@
         "lint": "eslint \"src/**/*.{tsx,ts}\""
     },
     "dependencies": {
-        "typescript": "latest"
+        "typescript": "latest",
+        "zod": "^3.23.8"
     },
     "devDependencies": {
         "eslint-config-custom": "*",
-        "tsconfig": "*",
-        "runtime": "*"
+        "runtime": "*",
+        "tsconfig": "*"
     }
 }

packages/shared/src/websocket-api/lobby-api.ts

@@ -0,0 +1,49 @@
+import { z } from "zod"
+
+// limit of amount of positions to be received from the client
+export const UPDATE_POSITIONS_EVERY_MS = 500
+export const UPDATE_POSITIONS_COUNT = Math.floor(60 * (UPDATE_POSITIONS_EVERY_MS / 1000))
+
+const positions = z.array(
+    z.object({
+        x: z.number(),
+        y: z.number(),
+    }),
+)
+
+const positionsPacket = z.object({
+    username: z.string(),
+    positions,
+})
+
+export type PositionsPacket = z.infer<typeof positionsPacket>
+
+const user = z.object({
+    username: z.string(),
+})
+
+export const updateFromClient = z
+    .object({
+        type: z.literal("clientUpdate"),
+        positions,
+    })
+    .strict()
+
+export type UpdateFromClient = z.infer<typeof updateFromClient>
+
+export const messageFromClient = updateFromClient
+export type MessageFromClient = z.infer<typeof messageFromClient>
+
+export const updateFromServer = z
+    .object({
+        type: z.literal("serverUpdate"),
+        positionPackets: z.array(positionsPacket),
+        usersConnected: z.array(user),
+        usersDisconnected: z.array(user),
+    })
+    .strict()
+
+export type UpdateFromServer = z.infer<typeof updateFromServer>
+
+export const messageFromServer = updateFromServer
+export type MessageFromServer = z.infer<typeof messageFromServer>

packages/shared/tsconfig.json

@@ -1,4 +1,7 @@
 {
     "extends": "tsconfig/base.json",
-    "include": ["src"]
+    "include": ["src"],
+    "compilerOptions": {
+        "strict": true
+    }
 }

packages/web-game/package.json

@@ -25,6 +25,7 @@
         "lil-gui": "^0.19.2",
         "poly-decomp-es": "^0.4.2",
         "sat": "^0.9.0",
+        "shared": "*",
         "three": "^0.162.0",
         "tsconfig": "*",
         "vite-plugin-top-level-await": "^1.4.1",

packages/web-game/src/game/modules/module-input/keyboard.ts

@@ -1,6 +1,8 @@
 export class Keyboard {
     private keyboardLeft = false
     private keyboardRight = false
+    private keyboardUpA = false
+    private keyboardUpD = false
 
     private keyboardUpArrow = false
     private keyboardUpW = false
@@ -37,11 +39,11 @@ export class Keyboard {
     }
 
     onPreFixedUpdate(delta: number) {
-        if (this.keyboardLeft) {
+        if (this.keyboardLeft || this.keyboardUpA) {
             this._rotation += delta * 0.001 * this.rotationSpeed
         }
 
-        if (this.keyboardRight) {
+        if (this.keyboardRight || this.keyboardUpD) {
             this._rotation -= delta * 0.001 * this.rotationSpeed
         }
 
@@ -62,6 +64,12 @@ export class Keyboard {
             case "w":
                 this.keyboardUpW = true
                 break
+            case "a":
+                this.keyboardUpA = true
+                break
+            case "d":
+                this.keyboardUpD = true
+                break
             case " ":
                 this.keyboardUpSpace = true
                 break
@@ -82,6 +90,12 @@ export class Keyboard {
             case "w":
                 this.keyboardUpW = false
                 break
+            case "a":
+                this.keyboardUpA = false
+                break
+            case "d":
+                this.keyboardUpD = false
+                break
             case " ":
                 this.keyboardUpSpace = false
                 break