IO Game

npm add rivetkit

/*
This matchmaker uses a two phase open lobby flow.
1. findLobby claims a slot immediately and returns matchId and playerId.
2. The client connects to the match actor with playerId.
3. The match actor claims that pending player through pendingPlayerConnected before first join.
4. Pending players expire after JOIN_RESERVATION_TTL_MS, which removes never connected players.
5. updateMatch reports occupied player count while player_count stays pending + occupied.
*/
import { type ActorContextOf, actor } from "rivetkit";
import { db, type RawAccess } from "rivetkit/db";

import type { registry } from "../index.ts";
import { CAPACITY } from "./config.ts";

const JOIN_RESERVATION_TTL_MS = 15_000;

interface MatchmakerVars {
	// The lobby being created right now, so requests that find every lobby
	// full at the same time wait for it instead of opening one each
	openingLobby: Promise<void> | null;
}

export const ioStyleMatchmaker = actor({
	options: { name: "IO - Matchmaker", icon: "earth-americas" },
	db: db({
		onMigrate: migrateTables,
	}),
	createVars: (): MatchmakerVars => ({ openingLobby: null }),
	actions: {
		findLobby: async (
			c,
		): Promise<{ matchId: string; playerId: string }> => {
			const playerId = crypto.randomUUID();
			let matchId = await reserveSeat(c, playerId);
			while (matchId === null) {
				await openLobby(c);
				matchId = await reserveSeat(c, playerId);
			}
			return { matchId, playerId };
		},
		pendingPlayerConnected: async (
			c,
			input: { matchId: string; playerId: string },
		): Promise<{ accepted: boolean }> => {
			const now = Date.now();
			return await c.db.transaction(
				async (tx) => {
					// Deleting the reservation is what claims it, so a player
					// who connects twice at once is accepted only once.
					const rows = await tx.execute<{ expires_at: number }>(
						`DELETE FROM pending_players WHERE match_id = ? AND player_id = ? RETURNING expires_at`,
						input.matchId,
						input.playerId,
					);
					const reservation = rows[0];
					if (!reservation) {
						return { accepted: false };
					}

					const accepted = reservation.expires_at > now;
					if (accepted) {
						// Count the seat as occupied right away. Otherwise it would
						// look free until the match reports its new count.
						await tx.execute(
							`UPDATE matches SET connected_player_count = connected_player_count + 1 WHERE match_id = ?`,
							input.matchId,
						);
					}
					await syncClaimedPlayerCount(tx, input.matchId, now);
					return { accepted };
				},
				{ name: "claim-seat" },
			);
		},
		updateMatch: async (
			c,
			input: { matchId: string; connectedPlayerCount: number },
		) => {
			const now = Date.now();
			await c.db.transaction(
				async (tx) => {
					await tx.execute(
						`UPDATE matches SET connected_player_count = ?, updated_at = ? WHERE match_id = ?`,
						input.connectedPlayerCount,
						now,
						input.matchId,
					);
					await syncClaimedPlayerCount(tx, input.matchId, now);
				},
				{ name: "update-match" },
			);
		},
		closeMatch: async (c, input: { matchId: string }) => {
			// Delete both in one transaction, so findLobby can't reserve a seat
			// in a match that is halfway closed.
			await c.db.transaction(
				async (tx) => {
					await tx.execute(
						`DELETE FROM pending_players WHERE match_id = ?`,
						input.matchId,
					);
					await tx.execute(
						`DELETE FROM matches WHERE match_id = ?`,
						input.matchId,
					);
				},
				{ name: "close-match" },
			);
		},
	},
});

/**
 * Reserves a seat for `playerId` in the fullest lobby with room and returns
 * its match id, or null when every lobby is full.
 */
