Something went wrong. Try again.
A Social Deduction / Deception Game
Something went wrong. Try again.
4.8 kB · 131 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132// Jetstream consumer (PLAN §10 discovery). We subscribe to the firehose filtered to// town.deception.game records; when one is created, we stand up a room for it. The host's own// create goes through an OPTIMISTIC local path (see POST /api/rooms) so play starts without waiting// for the firehose round-trip — so here we dedupe by the record's AT-URI and skip rooms we already// have. This also lets games declared out-of-band (any client writing the lexicon) become playable.
import WebSocket from 'ws'import { validateDeclaration } from '../../engine/lobby.js'import { MIN_PLAYERS_EFFECTIVE } from './config.js'import type { GameDeclaration } from '../../engine/lobby.js'import type { Registry, ResolvedWorld } from './rooms.js'import { resolveGameWorld } from './maps.js'
const DEFAULT_ENDPOINT = 'wss://jetstream2.us-east.bsky.network/subscribe'const GAME_NSID = 'town.deception.game'
interface JetstreamEvent { did: string kind: string commit?: { operation: 'create' | 'update' | 'delete' collection: string rkey: string record?: Record<string, unknown> }}
/** Parse a town.deception.game record into a GameDeclaration (loose; validated by caller). */function toDeclaration(record: Record<string, unknown>): GameDeclaration | null { const players = Array.isArray(record.players) ? record.players.filter((p): p is string => typeof p === 'string') : null if (!players) return null return { players, roundDurationSec: typeof record.roundDurationSec === 'number' ? record.roundDurationSec : 600, resourceCount: typeof record.resourceCount === 'number' ? record.resourceCount : 4, code: typeof record.code === 'string' ? record.code : undefined, // A game points at its world.tilemap.map directly (station markers carry resource AT-URIs). mapUri: typeof record.map === 'string' ? record.map : undefined, createdAt: typeof record.createdAt === 'string' ? record.createdAt : undefined, }}
export function startJetstream(registry: Registry): { stop: () => void } { const endpoint = process.env.JETSTREAM_URL ?? DEFAULT_ENDPOINT let ws: WebSocket | null = null let stopped = false let backoff = 1000
function connect() { if (stopped) return const url = `${endpoint}?wantedCollections=${encodeURIComponent(GAME_NSID)}` ws = new WebSocket(url)
ws.on('open', () => { backoff = 1000 console.log(`[jetstream] connected, watching ${GAME_NSID}`) })
ws.on('message', (data: WebSocket.RawData) => { let evt: JetstreamEvent try { evt = JSON.parse(data.toString()) } catch { return } const c = evt.commit if (evt.kind !== 'commit' || !c || c.collection !== GAME_NSID) return
const uri = `at://${evt.did}/${c.collection}/${c.rkey}`
// The game declaration was deleted — tear down its room (if we're hosting one) so a cancelled // game stops consuming memory and can't be (re)joined. if (c.operation === 'delete') { if (registry.getByUri(uri)) { registry.removeByUri(uri) console.log(`[jetstream] removed room for deleted game ${uri}`) } return }
if (c.operation !== 'create' || !c.record) return if (registry.getByUri(uri)) return // already created (optimistic path or a duplicate event)
const decl = toDeclaration(c.record) if (!decl) return // the roster's first DID is the host; the record's repo owner should be in the roster if (!decl.players.includes(evt.did)) return const check = validateDeclaration(decl, MIN_PLAYERS_EFFECTIVE) if (!check.ok) return
// Resolve the map, then create — re-checking the URI in case the optimistic path landed first. // A declared-but-unresolvable map means the game is unplayable, so we skip it rather than // falling back to a default layout. void (async () => { let world: ResolvedWorld | undefined if (decl.mapUri) { const res = await resolveGameWorld(decl.mapUri) if (!res.ok) { console.warn(`[jetstream] skipping ${uri}: ${res.reason}`) return } world = res.world } if (registry.getByUri(uri)) return const room = registry.create(decl, uri, world) console.log(`[jetstream] created room ${room.code} from ${uri}`) })() })
ws.on('close', () => { if (stopped) return setTimeout(connect, backoff) backoff = Math.min(backoff * 2, 30_000) })
ws.on('error', (err) => { console.warn('[jetstream] socket error:', err instanceof Error ? err.message : err) ws?.close() }) }
connect() return { stop() { stopped = true ws?.close() }, }}