[wip] so close; check the ws messages

pkg/server/src/api.ts

@@ -6,6 +6,7 @@ import { combine } from "kefir";
 import Bus from "kefir-bus";
 import db from "./db";
 import { liveTable, WsIn, WsOut } from "./table";
+import { err } from "./logging";
 
 export const WS = Bus<
 	{
@@ -76,13 +77,17 @@ const api = new Elysia({ prefix: "/api" })
 			})
 	)
 	.ws("/ws/:tableKey", {
-		async open({
+		body: WsIn,
+		response: WsOut,
+
+		open: ({
 			data: {
 				params: { tableKey },
 				humanKey,
 			},
 			send,
-		}) {
+		}) => {
+			console.log("websocket opened");
 			const table = liveTable(tableKey);
 
 			table.inputs.connectionChanges.emit({
@@ -90,22 +95,14 @@ const api = new Elysia({ prefix: "/api" })
 				presence: "joined",
 			});
 
-			Object.entries(table.outputs.global).forEach(([type, stream]) =>
+			Object.entries({
+				...table.outputs.global,
+				...(table.outputs.player[humanKey] ?? {}),
+			}).forEach(([type, stream]) =>
 				stream.onValue((v) => send({ [type]: v }))
 			);
-
-			combine(
-				[table.outputs.gameState],
-				[table.outputs.gameImpl],
-				(state, game: Game) => state && game.getView({ state, humanKey })
-			)
-				.toProperty()
-				.onValue((view) => send({ view }));
 		},
 
-		body: WsIn,
-		response: WsOut,
-
 		message: (
 			{
 				data: {
@@ -114,19 +111,20 @@ const api = new Elysia({ prefix: "/api" })
 				},
 			},
 			body
-		) => WS.emit({ ...body, type: "message", humanKey, tableKey }),
+		) => liveTable(tableKey).inputs.messages.emit({ ...body, humanKey }),
 
-		close({
+		close: ({
 			data: {
 				params: { tableKey },
 				humanKey,
 			},
-		}) {
+		}) =>
 			liveTable(tableKey).inputs.connectionChanges.emit({
 				humanKey,
 				presence: "left",
-			});
-		},
+			}),
+
+		error: (error) => err(error),
 	});
 
 export default api;

pkg/server/src/index.ts

@@ -16,8 +16,8 @@ new Elysia()
 		})
 	)
 
-	.onRequest(({ request }) => log.log(request))
-	.onError(({ error }) => log.err(error))
+	.onRequest(({ request }) => console.log(request.url))
+	.onError(({ error }) => console.error(error))
 
 	.get("/ping", () => "pong")
 	.use(api)

pkg/server/src/logging.ts

@@ -10,4 +10,7 @@ export const log = (value: unknown) => LogBus.emit(value);
 export const err = (value: unknown) =>
 	LogBus.emitEvent({ type: "error", value });
 
-LogPool.log();
+LogStream.log();
+LogStream.onError((err) => {
+	console.error(err);
+});

pkg/server/src/table.ts

@@ -1,7 +1,14 @@
 import GAMES, { Game, GameKey } from "@games/shared/games";
-import { isEmpty, multiScan, ValueWithin } from "@games/shared/kefir";
+import {
+	isEmpty,
+	multiScan,
+	partition,
+	set,
+	setDiff,
+	ValueWithin,
+} from "@games/shared/kefirs";
 import { t } from "elysia";
-import { combine, pool, Property } from "kefir";
+import { combine, Observable, pool, Property } from "kefir";
 import Bus, { type Bus as TBus } from "kefir-bus";
 import { log } from "./logging";
 
@@ -31,15 +38,7 @@ type TablePayload<
 			},
 			never
 		>;
-
-		readys: TBus<
-			Attributed & {
-				ready: boolean;
-			},
-			any
-		>;
-		actions: TBus<Attributed & GameAction, any>;
-		quits: TBus<Attributed, any>;
+		messages: TBus<Attributed & TWsIn, any>;
 	};
 	outputs: {
 		global: {
@@ -77,14 +76,11 @@ export const liveTable = <
 	if (!(key in tables)) {
 		const inputs: TablePayload<GameConfig, GameState, GameAction>["inputs"] = {
 			connectionChanges: Bus(),
-			readys: Bus(),
-			actions: Bus(),
-			quits: Bus(),
+			messages: Bus(),
 		};
-		const { connectionChanges, readys, actions, quits } = inputs;
+		const { connectionChanges, messages } = inputs;
 
 		// =======
-		const playerStreams = {};
 
 		// players who have at least one connection to the room
 		const playersPresent = connectionChanges
@@ -102,7 +98,37 @@ export const liveTable = <
 			.map((counts) => Object.keys(counts))
 			.toProperty();
 
-		const gameEnds = quits.map((_) => null);
+		const playerStreams: TablePayload<
+			GameConfig,
+			GameState,
+			GameAction
+		>["outputs"]["player"] = {};
+		playersPresent
+			.map(set)
+			.slidingWindow(2, 2)
+			.map(([prev, cur]) => setDiff([prev, cur]))
+			.onValue(({ added, removed }) => {
+				added.forEach((p) => {
+					playerStreams[p] = {
+						view: Bus(),
+					};
+				});
+				removed.forEach((p) => {
+					delete playerStreams[p];
+				});
+			});
+
+		const { ready, action, quit } = partition(
+			["ready", "action", "quit"],
+			messages
+		) as unknown as {
+			// yuck
+			ready: Observable<Attributed & { ready: boolean }, any>;
+			action: Observable<Attributed & { action: GameAction }, any>;
+			quit: Observable<Attributed, any>;
+		};
+
+		const gameEnds = quit.map((_) => null);
 
 		const playersReady = multiScan(
 			null as {
@@ -114,8 +140,8 @@ export const liveTable = <
 					Object.fromEntries(players.map((p) => [p, prev?.[p] ?? false])),
 			],
 			[
-				readys,
-				(prev, evt: ValueWithin<typeof readys>) =>
+				ready,
+				(prev, evt: ValueWithin<typeof ready>) =>
 					prev?.[evt.humanKey] != null
 						? {
 								...prev,
@@ -165,7 +191,7 @@ export const liveTable = <
 					prev || (game.init() as GameState),
 			],
 			[
-				combine([actions], [gameImpl], (action, impl) => ({
+				combine([action], [gameImpl], (action, impl) => ({
 					action,
 					...impl,
 				})),
@@ -185,7 +211,7 @@ export const liveTable = <
 						action,
 					}) as GameState),
 			],
-			[quits, () => null]
+			[quit, () => null]
 		).toProperty();
 
 		const gameIsActive = gameState