async function reserveSeat(
	c: ActorContextOf<typeof ioStyleMatchmaker>,
	playerId: string,
): Promise<string | null> {
	const now = Date.now();

	// Actions run concurrently, so pick the lobby and take the seat in one
	// transaction. Two players asking at once can't both take the last seat.
	return await c.db.transaction(
		async (tx) => {
			await expirePendingPlayers(tx, now);

			const rows = await tx.execute<{ match_id: string }>(
				`SELECT match_id FROM matches WHERE player_count < ? ORDER BY player_count DESC, updated_at DESC LIMIT 1`,
				CAPACITY,
			);
			const matchId = rows[0]?.match_id;
			if (!matchId) {
				return null;
			}

			await tx.execute(
				`INSERT INTO pending_players (match_id, player_id, expires_at, created_at) VALUES (?, ?, ?, ?)`,
				matchId,
				playerId,
				now + JOIN_RESERVATION_TTL_MS,
				now,
			);
			await syncClaimedPlayerCount(tx, matchId, now);
			return matchId;
		},
		{ name: "reserve-seat" },
	);
}

/**
 * Opens one new empty lobby. Requests that find every lobby full at the same
 * time share the lobby that is already being opened.
 */
function openLobby(c: ActorContextOf<typeof ioStyleMatchmaker>): Promise<void> {
	if (!c.vars.openingLobby) {
		c.vars.openingLobby = createLobby(c).finally(() => {
			c.vars.openingLobby = null;
		});
	}
	return c.vars.openingLobby;
}

async function createLobby(
	c: ActorContextOf<typeof ioStyleMatchmaker>,
): Promise<void> {
	const matchId = crypto.randomUUID();

	// Create the match before saving its row, so findLobby never hands out a
	// match that doesn't exist yet. This can't run inside a transaction,
	// because a transaction must not wait on another actor.
	const client = c.client<typeof registry>();
	await client.ioStyleMatch.create([matchId]);

	await c.db.execute(
		`INSERT INTO matches (match_id, player_count, connected_player_count, updated_at) VALUES (?, ?, ?, ?)`,
		matchId,
		0,
		0,
		Date.now(),
	);
}

async function expirePendingPlayers(tx: RawAccess, now: number): Promise<void> {
	const rows = await tx.execute<{ match_id: string }>(
		`SELECT DISTINCT match_id FROM pending_players WHERE expires_at <= ?`,
		now,
	);
	if (rows.length === 0) return;

	await tx.execute(`DELETE FROM pending_players WHERE expires_at <= ?`, now);

	for (const row of rows) {
		await syncClaimedPlayerCount(tx, row.match_id, now);
	}
}

async function syncClaimedPlayerCount(
	tx: RawAccess,
	matchId: string,
	now: number,
): Promise<void> {
	await tx.execute(
		`UPDATE matches
			SET player_count = connected_player_count + COALESCE(
				(SELECT COUNT(*) FROM pending_players WHERE match_id = ?),
				0
			),
			updated_at = ?
			WHERE match_id = ?`,
		matchId,
		now,
		matchId,
	);
}

async function migrateTables(dbHandle: RawAccess) {
	await dbHandle.execute(`
		CREATE TABLE IF NOT EXISTS matches (
			match_id TEXT PRIMARY KEY,
			player_count INTEGER NOT NULL,
			connected_player_count INTEGER NOT NULL DEFAULT 0,
			updated_at INTEGER NOT NULL
		)
	`);
	await ensureColumn(
		dbHandle,
		"matches",
		"connected_player_count",
		"INTEGER NOT NULL DEFAULT 0",
	);
	await ensureColumn(
		dbHandle,
		"matches",
		"updated_at",
		"INTEGER NOT NULL DEFAULT 0",
	);
	await dbHandle.execute(`
		CREATE TABLE IF NOT EXISTS pending_players (
			match_id TEXT NOT NULL,
			player_id TEXT NOT NULL,
			expires_at INTEGER NOT NULL,
			created_at INTEGER NOT NULL,
			PRIMARY KEY (match_id, player_id)
		)
	`);
	await dbHandle.execute(
		"CREATE INDEX IF NOT EXISTS matches_open_idx ON matches (player_count, updated_at)",
	);
	await dbHandle.execute(
		"CREATE INDEX IF NOT EXISTS pending_players_match_idx ON pending_players (match_id)",
	);
	await dbHandle.execute(
		"CREATE INDEX IF NOT EXISTS pending_players_expire_idx ON pending_players (expires_at)",
	);
}

