feat: initialize sunrise with 100-player shared rooms
This commit is contained in:
+114
@@ -0,0 +1,114 @@
|
||||
import { createServer } from 'node:http';
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { realpath, stat } from 'node:fs/promises';
|
||||
import { resolve, relative, isAbsolute, extname } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { pipeline } from 'node:stream';
|
||||
import { Server } from 'socket.io';
|
||||
import { MULTIPLAYER } from '../src/constants/multiplayer.ts';
|
||||
import { observer } from '../src/config/observer.ts';
|
||||
import { WeatherController } from '../src/weather/weather.ts';
|
||||
import type { ClientEvents, ServerEvents } from '../src/multiplayer/protocol.ts';
|
||||
import type { VisitStore } from './visits.ts';
|
||||
import { SummitRoom } from './room.ts';
|
||||
import { SERVER_LIMITS } from './constants/multiplayer.ts';
|
||||
import { LOBBY, LOBBY_SERVERS } from '../src/constants/lobby.ts';
|
||||
|
||||
export interface AppOptions {
|
||||
visitStore: VisitStore;
|
||||
weather?: boolean;
|
||||
distDirectory?: string;
|
||||
allowedOrigins?: string[];
|
||||
onError?: (error: unknown) => void;
|
||||
}
|
||||
const mime: Record<string, string> = { '.html': 'text/html; charset=utf-8', '.js': 'text/javascript; charset=utf-8',
|
||||
'.css': 'text/css; charset=utf-8', '.json': 'application/json', '.svg': 'image/svg+xml', '.png': 'image/png',
|
||||
'.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.webp': 'image/webp', '.ico': 'image/x-icon', '.woff2': 'font/woff2' };
|
||||
function contained(root: string, file: string) {
|
||||
const path = relative(root, file);
|
||||
return !path.startsWith('..') && !isAbsolute(path);
|
||||
}
|
||||
|
||||
/** Single-instance factory. Caller owns listen(); close() drains visits and closes the injected store. */
|
||||
export function createApp(options: AppOptions) {
|
||||
const dist = resolve(options.distDirectory ?? fileURLToPath(new URL('../dist/', import.meta.url)));
|
||||
const onError = options.onError ?? (error => console.error('Server operation failed', error));
|
||||
const origins = options.allowedOrigins ?? [...SERVER_LIMITS.defaultOrigins];
|
||||
const allowed = (origin: string | undefined) => origin === undefined || origins.includes(origin);
|
||||
const httpServer = createServer((request, response) => {
|
||||
response.setHeader('X-Content-Type-Options', 'nosniff');
|
||||
const serve = async () => {
|
||||
const rawPath = (request.url ?? '/').split('?')[0];
|
||||
if (rawPath === '/health') {
|
||||
response.statusCode = room.healthy ? 200 : 503;
|
||||
response.setHeader('Content-Type', 'application/json');
|
||||
response.setHeader('Cache-Control', 'no-store');
|
||||
response.end(request.method === 'HEAD' ? undefined : JSON.stringify({ ok: room.healthy })); return;
|
||||
}
|
||||
if (request.method !== 'GET' && request.method !== 'HEAD') { response.writeHead(405, { Allow: 'GET, HEAD' }); response.end(); return; }
|
||||
if (rawPath === LOBBY.endpoint) {
|
||||
response.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' });
|
||||
response.end(request.method === 'HEAD' ? undefined : JSON.stringify({
|
||||
servers: LOBBY_SERVERS,
|
||||
channel: { id: MULTIPLAYER.roomId, name: LOBBY.channelName, capacity: MULTIPLAYER.capacity,
|
||||
players: room.size, available: room.healthy && room.reserved < MULTIPLAYER.capacity },
|
||||
}));
|
||||
return;
|
||||
}
|
||||
let path: string;
|
||||
try { path = decodeURIComponent(rawPath); } catch { response.writeHead(400); response.end(); return; }
|
||||
if (/[\\\u0000:]/.test(path) || path.split('/').some(segment => segment.startsWith('.'))) {
|
||||
response.writeHead(404); response.end(); return;
|
||||
}
|
||||
try {
|
||||
const root = await realpath(dist);
|
||||
let file = await realpath(resolve(root, '.' + (path === '/' ? '/index.html' : path)));
|
||||
if (!contained(root, file)) { response.writeHead(404); response.end(); return; }
|
||||
const type = mime[extname(file)] ?? 'application/octet-stream';
|
||||
// Precompressed assets only, with the same realpath containment check as identity files.
|
||||
const encodings = (request.headers['accept-encoding'] ?? '').toLowerCase().split(',').map(entry => {
|
||||
const [name, ...params] = entry.trim().split(';');
|
||||
const q = params.find(p => p.trim().startsWith('q='));
|
||||
return { name, q: q ? Number(q.trim().slice(2)) : 1 };
|
||||
});
|
||||
const gzip = encodings.find(e => e.name === 'gzip') ?? encodings.find(e => e.name === '*');
|
||||
response.setHeader('Vary', 'Accept-Encoding');
|
||||
if (gzip && gzip.q > 0 && !request.headers.range) {
|
||||
try {
|
||||
const compressed = await realpath(file + '.gz');
|
||||
if (contained(root, compressed) && (await stat(compressed)).isFile()) {
|
||||
file = compressed; response.setHeader('Content-Encoding', 'gzip');
|
||||
}
|
||||
} catch { /* Identity file remains available. */ }
|
||||
}
|
||||
const info = await stat(file);
|
||||
if (!info.isFile()) { response.writeHead(404); response.end(); return; }
|
||||
response.writeHead(200, { 'Content-Type': type, 'Content-Length': info.size, 'Cache-Control': 'no-cache' });
|
||||
if (request.method === 'HEAD') response.end();
|
||||
else pipeline(createReadStream(file), response, error => { if (error) response.destroy(); });
|
||||
} catch { response.writeHead(404); response.end(); }
|
||||
};
|
||||
void serve().catch(error => { onError(error); if (!response.headersSent) response.writeHead(500); response.end(); });
|
||||
});
|
||||
const io = new Server<ClientEvents, ServerEvents>(httpServer, {
|
||||
maxHttpBufferSize: MULTIPLAYER.maxPacketBytes,
|
||||
cors: { origin(origin, callback) { callback(null, allowed(origin)); } },
|
||||
allowRequest(request, callback) { callback(null, allowed(request.headers.origin)); },
|
||||
});
|
||||
const room = new SummitRoom(io, options.visitStore, onError);
|
||||
const timers = [setInterval(() => room.broadcast(), MULTIPLAYER.tickMs),
|
||||
setInterval(() => room.broadcast(true), MULTIPLAYER.snapshotMs),
|
||||
setInterval(() => { void room.heartbeat(); }, MULTIPLAYER.visitHeartbeatMs)];
|
||||
const weather = options.weather === false ? undefined : new WeatherController(observer, state => room.setWeather(state));
|
||||
if (weather) { weather.update(new Date()); timers.push(setInterval(() => weather.update(new Date()), SERVER_LIMITS.weatherSampleMs)); }
|
||||
for (const timer of timers) timer.unref();
|
||||
let closing: Promise<void> | undefined;
|
||||
return { httpServer, io, room, close(): Promise<void> {
|
||||
return closing ??= (async () => {
|
||||
timers.forEach(clearInterval); weather?.dispose();
|
||||
await room.close();
|
||||
await new Promise<void>(resolve => io.close(() => resolve()));
|
||||
await options.visitStore.close();
|
||||
})();
|
||||
} };
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/** Backend-only limits; shared room/wire settings live in src/constants/multiplayer.ts. */
|
||||
export const SERVER_LIMITS = {
|
||||
movementBudgetSeconds: 0.5,
|
||||
updateIntervalMs: 100,
|
||||
weatherSampleMs: 60000,
|
||||
defaultOrigins: ['http://localhost:3000', 'http://127.0.0.1:3000', 'http://localhost:5173', 'http://127.0.0.1:5173'],
|
||||
} as const;
|
||||
@@ -0,0 +1,24 @@
|
||||
import { createApp } from './app.ts';
|
||||
import { MemoryVisitStore, MongoVisitStore, type VisitStore } from './visits.ts';
|
||||
|
||||
async function main() {
|
||||
const port = Number(process.env.PORT ?? 3000);
|
||||
if (!Number.isInteger(port) || port < 0 || port > 65535) throw new Error('Invalid PORT');
|
||||
if (process.env.VISIT_STORE && !['memory', 'mongo'].includes(process.env.VISIT_STORE)) throw new Error('Invalid VISIT_STORE');
|
||||
let visitStore: VisitStore;
|
||||
if (process.env.VISIT_STORE === 'memory') {
|
||||
console.warn('VISIT_STORE=memory: visit history is volatile; development/tests only.');
|
||||
visitStore = new MemoryVisitStore();
|
||||
} else {
|
||||
visitStore = await MongoVisitStore.connect(process.env.MONGODB_URI ?? 'mongodb://172.16.0.7:27017', process.env.MONGODB_DATABASE ?? 'sunrise');
|
||||
}
|
||||
const app = createApp({ visitStore,
|
||||
allowedOrigins: process.env.ALLOWED_ORIGINS?.split(',').map(origin => origin.trim()).filter(Boolean) });
|
||||
const stop = () => { void app.close().catch(error => { console.error('Shutdown failed', error); process.exitCode = 1; }); };
|
||||
process.once('SIGINT', stop); process.once('SIGTERM', stop);
|
||||
app.httpServer.once('error', error => { console.error('HTTP server failed', error); process.exitCode = 1; stop(); });
|
||||
app.httpServer.listen(port, process.env.HOST ?? '127.0.0.1', () => {
|
||||
console.log(`Sunrise single-instance server listening on ${process.env.HOST ?? '127.0.0.1'}:${(app.httpServer.address() as { port: number }).port}`);
|
||||
});
|
||||
}
|
||||
void main().catch(error => { console.error('Server startup failed; admission disabled', error); process.exitCode = 1; });
|
||||
+204
@@ -0,0 +1,204 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { performance } from 'node:perf_hooks';
|
||||
import type { Server, Socket } from 'socket.io';
|
||||
import { MULTIPLAYER } from '../src/constants/multiplayer.ts';
|
||||
import { MOVEMENT } from '../src/constants/movement.ts';
|
||||
import { CHARACTER } from '../src/constants/character.ts';
|
||||
import { SUMMIT_TERRAIN } from '../src/constants/summitTerrain.ts';
|
||||
import { observer } from '../src/config/observer.ts';
|
||||
import { createSummitWalking } from '../src/controls/summitWalking.ts';
|
||||
import { canPray } from '../src/controls/emotes.ts';
|
||||
import { calculateSun } from '../src/astronomy/sunCalculator.ts';
|
||||
import type { ClientEvents, ServerEvents, JoinRequest, JoinResult, PlayerState, PlayerUpdate, RoomState } from '../src/multiplayer/protocol.ts';
|
||||
import type { WeatherStatus } from '../src/weather/weather.ts';
|
||||
import type { EndReason, VisitStore } from './visits.ts';
|
||||
import { SERVER_LIMITS } from './constants/multiplayer.ts';
|
||||
|
||||
export type RoomServer = Server<ClientEvents, ServerEvents>;
|
||||
type RoomSocket = Socket<ClientEvents, ServerEvents>;
|
||||
interface Session {
|
||||
id: string;
|
||||
socket: RoomSocket;
|
||||
player?: PlayerState;
|
||||
departed: boolean;
|
||||
departureAt?: Date;
|
||||
endReason?: EndReason;
|
||||
budget: number;
|
||||
budgetAt: number;
|
||||
packetAt: number;
|
||||
}
|
||||
const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
||||
function validJoin(value: unknown): value is JoinRequest {
|
||||
if (!value || typeof value !== 'object') return false;
|
||||
const v = value as JoinRequest;
|
||||
return v.locationId === MULTIPLAYER.locationId && v.roomId === MULTIPLAYER.roomId
|
||||
&& typeof v.visitorId === 'string' && uuid.test(v.visitorId)
|
||||
&& typeof v.nickname === 'string' && v.nickname.trim().length > 0
|
||||
&& v.nickname.length <= MULTIPLAYER.maxNicknameLength && !/[\u0000-\u001f\u007f]/.test(v.nickname);
|
||||
}
|
||||
function validUpdate(value: unknown): value is PlayerUpdate {
|
||||
if (!value || typeof value !== 'object') return false;
|
||||
const v = value as PlayerUpdate, gear = v.equipment;
|
||||
return Number.isFinite(v.x) && Number.isFinite(v.z) && Number.isFinite(v.yaw)
|
||||
&& Math.abs(v.yaw) <= Math.PI * 2 && typeof v.moving === 'boolean'
|
||||
&& ['idle', 'wave', 'pray'].includes(v.emote) && !!gear
|
||||
&& typeof gear.umbrella === 'boolean' && typeof gear.raincoat === 'boolean'
|
||||
&& typeof gear.hood === 'boolean' && (!gear.hood || gear.raincoat);
|
||||
}
|
||||
|
||||
export class SummitRoom {
|
||||
private io: RoomServer;
|
||||
private store: VisitStore;
|
||||
private onError: (error: unknown) => void;
|
||||
private sessions = new Map<string, Session>();
|
||||
private pending = new Set<Promise<void>>();
|
||||
private writes: Promise<void> = Promise.resolve();
|
||||
private storageHealthy = true;
|
||||
private stopping = false;
|
||||
private dirty = false;
|
||||
private sequence = 0;
|
||||
private serverInstanceId = randomUUID();
|
||||
private walking = createSummitWalking(observer);
|
||||
private weather: WeatherStatus = { status: 'unavailable', weather: null };
|
||||
private maxBudget = MOVEMENT.speed * SERVER_LIMITS.movementBudgetSeconds;
|
||||
constructor(io: RoomServer, store: VisitStore, onError: (error: unknown) => void) {
|
||||
this.io = io; this.store = store; this.onError = onError;
|
||||
io.on('connection', socket => this.attach(socket));
|
||||
}
|
||||
get size() { return [...this.sessions.values()].filter(s => s.player && !s.departed).length; }
|
||||
get reserved() { return this.sessions.size; }
|
||||
get healthy() { return this.storageHealthy && !this.stopping; }
|
||||
snapshot(): RoomState {
|
||||
return { locationId: MULTIPLAYER.locationId, roomId: MULTIPLAYER.roomId, capacity: MULTIPLAYER.capacity,
|
||||
sequence: this.sequence, serverTime: Date.now(), players: [...this.sessions.values()]
|
||||
.filter(s => s.player && !s.departed).map(s => ({ ...s.player!, equipment: { ...s.player!.equipment } })) };
|
||||
}
|
||||
broadcast(force = false) {
|
||||
const at = Date.now();
|
||||
let prayerAvailable: boolean | undefined;
|
||||
for (const { player } of this.sessions.values()) {
|
||||
if (!player || player.emote === 'idle') continue;
|
||||
if (player.emote === 'pray' && prayerAvailable === undefined) {
|
||||
prayerAvailable = canPray(new Date(at), calculateSun(new Date(at), observer).sunrise);
|
||||
}
|
||||
if ((player.emote === 'wave' && at - player.emoteStartedAt >= CHARACTER.waveSeconds * 1000)
|
||||
|| (player.emote === 'pray' && !prayerAvailable)) {
|
||||
player.emote = 'idle'; player.emoteStartedAt = at; player.updatedAt = at; this.dirty = true;
|
||||
}
|
||||
}
|
||||
if (!force && !this.dirty) return;
|
||||
this.sequence++; this.dirty = false;
|
||||
this.io.to(MULTIPLAYER.roomId).emit('room:state', this.snapshot());
|
||||
}
|
||||
setWeather(state: WeatherStatus) { this.weather = state; this.io.to(MULTIPLAYER.roomId).emit('weather:state', state); }
|
||||
private failure(error: unknown) { this.storageHealthy = false; this.onError(error); }
|
||||
private enqueue(write: () => Promise<void>) {
|
||||
this.writes = this.writes.then(write).catch(error => this.failure(error));
|
||||
return this.writes;
|
||||
}
|
||||
heartbeat() {
|
||||
const ids = [...this.sessions.values()].filter(s => s.player && !s.departed).map(s => s.id);
|
||||
const at = new Date();
|
||||
return this.enqueue(() => this.store.heartbeat(ids, at));
|
||||
}
|
||||
private spawn(id: string, nickname: string): PlayerState {
|
||||
const heading = SUMMIT_TERRAIN.heading * Math.PI / 180;
|
||||
return { id, nickname, ...this.walking.start, yaw: Math.atan2(Math.sin(heading), -Math.cos(heading)),
|
||||
moving: false, emote: 'idle', equipment: { umbrella: false, raincoat: false, hood: false },
|
||||
updatedAt: Date.now(), emoteStartedAt: Date.now() };
|
||||
}
|
||||
private attach(socket: RoomSocket) {
|
||||
socket.on('clock:sync', reply => { if (typeof reply === 'function') reply(Date.now()); });
|
||||
socket.on('join', (request, reply) => {
|
||||
if (typeof reply !== 'function') return;
|
||||
const job = this.join(socket, request, reply);
|
||||
this.pending.add(job);
|
||||
void job.finally(() => this.pending.delete(job));
|
||||
});
|
||||
socket.on('player:update', update => this.update(socket, update));
|
||||
socket.on('player:reset', reply => {
|
||||
const s = this.sessions.get(socket.id);
|
||||
if (!s?.player || s.departed) { if (typeof reply === 'function') reply(null); return; }
|
||||
s.player = { ...this.spawn(s.id, s.player.nickname), equipment: { ...s.player.equipment } };
|
||||
s.budget = this.maxBudget; s.budgetAt = performance.now();
|
||||
this.dirty = true;
|
||||
if (typeof reply === 'function') reply(s.player);
|
||||
});
|
||||
socket.on('disconnect', () => this.depart(socket.id, this.stopping ? 'shutdown' : 'disconnect'));
|
||||
}
|
||||
private async join(socket: RoomSocket, request: JoinRequest, reply: (result: JoinResult) => void) {
|
||||
if (!validJoin(request)) { reply({ ok: false, code: 'INVALID_JOIN' }); return; }
|
||||
const existing = this.sessions.get(socket.id);
|
||||
if (existing) {
|
||||
reply(existing.player ? { ok: true, playerId: existing.id, state: this.snapshot() } : { ok: false, code: 'JOIN_PENDING' }); return;
|
||||
}
|
||||
if (!this.healthy) { reply({ ok: false, code: 'STORAGE_UNAVAILABLE' }); return; }
|
||||
if (this.reserved >= MULTIPLAYER.capacity) { reply({ ok: false, code: 'ROOM_FULL' }); return; }
|
||||
const at = new Date(), s: Session = { id: randomUUID(), socket, departed: false, budget: this.maxBudget, budgetAt: performance.now(), packetAt: -Infinity };
|
||||
// Reservation is synchronous, before any awaited storage work.
|
||||
this.sessions.set(socket.id, s);
|
||||
try {
|
||||
await this.store.insert({ _id: s.id, visitorId: request.visitorId, nickname: request.nickname.trim(),
|
||||
locationId: request.locationId, roomId: request.roomId, joinedAt: at, lastSeenAt: at,
|
||||
leftAt: null, endReason: null, serverInstanceId: this.serverInstanceId });
|
||||
} catch (error) {
|
||||
this.sessions.delete(socket.id); this.failure(error);
|
||||
if (socket.connected) reply({ ok: false, code: 'STORAGE_UNAVAILABLE' });
|
||||
// A network error may follow an acknowledged-on-server insert. Close it if it exists.
|
||||
await this.enqueue(() => this.store.finish(s.id, s.departureAt ?? new Date(), s.endReason ?? 'interrupted'));
|
||||
return;
|
||||
}
|
||||
if (s.departed || !socket.connected || this.stopping) {
|
||||
this.sessions.delete(socket.id);
|
||||
await this.enqueue(() => this.store.finish(s.id, s.departureAt ?? new Date(), s.endReason ?? 'shutdown'));
|
||||
return;
|
||||
}
|
||||
s.player = this.spawn(s.id, request.nickname.trim());
|
||||
await socket.join(MULTIPLAYER.roomId);
|
||||
this.broadcast(true);
|
||||
reply({ ok: true, playerId: s.id, state: this.snapshot() });
|
||||
socket.emit('weather:state', this.weather);
|
||||
}
|
||||
private depart(socketId: string, reason: EndReason) {
|
||||
const s = this.sessions.get(socketId);
|
||||
if (!s || s.departed) return;
|
||||
s.departed = true; s.departureAt = new Date(); s.endReason = reason;
|
||||
// Pending inserts retain their reservation until settled; no double release.
|
||||
if (!s.player) return;
|
||||
this.sessions.delete(socketId);
|
||||
void this.enqueue(() => this.store.finish(s.id, s.departureAt!, reason));
|
||||
this.broadcast(true);
|
||||
}
|
||||
private update(socket: RoomSocket, update: PlayerUpdate) {
|
||||
const s = this.sessions.get(socket.id);
|
||||
if (!s?.player || s.departed) return;
|
||||
const now = performance.now();
|
||||
if (now - s.packetAt < SERVER_LIMITS.updateIntervalMs) return;
|
||||
s.packetAt = now;
|
||||
s.budget = Math.min(this.maxBudget, s.budget + Math.max(0, now - s.budgetAt) / 1000 * MOVEMENT.speed);
|
||||
s.budgetAt = now;
|
||||
const distance = validUpdate(update) ? Math.hypot(update.x - s.player.x, update.z - s.player.z) : Infinity;
|
||||
// Bound distance before walking's swept-path loop, including for malicious huge coordinates.
|
||||
if (!Number.isFinite(distance) || distance > s.budget + 1e-8 || !this.walking.canTraverse(s.player, update)) {
|
||||
socket.emit('player:correction', s.player); return;
|
||||
}
|
||||
s.budget -= distance;
|
||||
const at = Date.now();
|
||||
let emote = update.moving ? 'idle' as const : update.emote;
|
||||
if (emote === 'pray' && !canPray(new Date(at), calculateSun(new Date(at), observer).sunrise)) emote = 'idle';
|
||||
const changed = update.x !== s.player.x || update.z !== s.player.z || update.yaw !== s.player.yaw
|
||||
|| update.moving !== s.player.moving || emote !== s.player.emote
|
||||
|| Object.keys(update.equipment).some(key => update.equipment[key as keyof typeof update.equipment] !== s.player!.equipment[key as keyof typeof update.equipment]);
|
||||
if (!changed) return;
|
||||
s.player = { ...s.player, x: update.x, z: update.z, yaw: update.yaw, moving: update.moving, emote,
|
||||
equipment: { umbrella: update.equipment.umbrella, raincoat: update.equipment.raincoat, hood: update.equipment.hood },
|
||||
updatedAt: at, emoteStartedAt: emote === s.player.emote ? s.player.emoteStartedAt : at };
|
||||
this.dirty = true;
|
||||
}
|
||||
async close() {
|
||||
this.stopping = true;
|
||||
for (const id of this.sessions.keys()) this.depart(id, 'shutdown');
|
||||
await Promise.all(this.pending);
|
||||
await this.writes;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
import { MongoClient, type Collection } from 'mongodb';
|
||||
|
||||
export type EndReason = 'disconnect' | 'shutdown' | 'interrupted';
|
||||
export interface VisitRecord {
|
||||
_id: string;
|
||||
visitorId: string;
|
||||
nickname: string;
|
||||
locationId: string;
|
||||
roomId: string;
|
||||
joinedAt: Date;
|
||||
lastSeenAt: Date;
|
||||
leftAt: Date | null;
|
||||
endReason: EndReason | null;
|
||||
serverInstanceId: string;
|
||||
}
|
||||
/** Only visit lifecycle metadata is persisted; visitorId is anonymous, not authentication. */
|
||||
export interface VisitStore {
|
||||
insert(record: VisitRecord): Promise<void>;
|
||||
heartbeat(sessionIds: string[], at: Date): Promise<void>;
|
||||
finish(sessionId: string, at: Date, reason: EndReason): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
/** Explicit test/development fixture. Never selected as a fallback. */
|
||||
export class MemoryVisitStore implements VisitStore {
|
||||
readonly records = new Map<string, VisitRecord>();
|
||||
async insert(record: VisitRecord) {
|
||||
if (this.records.has(record._id)) throw new Error('Duplicate visit');
|
||||
this.records.set(record._id, structuredClone(record));
|
||||
}
|
||||
async heartbeat(ids: string[], at: Date) {
|
||||
for (const id of ids) { const r = this.records.get(id); if (r && r.endReason === null) r.lastSeenAt = at; }
|
||||
}
|
||||
async finish(id: string, at: Date, reason: EndReason) {
|
||||
const r = this.records.get(id);
|
||||
if (r && r.endReason === null) { r.lastSeenAt = at; r.leftAt = at; r.endReason = reason; }
|
||||
}
|
||||
async close() {}
|
||||
}
|
||||
|
||||
export class MongoVisitStore implements VisitStore {
|
||||
private client: MongoClient;
|
||||
private records: Collection<VisitRecord>;
|
||||
private constructor(client: MongoClient, database: string) {
|
||||
this.client = client;
|
||||
this.records = client.db(database).collection<VisitRecord>('visitSessions');
|
||||
}
|
||||
static async connect(uri: string, database: string): Promise<MongoVisitStore> {
|
||||
const client = new MongoClient(uri, { serverSelectionTimeoutMS: 5000, connectTimeoutMS: 5000, socketTimeoutMS: 10000, writeConcern: { w: 'majority', wtimeoutMS: 5000 } });
|
||||
try {
|
||||
await client.connect();
|
||||
const store = new MongoVisitStore(client, database);
|
||||
await store.records.createIndex({ endReason: 1, lastSeenAt: 1 });
|
||||
// Never rewrite historical visits on connection. A crashed session retains
|
||||
// its lastSeenAt and unknown leftAt; this process owns only its new sessions.
|
||||
return store;
|
||||
} catch (error) { await client.close(); throw error; }
|
||||
}
|
||||
async insert(record: VisitRecord) { await this.records.insertOne(record); }
|
||||
async heartbeat(ids: string[], at: Date) {
|
||||
if (ids.length) await this.records.updateMany({ _id: { $in: ids }, endReason: null }, { $max: { lastSeenAt: at } });
|
||||
}
|
||||
async finish(id: string, at: Date, reason: EndReason) {
|
||||
await this.records.updateOne({ _id: id, endReason: null }, { $set: { leftAt: at, endReason: reason }, $max: { lastSeenAt: at } });
|
||||
}
|
||||
async close() { await this.client.close(); }
|
||||
}
|
||||
Reference in New Issue
Block a user