TheSunIsLeaking/recflare
3
0
Fork 0
mirror of https://github.com/djdevin/recflare.git synced 2026-09-08 14:41:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

add playersettings, img, more match endpoints

Browse Source
This commit is contained in:
Devin Zuczek
2026-06-12 17:15:18 -04:00
parent 4a97e05866
commit 4d1ea9fe3e
42 changed files with 21397 additions and 2538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/notify/src/notifications-hub.ts
+20 -5
View File
@@ -104,7 +104,10 @@ export class NotificationsHub extends DurableObject<Env> {
if (state) {
// Mirrors OnDisconnected: drop this connection's subscriptions, which
// also removes it from every player's connection set.
this.ctx.storage.sql.exec('DELETE FROM subscriptions WHERE connectionId = ?', state.connectionId)
this.ctx.storage.sql.exec(
'DELETE FROM subscriptions WHERE connectionId = ?',
state.connectionId
)
}
try {
ws.close()
@@ -191,7 +194,9 @@ export class NotificationsHub extends DurableObject<Env> {
// Flush any notifications queued while these players were offline.
for (const playerId of unique) {
const pending = this.ctx.storage.sql
.exec<{ payload: string }>('SELECT payload FROM pending WHERE playerId = ? ORDER BY id', playerId)
.exec<{
payload: string
}>('SELECT payload FROM pending WHERE playerId = ? ORDER BY id', playerId)
.toArray()
if (pending.length === 0) continue
for (const row of pending) {
@@ -241,14 +246,21 @@ export class NotificationsHub extends DurableObject<Env> {
}
if (delivered === 0) {
this.ctx.storage.sql.exec('INSERT INTO pending (playerId, payload) VALUES (?, ?)', playerId, payload)
this.ctx.storage.sql.exec(
'INSERT INTO pending (playerId, payload) VALUES (?, ?)',
playerId,
payload
)
return { delivered: 0, queued: true }
}
return { delivered, queued: false }
}
/** Broadcast a notification to every connected (handshaken) client. */
async broadcast(notificationType: number, data?: Record<string, unknown>): Promise<{ delivered: number }> {
async broadcast(
notificationType: number,
data?: Record<string, unknown>
): Promise<{ delivered: number }> {
const payload = this.buildNotificationPayload(notificationType, data)
let delivered = 0
for (const ws of this.ctx.getWebSockets()) {
@@ -266,7 +278,10 @@ export class NotificationsHub extends DurableObject<Env> {
* Build the `Notification` argument: a JSON string `{ Id, Msg }`, matching the
* C# `SendToConnection` (null values are dropped from `Msg`).
*/
private buildNotificationPayload(notificationType: number, data?: Record<string, unknown>): string {
private buildNotificationPayload(
notificationType: number,
data?: Record<string, unknown>
): string {
const msg: Record<string, unknown> = {}
if (data) {
for (const [key, value] of Object.entries(data)) {
apps/notify/src/notify.app.ts
+2 -1
View File
@@ -3,9 +3,10 @@ import { useWorkersLogger } from 'workers-tagged-logger'
import { logger, withNotFound, withOnError } from '@repo/hono-helpers'
import type { App } from './context'
import { NotificationsHub } from './notifications-hub'
import type { App } from './context'
/**
* Ported from the C# `NotifyController`, which maps a SignalR hub at `/hub/v1`
* (see `NotificationsHub` / `NotificationService`). The hub itself — WebSocket
apps/notify/src/test/integration/api.test.ts
+39 -9
View File
@@ -15,7 +15,9 @@ interface HubRecord {
}
/** Open a hub WebSocket, accept it, and complete the SignalR handshake. */
async function connect(id: string): Promise<{ ws: WebSocket; waitFor: (pred: (r: HubRecord) => boolean) => Promise<HubRecord> }> {
async function connect(
id: string
): Promise<{ ws: WebSocket; waitFor: (pred: (r: HubRecord) => boolean) => Promise<HubRecord> }> {
const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=${id}`, {
headers: { Upgrade: 'websocket' },
})
@@ -26,7 +28,8 @@ async function connect(id: string): Promise<{ ws: WebSocket; waitFor: (pred: (r:
const records: HubRecord[] = []
const waiters: Array<{ pred: (r: HubRecord) => boolean; resolve: (r: HubRecord) => void }> = []
ws.addEventListener('message', (e: MessageEvent) => {
const text = typeof e.data === 'string' ? e.data : new TextDecoder().decode(e.data as ArrayBuffer)
const text =
typeof e.data === 'string' ? e.data : new TextDecoder().decode(e.data as ArrayBuffer)
for (const part of text.split(RS)) {
if (!part) continue
const rec = JSON.parse(part) as HubRecord
@@ -45,7 +48,13 @@ async function connect(id: string): Promise<{ ws: WebSocket; waitFor: (pred: (r:
if (existing) return Promise.resolve(existing)
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error('timed out waiting for hub message')), 2000)
waiters.push({ pred, resolve: (r) => { clearTimeout(timer); resolve(r) } })
waiters.push({
pred,
resolve: (r) => {
clearTimeout(timer)
resolve(r)
},
})
})
}
@@ -67,7 +76,9 @@ const post = (path: string, body: unknown) =>
describe('negotiate', () => {
test('POST /hub/v1/negotiate advertises the WebSocket transport', async () => {
const res = await exports.default.fetch(`${ORIGIN}/hub/v1/negotiate?negotiateVersion=1`, { method: 'POST' })
const res = await exports.default.fetch(`${ORIGIN}/hub/v1/negotiate?negotiateVersion=1`, {
method: 'POST',
})
expect(res.status).toBe(200)
const body = (await res.json()) as {
connectionId: string
@@ -113,7 +124,11 @@ describe('hub protocol', () => {
describe('notification delivery', () => {
test('queues for an offline player and flushes on subscribe', async () => {
const playerId = 9001
const queued = await post('/internal/notify', { playerId, notificationType: 2, data: { messageId: 'm1' } })
const queued = await post('/internal/notify', {
playerId,
notificationType: 2,
data: { messageId: 'm1' },
})
expect(await queued.json()).toMatchObject({ queued: true, delivered: 0 })
const { ws, waitFor } = await connect('conn-pending')
@@ -127,10 +142,19 @@ describe('notification delivery', () => {
test('delivers live to a subscribed player', async () => {
const playerId = 9002
const { ws, waitFor } = await connect('conn-live')
send(ws, { type: 1, invocationId: 's', target: 'SubscribeToPlayers', arguments: [{ playerIds: [playerId] }] })
send(ws, {
type: 1,
invocationId: 's',
target: 'SubscribeToPlayers',
arguments: [{ playerIds: [playerId] }],
})
await waitFor((r) => r.type === 3 && r.invocationId === 's')
const res = await post('/internal/notify', { playerId, notificationType: 1, data: { accountId: 42 } })
const res = await post('/internal/notify', {
playerId,
notificationType: 1,
data: { accountId: 42 },
})
expect(await res.json()).toMatchObject({ delivered: 1, queued: false })
const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
@@ -141,10 +165,16 @@ describe('notification delivery', () => {
test('broadcast reaches connected clients', async () => {
const { ws, waitFor } = await connect('conn-broadcast')
const res = await post('/internal/broadcast', { notificationType: 25, data: { message: 'maint' } })
const res = await post('/internal/broadcast', {
notificationType: 25,
data: { message: 'maint' },
})
expect(((await res.json()) as { delivered: number }).delivered).toBeGreaterThanOrEqual(1)
const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
expect(JSON.parse((note.arguments as string[])[0])).toEqual({ Id: 25, Msg: { message: 'maint' } })
expect(JSON.parse((note.arguments as string[])[0])).toEqual({
Id: 25,
Msg: { message: 'maint' },
})
ws.close()
})
})