async function ensureColumn(
	dbHandle: RawAccess,
	table: "matches",
	column: "connected_player_count" | "updated_at",
	definition: "INTEGER NOT NULL DEFAULT 0",
) {
	const columns = await dbHandle.execute<{ name: string }>(
		`PRAGMA table_info(${table})`,
	);
	if (!columns.some((col) => col.name === column)) {
		await dbHandle.execute(
			`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`,
		);
	}
}

import { type ActorContextOf, actor, event, UserError } from "rivetkit";
import { db } from "rivetkit/db";
import { interval } from "rivetkit/utils";
import type { registry } from "../index.ts";
import { getPlayerColor } from "../player-color.ts";
import { CAPACITY, SPEED, WORLD_SIZE } from "./config.ts";

const TICK_MS = 100;
const SNAPSHOT_INTERVAL_MS = 1_000; // Saving this often bounds how much progress a crash can lose.
const DISCONNECT_GRACE_MS = 5000;

interface PlayerEntry {
	connId: string | null;
	color: string;
	x: number;
	y: number;
	inputX: number;
	inputY: number;
	disconnectedAt: number | null;
}

interface MatchVars {
	matchId: string;
	tick: number;
	players: Record<string, PlayerEntry>;
	// `onWake` stores its loop here so `onSleep` can wait for it.
	loop: Promise<void> | null;
}

interface ConnParams {
	playerId?: string;
}

type PlayerRow = {
	player_id: string;
	x: number;
	y: number;
};

export const ioStyleMatch = actor({
	options: { name: "IO - Match", icon: "earth-americas" },
	events: {
		snapshot: event<Snapshot>(),
	},
	db: db({
		onMigrate: async (db) => {
			await db.execute(`
				CREATE TABLE IF NOT EXISTS match_info (
					id INTEGER PRIMARY KEY CHECK (id = 1),
					tick INTEGER NOT NULL
				)
			`);
			await db.execute(`
				CREATE TABLE IF NOT EXISTS players (
					player_id TEXT PRIMARY KEY,
					x REAL NOT NULL,
					y REAL NOT NULL
				)
			`);
		},
	}),
	// The match runs in memory and starts from the last save. Its key is the
	// match id.
	createVars: async (c): Promise<MatchVars> => {
		const [info] = await c.db.execute<{ tick: number }>(
			"SELECT tick FROM match_info WHERE id = 1",
		);
		const rows = await c.db.execute<PlayerRow>(
			"SELECT player_id, x, y FROM players",
		);

		// No connection survives a restart, so every claimed player starts
		// disconnected and is dropped if they don't reconnect in time.
		const players: Record<string, PlayerEntry> = {};
		for (const row of rows) {
			players[row.player_id] = {
				connId: null,
				color: getPlayerColor(row.player_id),
				x: row.x,
				y: row.y,
				inputX: 0,
				inputY: 0,
				disconnectedAt: Date.now(),
			};
		}
		return {
			matchId: c.key[0]!,
			tick: info?.tick ?? 0,
			players,
			loop: null,
		};
	},
	onConnect: async (c, conn) => {
		const params = (conn.params as ConnParams | null) ?? {};
		const playerId = params.playerId;
		if (!playerId) {
			conn.disconnect("invalid_player");
			return;
		}
		const existingPlayer = c.vars.players[playerId];
		if (existingPlayer) {
			existingPlayer.connId = conn.id;
			existingPlayer.disconnectedAt = null;
		} else {
			const confirmed = await claimPendingPlayer(c, playerId);
			if (!confirmed) {
				conn.disconnect("player_not_pending");
				return;
			}
			c.vars.players[playerId] = {
				connId: conn.id,
				color: getPlayerColor(playerId),
				x: Math.random() * WORLD_SIZE,
				y: Math.random() * WORLD_SIZE,
				inputX: 0,
				inputY: 0,
				disconnectedAt: null,
			};

			// The matchmaker has used up this player's reservation, so save
			// the player now. A crash before the next save would otherwise
			// turn them away with player_not_pending when they reconnect.
			await saveMatch(c);
		}

		await updateMatchmaker(c);
		broadcastSnapshot(c);
	},
	onDestroy: async (c) => {
		// `c.abortSignal` also fires on destroy. Let the loop finish its last
		// save before the database closes.
		await c.vars.loop;
		const client = c.client<typeof registry>();
		await client.ioStyleMatchmaker
			.getOrCreate(["main"])
			.closeMatch({ matchId: c.vars.matchId });
	},
	onDisconnect: async (c, conn) => {
		const found = findPlayerByConnId(c.vars, conn.id);
		if (!found) return;
		const [, player] = found;
		player.connId = null;
		player.disconnectedAt = Date.now();

		await updateMatchmaker(c);
		broadcastSnapshot(c);
	},
	// `onWake` runs on every start, so the loop restarts after a sleep or crash
	onWake: (c) => {
		c.vars.loop = runMatch(c).catch((error) => {
			c.log.error({ msg: "match loop failed", error });
		});
	},
	// `c.abortSignal` fires on shutdown, so the loop saves and exits on its own
	onSleep: async (c) => {
		await c.vars.loop;
	},
	actions: {
		setInput: (c, input: { inputX: number; inputY: number }) => {
			const found = findPlayerByConnId(c.vars, c.conn.id);
			if (!found) {
				throw new UserError("player not found", {
					code: "player_not_found",
				});
			}
			const [, player] = found;
			player.inputX = Math.max(-1, Math.min(1, input.inputX));
			player.inputY = Math.max(-1, Math.min(1, input.inputY));
		},
		getSnapshot: (c) => buildSnapshot(c),
	},
});

