initial commit

apps/server/migrations/001_initial.sql

@@ -0,0 +1,25 @@
+CREATE TABLE IF NOT EXISTS viewer_accounts (
+  scope TEXT NOT NULL, room_id TEXT NOT NULL, uid TEXT NOT NULL, display_name TEXT NOT NULL,
+  points INTEGER NOT NULL DEFAULT 0 CHECK(points >= 0), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  PRIMARY KEY(scope, room_id, uid)
+);
+CREATE TABLE IF NOT EXISTS point_ledger (
+  id UUID PRIMARY KEY, scope TEXT NOT NULL, room_id TEXT NOT NULL, uid TEXT NOT NULL, delta INTEGER NOT NULL,
+  reason TEXT NOT NULL, source_event_id TEXT UNIQUE, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE TABLE IF NOT EXISTS wheel_spins (
+  id UUID PRIMARY KEY, scope TEXT NOT NULL, room_id TEXT NOT NULL, uid TEXT NOT NULL, category TEXT NOT NULL,
+  song_id INTEGER NOT NULL, song_title TEXT NOT NULL, fallback BOOLEAN NOT NULL, cost INTEGER NOT NULL, balance_after INTEGER NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS viewer_accounts_name_idx ON viewer_accounts(scope, room_id, display_name);
+CREATE INDEX IF NOT EXISTS point_ledger_recent_idx ON point_ledger(scope, room_id, created_at DESC);
+CREATE TABLE IF NOT EXISTS live_session_outbox (
+  id BIGSERIAL PRIMARY KEY,
+  streamer_uid BIGINT NOT NULL,
+  event_ts_ms BIGINT NOT NULL,
+  payload BYTEA NOT NULL,
+  retry_count INTEGER NOT NULL DEFAULT 0,
+  next_retry_at_ms BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS live_session_outbox_due_idx ON live_session_outbox(next_retry_at_ms, id);

apps/server/package.json

@@ -0,0 +1,28 @@
+{
+  "name": "@lxc/server",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "build": "tsc -p tsconfig.json",
+    "check": "tsc -p tsconfig.json --noEmit",
+    "dev": "tsx watch src/index.ts",
+    "test": "node --test test/*.test.mjs"
+  },
+  "dependencies": {
+    "@fastify/cookie": "^11.0.2",
+    "@fastify/static": "^8.1.1",
+    "@fastify/websocket": "^11.0.0",
+    "@laplace.live/ws": "^8.0.1",
+    "@lxc/protocol": "*",
+    "fastify": "^5.2.1",
+    "pg": "^8.13.3",
+    "zod": "^3.24.2"
+  },
+  "devDependencies": {
+    "@types/node": "^22.13.10",
+    "@types/pg": "^8.11.11",
+    "tsx": "^4.19.3",
+    "typescript": "^5.8.3",
+    "vitest": "^3.1.1"
+  }
+}

apps/server/src/config.ts

@@ -0,0 +1,9 @@
+import { z } from "zod";
+const bool = z.enum(["true", "false"]).default("false").transform(v => v === "true");
+const schema = z.object({
+  NODE_ENV: z.enum(["development", "test", "production"]).default("development"), PORT: z.coerce.number().int().positive().default(3000), ROOM_ID: z.string().min(1),
+  DATABASE_URL: z.string().url(), SONGLIST_DATABASE_URL: z.string().url(), COOKIECLOUD_HOST: z.string().url(), COOKIECLOUD_KEY: z.string().min(1), COOKIECLOUD_PASSWORD: z.string().min(1),
+  ADMIN_PASSWORD: z.string().min(12), SESSION_SECRET: z.string().min(32), OBS_ACCESS_TOKEN: z.string().min(16), BILI_REPLY_ENABLED: bool
+});
+export type Config = z.infer<typeof schema>;
+export const loadConfig = (source = process.env): Config => schema.parse(source);

apps/server/src/db.ts

@@ -0,0 +1,44 @@
+import { readFile, readdir } from "node:fs/promises";
+import { join } from "node:path";
+import { Pool, type PoolClient } from "pg";
+type Queryable = Pick<Pool, "query">;
+
+export interface ViewerRow { uid: string; displayName: string; points: number; }
+export interface LedgerRow { id: string; uid: string; displayName: string; delta: number; reason: string; createdAt: string; }
+interface LiveSessionOutboxInsert { streamerUid: number; eventTsMs: number; payload: Uint8Array; }
+interface LiveSessionOutboxItem extends LiveSessionOutboxInsert { id: number; retryCount: number; nextRetryAtMs: number; }
+interface LiveSessionOutboxUpdate { id: number; retryCount: number; nextRetryAtMs: number; }
+export interface LiveSessionOutboxStore { append(items: LiveSessionOutboxInsert[]): Promise<number>; listDue(options: { nowMs: number; limit?: number }): Promise<LiveSessionOutboxItem[]>; ack(ids: number[]): Promise<number>; reschedule(updates: LiveSessionOutboxUpdate[]): Promise<number>; countPending(): Promise<number>; }
+export class WheelDatabase {
+  readonly pool: Pool;
+  constructor(url: string) { this.pool = new Pool({ connectionString: url }); }
+  async migrate(directory: string) { for (const file of (await readdir(directory)).filter(x => x.endsWith(".sql")).sort()) await this.pool.query(await readFile(join(directory, file), "utf8")); }
+  async close() { await this.pool.end(); }
+  async withTransaction<T>(work: (client: PoolClient) => Promise<T>) { const client = await this.pool.connect(); try { await client.query("BEGIN"); const result = await work(client); await client.query("COMMIT"); return result; } catch (error) { await client.query("ROLLBACK"); throw error; } finally { client.release(); } }
+  async ensureViewer(scope: string, roomId: string, viewer: ViewerRow, client: Queryable = this.pool) {
+    await client.query("INSERT INTO viewer_accounts(scope,room_id,uid,display_name) VALUES($1,$2,$3,$4) ON CONFLICT(scope,room_id,uid) DO UPDATE SET display_name=EXCLUDED.display_name,updated_at=now()", [scope, roomId, viewer.uid, viewer.displayName]);
+  }
+  async getViewer(scope: string, roomId: string, uid: string, client: Queryable = this.pool): Promise<ViewerRow | undefined> { const { rows } = await client.query("SELECT uid, display_name AS \"displayName\", points FROM viewer_accounts WHERE scope=$1 AND room_id=$2 AND uid=$3", [scope, roomId, uid]); return rows[0]; }
+  async listViewers(scope: string, roomId: string, search = ""): Promise<ViewerRow[]> { const { rows } = await this.pool.query("SELECT uid,display_name AS \"displayName\",points FROM viewer_accounts WHERE scope=$1 AND room_id=$2 AND (uid ILIKE $3 OR display_name ILIKE $3) ORDER BY updated_at DESC LIMIT 100", [scope, roomId, `%${search}%`]); return rows; }
+  async listLedger(scope: string, roomId: string): Promise<LedgerRow[]> { const { rows } = await this.pool.query("SELECT l.id,l.uid,a.display_name AS \"displayName\",l.delta,l.reason,l.created_at AS \"createdAt\" FROM point_ledger l JOIN viewer_accounts a ON(a.scope=l.scope AND a.room_id=l.room_id AND a.uid=l.uid) WHERE l.scope=$1 AND l.room_id=$2 ORDER BY l.created_at DESC LIMIT 100", [scope, roomId]); return rows; }
+  createLiveSessionOutbox(): LiveSessionOutboxStore {
+    const prune = () => this.pool.query("DELETE FROM live_session_outbox WHERE event_ts_ms < $1", [Date.now() - 7 * 24 * 60 * 60 * 1000]);
+    return {
+      append: async items => { if (!items.length) return 0; await prune(); const values: unknown[] = []; const rows = items.map((item, index) => { const offset = index * 4; values.push(Math.floor(item.streamerUid), Math.floor(item.eventTsMs), Buffer.from(item.payload), Math.floor(item.eventTsMs)); return `($${offset + 1},$${offset + 2},$${offset + 3},0,$${offset + 4})`; }).join(","); const result = await this.pool.query(`INSERT INTO live_session_outbox(streamer_uid,event_ts_ms,payload,retry_count,next_retry_at_ms) VALUES ${rows}`, values); return result.rowCount ?? 0; },
+      listDue: async ({ nowMs, limit = 100 }) => { await prune(); const result = await this.pool.query<{ id: string; streamer_uid: string; event_ts_ms: string; payload: Buffer; retry_count: number; next_retry_at_ms: string }>("SELECT id,streamer_uid,event_ts_ms,payload,retry_count,next_retry_at_ms FROM live_session_outbox WHERE next_retry_at_ms <= $1 ORDER BY next_retry_at_ms,id LIMIT $2", [Math.floor(nowMs), Math.min(500, Math.max(1, Math.floor(limit)))]); return result.rows.map(row => ({ id: Number(row.id), streamerUid: Number(row.streamer_uid), eventTsMs: Number(row.event_ts_ms), payload: new Uint8Array(row.payload), retryCount: row.retry_count, nextRetryAtMs: Number(row.next_retry_at_ms) })); },
+      ack: async ids => { if (!ids.length) return 0; const result = await this.pool.query("DELETE FROM live_session_outbox WHERE id = ANY($1::bigint[])", [ids]); return result.rowCount ?? 0; },
+      reschedule: async updates => { if (!updates.length) return 0; const ids = updates.map(item => item.id); const retries = updates.map(item => item.retryCount); const due = updates.map(item => item.nextRetryAtMs); const result = await this.pool.query("UPDATE live_session_outbox AS o SET retry_count = u.retry_count, next_retry_at_ms = u.next_retry_at_ms FROM unnest($1::bigint[],$2::integer[],$3::bigint[]) AS u(id,retry_count,next_retry_at_ms) WHERE o.id=u.id", [ids, retries, due]); return result.rowCount ?? 0; },
+      countPending: async () => { await prune(); const result = await this.pool.query<{ count: string }>("SELECT COUNT(*)::text AS count FROM live_session_outbox"); return Number(result.rows[0]?.count ?? 0); }
+    };
+  }
+}
+export interface Song { id: number; title: string; tags: string[]; }
+export class SongCatalog {
+  readonly pool: Pool; constructor(url: string) { this.pool = new Pool({ connectionString: url }); }
+  async close() { await this.pool.end(); }
+  async random(category: string): Promise<{ song: Song; fallback: boolean } | undefined> {
+    const categoryKey = category.trim().replace(/\s+/g, " ").toLowerCase();
+    const select = async (matching: boolean) => (await this.pool.query<Song>(`SELECT s."Id" id,s."Title" title,COALESCE(array_agg(t."Name") FILTER (WHERE t."Name" IS NOT NULL), '{}') tags FROM "Songs" s LEFT JOIN "SongTags" st ON st."SongId"=s."Id" LEFT JOIN "Tags" t ON t."Id"=st."TagId" WHERE s."IsHidden"=false ${matching ? 'AND EXISTS (SELECT 1 FROM "SongTags" mst JOIN "Tags" mt ON mt."Id"=mst."TagId" WHERE mst."SongId"=s."Id" AND lower(mt."NormalizedName") LIKE $1)' : ""} GROUP BY s."Id" ORDER BY random() LIMIT 1`, matching ? [`%${categoryKey}%`] : [])).rows[0];
+    const match = await select(true); if (match) return { song: match, fallback: false }; const fallback = await select(false); return fallback && { song: fallback, fallback: true };
+  }
+}

apps/server/src/hub.ts

@@ -0,0 +1,6 @@
+import type { LiveEvent } from "@lxc/protocol";
+export class EventHub { private listeners = new Set<(event: LiveEvent) => void>();
+  publish(event: LiveEvent) { this.listeners.forEach(listener => listener(event)); }
+  subscribe(listener: (event: LiveEvent) => void) { this.listeners.add(listener); return () => this.listeners.delete(listener); }
+  get subscriberCount() { return this.listeners.size; }
+}

apps/server/src/index.ts

@@ -0,0 +1,46 @@
+import { createHmac, timingSafeEqual } from "node:crypto";
+import { existsSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+import { dirname, join } from "node:path";
+import Fastify from "fastify";
+import cookie from "@fastify/cookie";
+import statik from "@fastify/static";
+import websocket from "@fastify/websocket";
+import { z } from "zod";
+import { loadConfig } from "./config.js";
+import { WheelDatabase, SongCatalog } from "./db.js";
+import { EventHub } from "./hub.js";
+import { BilibiliLiveSource, type IncomingLiveMessage } from "./live.js";
+import { BilibiliReply, DisabledReply, SwitchableReply } from "./reply.js";
+import { WheelService } from "./wheel.js";
+
+const here = dirname(fileURLToPath(import.meta.url));
+const equal = (actual: string, expected: string) => { const a = Buffer.from(actual); const b = Buffer.from(expected); return a.length === b.length && timingSafeEqual(a, b); };
+
+export async function buildApp() {
+  const config = loadConfig(); const app = Fastify({ logger: true }); const database = new WheelDatabase(config.DATABASE_URL); const songs = new SongCatalog(config.SONGLIST_DATABASE_URL); const hub = new EventHub();
+  const cookieCloudReplyCookie = async () => { const host = config.COOKIECLOUD_HOST.replace(/\/+$/, ""); const response = await fetch(`${host}/get/${encodeURIComponent(config.COOKIECLOUD_KEY)}`, { method: "POST", headers: { "content-type": "application/x-www-form-urlencoded" }, body: new URLSearchParams({ password: config.COOKIECLOUD_PASSWORD }) }); if (!response.ok) throw new Error(`CookieCloud download failed: HTTP ${response.status}`); const payload = await response.json() as { cookie_data?: Record<string, Record<string, { name: string; value: string; domain: string }>> }; const cookies = Object.values(payload.cookie_data ?? {}).flatMap(domain => Object.values(domain)).filter(cookie => cookie.domain.includes("bilibili.com")); if (!cookies.length) throw new Error("CookieCloud has no bilibili.com cookies"); return cookies.map(cookie => `${cookie.name}=${cookie.value}`).join("; "); };
+  const rawReply = config.BILI_REPLY_ENABLED ? new BilibiliReply(config.ROOM_ID, cookieCloudReplyCookie) : new DisabledReply("Reply disabled by configuration"); const reply = new SwitchableReply(rawReply, config.BILI_REPLY_ENABLED);
+  const service = new WheelService(database, songs, hub, reply, config.ROOM_ID); const source = new BilibiliLiveSource(config);
+  const sessionFor = () => createHmac("sha256", config.SESSION_SECRET).update("admin").digest("base64url"); const admin = (request: any) => equal(request.cookies?.lxc_session ?? "", sessionFor());
+  await app.register(cookie); await app.register(websocket);
+  const webRoot = join(here, "../../web/dist"); if (existsSync(webRoot)) await app.register(statik, { root: webRoot, prefix: "/" });
+  app.get("/health", async () => ({ ok: true, roomId: config.ROOM_ID, subscribers: hub.subscriberCount }));
+  app.post("/api/auth/login", async (request, replyTo) => { const body = z.object({ password: z.string() }).parse(request.body); if (!equal(body.password, config.ADMIN_PASSWORD)) return replyTo.code(401).send({ error: "Invalid password" }); replyTo.setCookie("lxc_session", sessionFor(), { httpOnly: true, secure: "auto", sameSite: "lax", path: "/", maxAge: 60 * 60 * 12 }); return { ok: true }; });
+  app.post("/api/auth/logout", async (_request, replyTo) => { replyTo.clearCookie("lxc_session", { path: "/" }); return { ok: true }; });
+  const guard = async (request: any, replyTo: any) => { if (!admin(request)) return replyTo.code(401).send({ error: "Administrator authentication required" }); };
+  app.get("/api/admin/status", { preHandler: guard }, async () => ({ roomId: config.ROOM_ID, source: source.status(), reply: reply.status(), websocketClients: hub.subscriberCount, cookieCloudHost: config.COOKIECLOUD_HOST }));
+  app.get("/api/admin/viewers", { preHandler: guard }, async request => database.listViewers("live", config.ROOM_ID, String((request.query as any).search ?? "")));
+  app.get("/api/admin/ledger", { preHandler: guard }, async () => database.listLedger("live", config.ROOM_ID));
+  app.post("/api/admin/reconnect", { preHandler: guard }, async () => { await source.stop(); await source.start(); return { ok: true, source: source.status() }; });
+  app.post("/api/admin/reply", { preHandler: guard }, async request => { reply.enabled = z.object({ enabled: z.boolean() }).parse(request.body).enabled; return reply.status(); });
+  app.get("/api/admin/obs-url", { preHandler: guard }, async request => `${request.protocol}://${request.hostname}/obs?token=${encodeURIComponent(config.OBS_ACCESS_TOKEN)}`);
+  app.post("/api/test/event", { preHandler: guard }, async request => { const data = z.discriminatedUnion("kind", [z.object({ kind: z.literal("enter"), uid: z.string(), name: z.string() }), z.object({ kind: z.literal("gift"), uid: z.string(), name: z.string(), giftName: z.string().default("测试礼物"), battery: z.number().int().nonnegative(), quantity: z.number().int().positive().default(1) }), z.object({ kind: z.literal("danmaku"), uid: z.string(), name: z.string(), text: z.string() })]).parse(request.body); const viewer = { uid: data.uid, name: data.name }; const message: IncomingLiveMessage = data.kind === "gift" ? { ...data, viewer, sourceEventId: `test-${crypto.randomUUID()}` } : data.kind === "enter" ? { kind: "enter", viewer } : { kind: "danmaku", viewer, text: data.text }; await service.handle(message, "test"); return { ok: true }; });
+  app.get("/ws", { websocket: true }, (socket, request) => { const token = String((request.query as any)?.token ?? ""); if (!(admin(request) || equal(token, config.OBS_ACCESS_TOKEN))) { socket.close(1008, "Unauthorized"); return; } const unsubscribe = hub.subscribe(event => { if (socket.readyState === socket.OPEN) socket.send(JSON.stringify(event)); }); socket.on("close", unsubscribe); });
+  const webPage = async (_request: unknown, replyTo: any) => { if (!existsSync(join(webRoot, "index.html"))) return replyTo.code(503).send("Web frontend has not been built"); return replyTo.sendFile("index.html"); };
+  app.get("/admin", webPage); app.get("/test", webPage); app.get("/obs", webPage);
+  app.addHook("onClose", async () => { await source.stop(); await database.close(); await songs.close(); });
+  await database.migrate(join(here, "../migrations")); source.onMessage(message => void service.handle(message)); await source.start(); return app;
+}
+
+if (process.argv[1] === fileURLToPath(import.meta.url)) { const app = await buildApp(); const config = loadConfig(); await app.listen({ host: "0.0.0.0", port: config.PORT }); }

apps/server/src/live.ts

@@ -0,0 +1,55 @@
+import { EventEmitter } from "node:events";
+import { KeepLiveWS } from "@laplace.live/ws";
+import { makeEvent, type Viewer } from "@lxc/protocol";
+import type { Config } from "./config.js";
+import type { EventHub } from "./hub.js";
+
+export interface IncomingEnter { kind: "enter"; viewer: Viewer; }
+export interface IncomingGift { kind: "gift"; viewer: Viewer; giftName: string; battery: number; quantity: number; sourceEventId: string; }
+export interface IncomingDanmaku { kind: "danmaku"; viewer: Viewer; text: string; }
+export type IncomingLiveMessage = IncomingEnter | IncomingGift | IncomingDanmaku;
+export interface LiveSource { start(): Promise<void>; stop(): Promise<void>; onMessage(handler: (message: IncomingLiveMessage) => void): () => void; status(): { connected: boolean; cookieCloud: boolean; detail?: string }; }
+
+/** Maps the fields used by common Bilibili CMD payloads without leaking raw data into business code. */
+export function normalizeCoreMessage(raw: any): IncomingLiveMessage | undefined {
+  const cmd = String(raw?.cmd ?? "").split(":")[0];
+  if (cmd === "DANMU_MSG") { const info = raw.info ?? []; const user = info[2] ?? []; return { kind: "danmaku", viewer: { uid: String(user[0] ?? ""), name: String(user[1] ?? "匿名观众") }, text: String(info[1] ?? "") }; }
+  if (cmd === "SEND_GIFT") { const data = raw.data ?? raw; return { kind: "gift", viewer: { uid: String(data.uid ?? ""), name: String(data.uname ?? "匿名观众") }, giftName: String(data.giftName ?? "礼物"), battery: Number(data.price ?? 0), quantity: Number(data.num ?? 1), sourceEventId: String(data.tid ?? `${data.uid}-${data.timestamp}-${data.giftId}-${data.num}`) }; }
+  if (cmd === "INTERACT_WORD") { const data = raw.data ?? raw; return { kind: "enter", viewer: { uid: String(data.uid ?? ""), name: String(data.uname ?? "匿名观众") } }; }
+  return undefined;
+}
+
+interface CookieCloudCookie { name: string; value: string; domain: string; }
+interface CookieCloudPayload { cookie_data?: Record<string, Record<string, CookieCloudCookie>>; }
+interface DanmakuInfo { data?: { token?: string; host_list?: Array<{ host?: string; wss_port?: number }>; host_server_list?: Array<{ host?: string; wss_port?: number }> }; message?: string; code?: number; }
+export class BilibiliLiveSource implements LiveSource {
+  private emitter = new EventEmitter(); private client?: KeepLiveWS; private connected = false; private detail?: string; private cookieReady = false;
+  constructor(private readonly config: Config) {}
+  onMessage(handler: (message: IncomingLiveMessage) => void) { this.emitter.on("message", handler); return () => this.emitter.off("message", handler); }
+  status() { return { connected: this.connected, cookieCloud: this.cookieReady, detail: this.detail }; }
+  async start() {
+    try {
+      await this.cookieHeader(); const connection = await this.connectionInfo(); this.client = new KeepLiveWS(Number(this.config.ROOM_ID), { key: connection.key, address: connection.address });
+      // `@laplace.live/ws` dispatches the complete upstream packet on `msg`.
+      // Concrete event names preserve Bilibili suffixes (for example
+      // `DANMU_MSG:4:0`), so listening to bare command names loses messages.
+      this.client.addEventListener("msg", (event: any) => {
+        const raw = event.data?.msg ?? event.data;
+        const message = normalizeCoreMessage(raw);
+        if (message?.viewer.uid) this.emitter.emit("message", message);
+      });
+      this.client.addEventListener("heartbeat", () => { this.connected = true; this.detail = "Connected; waiting for live messages"; }); this.client.addEventListener("live", () => { this.connected = true; this.detail = "Connected; waiting for live messages"; }); this.client.addEventListener("close", () => { this.connected = false; }); this.client.addEventListener("e", () => { this.connected = false; this.detail = "Bilibili WebSocket disconnected; reconnecting"; });
+    } catch (error) { this.detail = error instanceof Error ? error.message : String(error); this.connected = false; }
+  }
+  async stop() { this.client?.close(); this.client = undefined; this.connected = false; }
+  private async cookieHeader() { const host = this.config.COOKIECLOUD_HOST.replace(/\/+$/, ""); const response = await fetch(`${host}/get/${encodeURIComponent(this.config.COOKIECLOUD_KEY)}`, { method: "POST", headers: { "content-type": "application/x-www-form-urlencoded" }, body: new URLSearchParams({ password: this.config.COOKIECLOUD_PASSWORD }) }); if (!response.ok) throw new Error(`CookieCloud download failed: HTTP ${response.status}`); const payload = await response.json() as CookieCloudPayload; const cookies = Object.values(payload.cookie_data ?? {}).flatMap(domain => Object.values(domain)).filter(cookie => cookie.domain.includes("bilibili.com")); if (!cookies.length) throw new Error("CookieCloud has no bilibili.com cookies"); this.cookieReady = true; }
+  private async connectionInfo() { const roomId = encodeURIComponent(this.config.ROOM_ID); const response = await fetch(`https://api.live.bilibili.com/room/v1/Danmu/getConf?room_id=${roomId}&platform=pc&player=web`, { headers: { accept: "application/json, text/plain, */*", referer: `https://live.bilibili.com/${roomId}`, "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" } }); const json = await response.json() as DanmakuInfo; const key = json.data?.token; const server = json.data?.host_server_list?.[0] ?? json.data?.host_list?.[0]; if (!response.ok || !key || !server?.host) throw new Error(`Bilibili Danmu getConf failed: ${json.message ?? "missing token"} (code ${json.code ?? response.status})`); return { key, address: `wss://${server.host}:${server.wss_port ?? 443}/sub` }; }
+}
+
+export class ManualLiveSource implements LiveSource {
+  private emitter = new EventEmitter(); async start() {} async stop() {} status() { return { connected: true, cookieCloud: true, detail: "manual test source" }; }
+  onMessage(handler: (message: IncomingLiveMessage) => void) { this.emitter.on("message", handler); return () => this.emitter.off("message", handler); }
+  emit(message: IncomingLiveMessage) { this.emitter.emit("message", message); }
+}
+
+export function publishUnknown(hub: EventHub, roomId: string, cmd: string, raw: unknown) { hub.publish(makeEvent(roomId, "live.unknown", { cmd, raw })); }

apps/server/src/reply.ts

@@ -0,0 +1,17 @@
+export interface ReplyStatus { enabled: boolean; available: boolean; detail?: string; }
+export interface ReplyPort { status(): ReplyStatus; send(text: string): Promise<void>; }
+export class DisabledReply implements ReplyPort { constructor(private readonly detail: string) {} status() { return { enabled: false, available: false, detail: this.detail }; } async send() {} }
+export class SwitchableReply implements ReplyPort {
+  enabled: boolean;
+  constructor(private readonly delegate: ReplyPort, enabled: boolean) { this.enabled = enabled; }
+  status() { const status = this.delegate.status(); return this.enabled ? status : { ...status, enabled: false, detail: "Disabled by administrator" }; }
+  async send(text: string) { if (this.enabled) await this.delegate.send(text); }
+}
+/** Bilibili sender with a queue. COOKIE must include bili_jct; it is intentionally isolated from scoring. */
+export class BilibiliReply implements ReplyPort {
+  private next = Promise.resolve(); private lastSent = 0;
+  private lastError?: string;
+  constructor(private readonly roomId: string, private readonly cookieProvider: () => Promise<string>) {}
+  status() { return { enabled: true, available: !this.lastError, detail: this.lastError ?? "CookieCloud-backed reply ready" }; }
+  send(text: string) { this.next = this.next.then(async () => { const cookie = await this.cookieProvider(); const csrf = cookie.match(/(?:^|;\s*)bili_jct=([^;]+)/)?.[1]; if (!csrf) throw new Error("CookieCloud Bilibili Cookie is missing bili_jct"); const wait = Math.max(0, 3000 - (Date.now() - this.lastSent)); if (wait) await new Promise(r => setTimeout(r, wait)); const body = new URLSearchParams({ roomid: this.roomId, msg: text.slice(0, 100), csrf, csrf_token: csrf }); const response = await fetch("https://api.live.bilibili.com/msg/send", { method: "POST", headers: { cookie, "content-type": "application/x-www-form-urlencoded", referer: `https://live.bilibili.com/${this.roomId}` }, body }); if (!response.ok) throw new Error(`Bilibili reply failed: ${response.status}`); this.lastError = undefined; this.lastSent = Date.now(); }).catch(error => { this.lastError = error instanceof Error ? error.message : String(error); }); return this.next; }
+}

apps/server/src/wheel.test.ts

@@ -0,0 +1,3 @@
+import { describe, expect, it, vi } from "vitest";
+import { WHEEL_COST } from "./wheel.js";
+describe("wheel rules", () => { it("uses the agreed fixed cost", () => expect(WHEEL_COST).toBe(150)); it("normalizes spaces", () => expect("转盘   摇滚".trim().replace(/\s+/g, " ")).toBe("转盘 摇滚")); });

apps/server/src/wheel.ts

@@ -0,0 +1,45 @@
+import { randomUUID } from "node:crypto";
+import { makeEvent, type Viewer } from "@lxc/protocol";
+import type { WheelDatabase, SongCatalog, ViewerRow } from "./db.js";
+import type { EventHub } from "./hub.js";
+import type { ReplyPort } from "./reply.js";
+import type { IncomingLiveMessage } from "./live.js";
+
+export const WHEEL_COST = 150;
+const normal = (text: string) => text.trim().replace(/\s+/g, " ");
+const toViewerRow = (viewer: Viewer): ViewerRow => ({ uid: viewer.uid, displayName: viewer.name, points: 0 });
+
+export class WheelService {
+  constructor(private readonly database: WheelDatabase, private readonly songs: SongCatalog, private readonly hub: EventHub, private readonly reply: ReplyPort, private readonly roomId: string) {}
+  async handle(message: IncomingLiveMessage, scope = "live") {
+    const viewer = message.viewer; await this.database.ensureViewer(scope, this.roomId, toViewerRow(viewer));
+    if (message.kind === "enter") { this.hub.publish(makeEvent(this.roomId, "live.enter", { viewer })); return; }
+    if (message.kind === "gift") return this.gift(message, scope);
+    this.hub.publish(makeEvent(this.roomId, "live.danmaku", { viewer, text: message.text })); return this.command(viewer, message.text, scope);
+  }
+  private async gift(message: Extract<IncomingLiveMessage, { kind: "gift" }>, scope: string) {
+    const amount = Math.max(0, Math.floor(message.battery)) * Math.max(1, Math.floor(message.quantity));
+    const result = await this.database.withTransaction(async client => {
+      await this.database.ensureViewer(scope, this.roomId, toViewerRow(message.viewer), client);
+      const inserted = await client.query("INSERT INTO point_ledger(id,scope,room_id,uid,delta,reason,source_event_id,metadata) VALUES($1,$2,$3,$4,$5,'gift',$6,$7) ON CONFLICT(source_event_id) DO NOTHING RETURNING id", [randomUUID(), scope, this.roomId, message.viewer.uid, amount, message.sourceEventId, JSON.stringify({ giftName: message.giftName, battery: message.battery, quantity: message.quantity })]);
+      if (!inserted.rowCount) return undefined;
+      const updated = await client.query<{ points: number }>("UPDATE viewer_accounts SET points=points+$1,updated_at=now() WHERE scope=$2 AND room_id=$3 AND uid=$4 RETURNING points", [amount, scope, this.roomId, message.viewer.uid]); return updated.rows[0]?.points;
+    });
+    this.hub.publish(makeEvent(this.roomId, "live.gift", { viewer: message.viewer, giftName: message.giftName, battery: message.battery, quantity: message.quantity, sourceEventId: message.sourceEventId }));
+    if (result !== undefined) this.hub.publish(makeEvent(this.roomId, "viewer.points.updated", { viewer: message.viewer, delta: amount, balance: result, reason: "gift" }));
+  }
+  private async command(viewer: Viewer, raw: string, scope: string) {
+    const text = normal(raw); if (text === "转盘查询") { const account = await this.database.getViewer(scope, this.roomId, viewer.uid); const balance = account?.points ?? 0; const event = makeEvent(this.roomId, "viewer.points.updated", { viewer, delta: 0, balance, reason: "gift" }); this.hub.publish(event); await this.reply.send(`${viewer.name} 当前转盘点数：${balance}`); return; }
+    if (text === "转盘") return this.invalid(viewer, "用法：转盘 [类别]");
+    const match = /^转盘\s+(.+)$/.exec(text); if (!match) return;
+    const category = match[1]!; const choice = await this.songs.random(category); if (!choice) return this.invalid(viewer, "歌单暂时没有可抽取的歌曲");
+    const result = await this.database.withTransaction(async client => {
+      const current = await this.database.getViewer(scope, this.roomId, viewer.uid, client); const balance = current?.points ?? 0; if (balance < WHEEL_COST) return { insufficient: true as const, balance };
+      const update = await client.query<{ points: number }>("UPDATE viewer_accounts SET points=points-$1,updated_at=now() WHERE scope=$2 AND room_id=$3 AND uid=$4 AND points >= $1 RETURNING points", [WHEEL_COST, scope, this.roomId, viewer.uid]); if (!update.rowCount) return { insufficient: true as const, balance: 0 };
+      const balanceAfter = update.rows[0]!.points; await client.query("INSERT INTO point_ledger(id,scope,room_id,uid,delta,reason,metadata) VALUES($1,$2,$3,$4,$5,'wheel',$6)", [randomUUID(), scope, this.roomId, viewer.uid, -WHEEL_COST, JSON.stringify({ category, songId: choice.song.id })]); await client.query("INSERT INTO wheel_spins(id,scope,room_id,uid,category,song_id,song_title,fallback,cost,balance_after) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)", [randomUUID(), scope, this.roomId, viewer.uid, category, choice.song.id, choice.song.title, choice.fallback, WHEEL_COST, balanceAfter]); return { insufficient: false as const, balance: balanceAfter };
+    });
+    if (result.insufficient) { this.hub.publish(makeEvent(this.roomId, "wheel.insufficient-balance", { viewer, balance: result.balance, cost: WHEEL_COST })); await this.reply.send(`${viewer.name} 点数不足（需要 ${WHEEL_COST}，当前 ${result.balance}）`); return; }
+    this.hub.publish(makeEvent(this.roomId, "viewer.points.updated", { viewer, delta: -WHEEL_COST, balance: result.balance, reason: "wheel" })); this.hub.publish(makeEvent(this.roomId, "wheel.result", { viewer, category, fallback: choice.fallback, song: choice.song, cost: WHEEL_COST, balance: result.balance })); await this.reply.send(`${viewer.name} 抽中了《${choice.song.title}》`);
+  }
+  private async invalid(viewer: Viewer, message: string) { this.hub.publish(makeEvent(this.roomId, "wheel.invalid-command", { viewer, message })); await this.reply.send(`${viewer.name}：${message}`); }
+}

apps/server/test/wheel.test.mjs

@@ -0,0 +1,17 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { WHEEL_COST } from "../dist/wheel.js";
+import { normalizeCoreMessage } from "../dist/live.js";
+test("wheel cost is 150", () => assert.equal(WHEEL_COST, 150));
+test("Bilibili gift messages normalize to typed inputs", () => {
+  const event = normalizeCoreMessage({ cmd: "SEND_GIFT", data: { uid: 4, uname: "测试", giftName: "辣条", price: 150, num: 2, tid: "gift-1" } });
+  assert.deepEqual(event, { kind: "gift", viewer: { uid: "4", name: "测试" }, giftName: "辣条", battery: 150, quantity: 2, sourceEventId: "gift-1" });
+});
+test("Bilibili danmaku commands with a suffix normalize from their raw packet", () => {
+  const event = normalizeCoreMessage({ cmd: "DANMU_MSG:4:0", info: [[], "转盘 流行", [42, "观众"]] });
+  assert.deepEqual(event, { kind: "danmaku", viewer: { uid: "42", name: "观众" }, text: "转盘 流行" });
+});
+test("Bilibili enter packets normalize from their raw packet", () => {
+  const event = normalizeCoreMessage({ cmd: "INTERACT_WORD", data: { uid: 7, uname: "进房观众" } });
+  assert.deepEqual(event, { kind: "enter", viewer: { uid: "7", name: "进房观众" } });
+});

apps/server/tsconfig.json

@@ -0,0 +1 @@
+{ "compilerOptions": { "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", "outDir": "dist", "rootDir": "src", "strict": true, "esModuleInterop": true, "skipLibCheck": true }, "include": ["src"] }