function broadcastSnapshot(c: ActorContextOf<typeof ioStyleMatch>) {
	c.broadcast("snapshot", buildSnapshot(c));
}

/** Steps the match every `TICK_MS` until the actor starts shutting down. */
async function runMatch(c: ActorContextOf<typeof ioStyleMatch>): Promise<void> {
	const tick = interval(TICK_MS);
	let savedAt = Date.now();
	while (!c.aborted) {
		await tick();
		if (c.aborted) break;
		c.vars.tick += 1;

		// The grace period is much shorter than the idle timeout, so an empty
		// match removes its last players here before it goes to sleep.
		const now = Date.now();
		let removedPlayer = false;
		for (const [id, player] of Object.entries(c.vars.players)) {
			if (
				player.disconnectedAt &&
				now - player.disconnectedAt > DISCONNECT_GRACE_MS
			) {
				delete c.vars.players[id];
				removedPlayer = true;
				continue;
			}

			player.x = Math.max(
				0,
				Math.min(WORLD_SIZE, player.x + SPEED * player.inputX),
			);
			player.y = Math.max(
				0,
				Math.min(WORLD_SIZE, player.y + SPEED * player.inputY),
			);
		}
		if (removedPlayer) {
			await updateMatchmaker(c);
		}

		broadcastSnapshot(c);

		// Ticks only change memory. Save to SQLite every
		// SNAPSHOT_INTERVAL_MS instead of on every tick.
		if (now - savedAt >= SNAPSHOT_INTERVAL_MS) {
			await saveMatch(c);
			savedAt = now;
		}
	}

	// Save once more when the Actor stops, so a clean restart loses nothing.
	await saveMatch(c);
}

/**
 * Writes the tick and every claimed player's position to SQLite in one
 * transaction, so a crash leaves either the previous save or this one, never
 * part of both.
 */
async function saveMatch(c: ActorContextOf<typeof ioStyleMatch>) {
	// Copy the values first, because ticks can run while the transaction
	// waits on SQLite.
	const tick = c.vars.tick;
	const players: PlayerRow[] = Object.entries(c.vars.players).map(
		([playerId, player]) => ({
			player_id: playerId,
			x: player.x,
			y: player.y,
		}),
	);

	await c.db.transaction(
		async (tx) => {
			await tx.execute(
				"INSERT OR REPLACE INTO match_info (id, tick) VALUES (1, ?)",
				tick,
			);
			await tx.execute("DELETE FROM players");
			for (const player of players) {
				await tx.execute(
					"INSERT INTO players (player_id, x, y) VALUES (?, ?, ?)",
					player.player_id,
					player.x,
					player.y,
				);
			}
		},
		{ name: "save-match" },
	);
}

/**
 * Tells the matchmaker how full this match is. The count only steers new
 * players to a lobby, and the next join or leave sends it again, so a failed
 * update is logged instead of stopping the caller or the tick loop.
 */
async function updateMatchmaker(c: ActorContextOf<typeof ioStyleMatch>) {
	const client = c.client<typeof registry>();
	try {
		await client.ioStyleMatchmaker.getOrCreate(["main"]).updateMatch({
			matchId: c.vars.matchId,
			connectedPlayerCount: occupiedPlayerCount(c.vars),
		});
	} catch (error) {
		c.log.warn({ msg: "updating the matchmaker failed", error });
	}
}

async function claimPendingPlayer(
	c: ActorContextOf<typeof ioStyleMatch>,
	playerId: string,
): Promise<boolean> {
	const client = c.client<typeof registry>();
	const result = await client.ioStyleMatchmaker
		.getOrCreate(["main"])
		.pendingPlayerConnected({ matchId: c.vars.matchId, playerId });
	return result.accepted;
}

interface Snapshot {
	matchId: string;
	capacity: number;
	tick: number;
	playerCount: number;
	worldSize: number;
	players: Record<string, { x: number; y: number; color: string }>;
}

function buildSnapshot(c: ActorContextOf<typeof ioStyleMatch>): Snapshot {
	const players: Record<string, { x: number; y: number; color: string }> = {};
	for (const [id, entry] of Object.entries(c.vars.players)) {
		if (entry.disconnectedAt) continue;
		players[id] = { x: entry.x, y: entry.y, color: entry.color };
	}
	return {
		matchId: c.vars.matchId,
		capacity: CAPACITY,
		tick: c.vars.tick,
		playerCount: activePlayerCount(c.vars),
		worldSize: WORLD_SIZE,
		players,
	};
}

function findPlayerByConnId(
	vars: MatchVars,
	connId: string,
): [string, PlayerEntry] | null {
	for (const [id, entry] of Object.entries(vars.players)) {
		if (entry.connId === connId) return [id, entry];
	}
	return null;
}

function activePlayerCount(vars: MatchVars): number {
	let count = 0;
	for (const player of Object.values(vars.players)) {
		if (!player.disconnectedAt) count++;
	}
	return count;
}

function occupiedPlayerCount(vars: MatchVars): number {
	return Object.keys(vars.players).length;
}

import { type ActorContextOf, actor, event, UserError } from "rivetkit";
import { db } from "rivetkit/db";
import { interval } from "rivetkit/utils";
import type { registry } from "../index.ts";
import { getPlayerColor } from "../player-color.ts";
import { CAPACITY, SPEED, WORLD_SIZE } from "./config.ts";

const TICK_MS = 100;
const SNAPSHOT_INTERVAL_MS = 1_000; // Saving this often bounds how much progress a crash can lose.
const DISCONNECT_GRACE_MS = 5000;

interface PlayerEntry {
	connId: string | null;
	color: string;
	x: number;
	y: number;
	inputX: number;
	inputY: number;
	disconnectedAt: number | null;
}

interface MatchVars {
	matchId: string;
	tick: number;
	players: Record<string, PlayerEntry>;
	// `onWake` stores its loop here so `onSleep` can wait for it.
	loop: Promise<void> | null;
}

interface ConnParams {
	playerId?: string;
}

type PlayerRow = {
	player_id: string;
	x: number;
	y: number;
};

export const ioStyleMatch = actor({
	options: { name: "IO - Match", icon: "earth-americas" },
	events: {
		snapshot: event<Snapshot>(),
	},
	db: db({
		onMigrate: async (db) => {
			await db.execute(`
				CREATE TABLE IF NOT EXISTS match_info (
					id INTEGER PRIMARY KEY CHECK (id = 1),
					tick INTEGER NOT NULL
				)
			`);
			await db.execute(`
				CREATE TABLE IF NOT EXISTS players (
					player_id TEXT PRIMARY KEY,
					x REAL NOT NULL,
					y REAL NOT NULL
				)
			`);
		},
	}),
	// The match runs in memory and starts from the last save. Its key is the
	// match id.
	createVars: async (c): Promise<MatchVars> => {
		const [info] = await c.db.execute<{ tick: number }>(
			"SELECT tick FROM match_info WHERE id = 1",
		);
		const rows = await c.db.execute<PlayerRow>(
			"SELECT player_id, x, y FROM players",
		);

		// No connection survives a restart, so every claimed player starts
		// disconnected and is dropped if they don't reconnect in time.
		const players: Record<string, PlayerEntry> = {};
		for (const row of rows) {
			players[row.player_id] = {
				connId: null,
				color: getPlayerColor(row.player_id),
				x: row.x,
				y: row.y,
				inputX: 0,
				inputY: 0,
				disconnectedAt: Date.now(),
			};
		}
		return {
			matchId: c.key[0]!,
			tick: info?.tick ?? 0,
			players,
			loop: null,
		};
	},
	onConnect: async (c, conn) => {
		const params = (conn.params as ConnParams | null) ?? {};
		const playerId = params.playerId;
		if (!playerId) {
			conn.disconnect("invalid_player");
			return;
		}
		const existingPlayer = c.vars.players[playerId];
		if (existingPlayer) {
			existingPlayer.connId = conn.id;
			existingPlayer.disconnectedAt = null;
		} else {
			const confirmed = await claimPendingPlayer(c, playerId);
			if (!confirmed) {
				conn.disconnect("player_not_pending");
				return;
			}
			c.vars.players[playerId] = {
				connId: conn.id,
				color: getPlayerColor(playerId),
				x: Math.random() * WORLD_SIZE,
				y: Math.random() * WORLD_SIZE,
				inputX: 0,
				inputY: 0,
				disconnectedAt: null,
			};

			// The matchmaker has used up this player's reservation, so save
			// the player now. A crash before the next save would otherwise
			// turn them away with player_not_pending when they reconnect.
			await saveMatch(c);
		}

		await updateMatchmaker(c);
		broadcastSnapshot(c);
	},
	onDestroy: async (c) => {
		// `c.abortSignal` also fires on destroy. Let the loop finish its last
		// save before the database closes.
		await c.vars.loop;
		const client = c.client<typeof registry>();
		await client.ioStyleMatchmaker
			.getOrCreate(["main"])
			.closeMatch({ matchId: c.vars.matchId });
	},
	onDisconnect: async (c, conn) => {
		const found = findPlayerByConnId(c.vars, conn.id);
		if (!found) return;
		const [, player] = found;
		player.connId = null;
		player.disconnectedAt = Date.now();

		await updateMatchmaker(c);
		broadcastSnapshot(c);
	},
	// `onWake` runs on every start, so the loop restarts after a sleep or crash
	onWake: (c) => {
		c.vars.loop = runMatch(c).catch((error) => {
			c.log.error({ msg: "match loop failed", error });
		});
	},
	// `c.abortSignal` fires on shutdown, so the loop saves and exits on its own
	onSleep: async (c) => {
		await c.vars.loop;
	},
	actions: {
		setInput: (c, input: { inputX: number; inputY: number }) => {
			const found = findPlayerByConnId(c.vars, c.conn.id);
			if (!found) {
				throw new UserError("player not found", {
					code: "player_not_found",
				});
			}
			const [, player] = found;
			player.inputX = Math.max(-1, Math.min(1, input.inputX));
			player.inputY = Math.max(-1, Math.min(1, input.inputY));
		},
		getSnapshot: (c) => buildSnapshot(c),
	},
});

function broadcastSnapshot(c: ActorContextOf<typeof ioStyleMatch>) {
	c.broadcast("snapshot", buildSnapshot(c));
}

/** Steps the match every `TICK_MS` until the actor starts shutting down. */
async function runMatch(c: ActorContextOf<typeof ioStyleMatch>): Promise<void> {
	const tick = interval(TICK_MS);
	let savedAt = Date.now();
	while (!c.aborted) {
		await tick();
		if (c.aborted) break;
		c.vars.tick += 1;

		// The grace period is much shorter than the idle timeout, so an empty
		// match removes its last players here before it goes to sleep.
		const now = Date.now();
		let removedPlayer = false;
		for (const [id, player] of Object.entries(c.vars.players)) {
			if (
				player.disconnectedAt &&
				now - player.disconnectedAt > DISCONNECT_GRACE_MS
			) {
				delete c.vars.players[id];
				removedPlayer = true;
				continue;
			}

			player.x = Math.max(
				0,
				Math.min(WORLD_SIZE, player.x + SPEED * player.inputX),
			);
			player.y = Math.max(
				0,
				Math.min(WORLD_SIZE, player.y + SPEED * player.inputY),
			);
		}
		if (removedPlayer) {
			await updateMatchmaker(c);
		}

		broadcastSnapshot(c);

		// Ticks only change memory. Save to SQLite every
		// SNAPSHOT_INTERVAL_MS instead of on every tick.
		if (now - savedAt >= SNAPSHOT_INTERVAL_MS) {
			await saveMatch(c);
			savedAt = now;
		}
	}

	// Save once more when the Actor stops, so a clean restart loses nothing.
	await saveMatch(c);
}

/**
 * Writes the tick and every claimed player's position to SQLite in one
 * transaction, so a crash leaves either the previous save or this one, never
 * part of both.
 */
async function saveMatch(c: ActorContextOf<typeof ioStyleMatch>) {
	// Copy the values first, because ticks can run while the transaction
	// waits on SQLite.
	const tick = c.vars.tick;
	const players: PlayerRow[] = Object.entries(c.vars.players).map(
		([playerId, player]) => ({
			player_id: playerId,
			x: player.x,
			y: player.y,
		}),
	);

	await c.db.transaction(
		async (tx) => {
			await tx.execute(
				"INSERT OR REPLACE INTO match_info (id, tick) VALUES (1, ?)",
				tick,
			);
			await tx.execute("DELETE FROM players");
			for (const player of players) {
				await tx.execute(
					"INSERT INTO players (player_id, x, y) VALUES (?, ?, ?)",
					player.player_id,
					player.x,
					player.y,
				);
			}
		},
		{ name: "save-match" },
	);
}

/**
 * Tells the matchmaker how full this match is. The count only steers new
 * players to a lobby, and the next join or leave sends it again, so a failed
 * update is logged instead of stopping the caller or the tick loop.
 */
async function updateMatchmaker(c: ActorContextOf<typeof ioStyleMatch>) {
	const client = c.client<typeof registry>();
	try {
		await client.ioStyleMatchmaker.getOrCreate(["main"]).updateMatch({
			matchId: c.vars.matchId,
			connectedPlayerCount: occupiedPlayerCount(c.vars),
		});
	} catch (error) {
		c.log.warn({ msg: "updating the matchmaker failed", error });
	}
}

async function claimPendingPlayer(
	c: ActorContextOf<typeof ioStyleMatch>,
	playerId: string,
): Promise<boolean> {
	const client = c.client<typeof registry>();
	const result = await client.ioStyleMatchmaker
		.getOrCreate(["main"])
		.pendingPlayerConnected({ matchId: c.vars.matchId, playerId });
	return result.accepted;
}

interface Snapshot {
	matchId: string;
	capacity: number;
	tick: number;
	playerCount: number;
	worldSize: number;
	players: Record<string, { x: number; y: number; color: string }>;
}

function buildSnapshot(c: ActorContextOf<typeof ioStyleMatch>): Snapshot {
	const players: Record<string, { x: number; y: number; color: string }> = {};
	for (const [id, entry] of Object.entries(c.vars.players)) {
		if (entry.disconnectedAt) continue;
		players[id] = { x: entry.x, y: entry.y, color: entry.color };
	}
	return {
		matchId: c.vars.matchId,
		capacity: CAPACITY,
		tick: c.vars.tick,
		playerCount: activePlayerCount(c.vars),
		worldSize: WORLD_SIZE,
		players,
	};
}

function findPlayerByConnId(
	vars: MatchVars,
	connId: string,
): [string, PlayerEntry] | null {
	for (const [id, entry] of Object.entries(vars.players)) {
		if (entry.connId === connId) return [id, entry];
	}
	return null;
}

function activePlayerCount(vars: MatchVars): number {
	let count = 0;
	for (const player of Object.values(vars.players)) {
		if (!player.disconnectedAt) count++;
	}
	return count;
}

function occupiedPlayerCount(vars: MatchVars): number {
	return Object.keys(vars.players).length;
}

git clone https://github.com/rivet-dev/rivet.git
cd rivet/examples/multiplayer-game-patterns
pnpm install
pnpm dev