Add community map records and telemetry API; record Workshop 0.0.126

2026-10-30 14:19:39 +01:00
parent c6ee764c6c
commit 5453159867
26 changed files with 3201 additions and 20 deletions
@@ -0,0 +1,264 @@
+import {HttpError, requireValue, object, string, uuid, outcomes, validateEvent, canonical, readJson} from './validation.js';
+
+const json = (value, status = 200) => new Response(JSON.stringify(value), {
+  status, headers: {'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store', 'X-Content-Type-Options': 'nosniff'},
+});
+const now = () => Math.floor(Date.now() / 1000);
+async function hash(value) {
+  const bytes = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(value));
+  return [...new Uint8Array(bytes)].map(x => x.toString(16).padStart(2, '0')).join('');
+}
+function equal(a, b) {
+  if (a.length !== b.length) return false;
+  let diff = 0; for (let i = 0; i < a.length; i++) diff |= a.charCodeAt(i) ^ b.charCodeAt(i);
+  return diff === 0;
+}
+function newToken() {
+  return [...crypto.getRandomValues(new Uint8Array(32))].map(x => x.toString(16).padStart(2, '0')).join('');
+}
+async function authorize(request, env, readOnly = false) {
+  if (!string(env.ADMIN_TOKEN, 256, 32)) throw new HttpError(503, 'Service credentials are not configured');
+  const token = request.headers.get('authorization')?.match(/^Bearer (\S+)$/)?.[1];
+  if (!string(token, 256, 32)) throw new HttpError(401, 'Unauthorized');
+  const digest = await hash(token);
+  const admin = equal(digest, await hash(env.ADMIN_TOKEN));
+  const reader = readOnly && string(env.READ_TOKEN, 256, 32) && equal(digest, await hash(env.READ_TOKEN));
+  if (!admin && !reader) throw new HttpError(401, 'Unauthorized');
+}
+function method(request, expected) {
+  if (request.method !== expected) throw new HttpError(405, `Use ${expected}`);
+}
+function id(value) {
+  requireValue(uuid(value), 'Invalid UUID'); return value.toLowerCase();
+}
+function paging(url) {
+  const limit = url.searchParams.get('limit') ?? '50';
+  const before = url.searchParams.get('before') ?? String(Number.MAX_SAFE_INTEGER);
+  requireValue(/^\d+$/.test(limit) && +limit >= 1 && +limit <= 100, 'limit must be 1–100');
+  requireValue(/^\d+$/.test(before) && Number.isSafeInteger(+before) && +before > 0, 'Invalid before cursor');
+  return {limit: +limit, before: +before};
+}
+function serverFilter(url) {
+  const value = url.searchParams.get('serverId'); return value === null ? null : id(value);
+}
+async function rateLimit(binding, key) {
+  if (!binding) throw new HttpError(503, 'Ingestion rate limits are not configured');
+  if (!(await binding.limit({key})).success) throw new HttpError(429, 'Too many requests; retry later');
+}
+
+// Count terminal gameplay results once per mission, scoped to the actual loaded
+// terrain resource. Startup failures, interrupted sessions and empty restarts
+// never become map losses. No identities or arbitrary player text leave this API.
+async function mapRecords(db, terrain, serverId = null) {
+  const scope = terrain === null ? '' : " AND lower(json_extract(payload, '$.terrain')) = lower(?)";
+  const terrainValues = terrain === null ? [] : [terrain];
+  const select = `SELECT COUNT(*) AS completed,
+    COALESCE(SUM(outcome = 'victory'), 0) AS wins,
+    COALESCE(SUM(outcome = 'defeat'), 0) AS losses,
+    COALESCE(SUM(partial), 0) AS partialMissions
+    FROM missions WHERE terminal = 1 AND outcome IN ('victory', 'defeat')${scope}`;
+  const statements = [db.prepare(select).bind(...terrainValues)];
+  if (serverId) statements.push(db.prepare(`${select} AND server_id = ?`).bind(...terrainValues, serverId));
+  const results = await db.batch(statements);
+  const record = (row) => ({...row, winRate: row.completed ? Math.round(1000 * row.wins / row.completed) / 10 : null});
+  return {schemaVersion: 1, terrain, allServers: record(results[0].results[0]),
+    server: serverId ? record(results[1].results[0]) : null};
+}
+
+async function publicSummary(request, env, url) {
+  method(request, 'GET');
+  await rateLimit(env.REQUEST_LIMITER, 'summary:' + await hash(request.headers.get('cf-connecting-ip') ?? 'local'));
+  const terrain = url.searchParams.get('terrain');
+  requireValue(terrain === null || string(terrain, 1024, 1), 'Invalid terrain');
+  const summary = await mapRecords(env.DB, terrain, serverFilter(url));
+  if (terrain === null) {
+    // A bounded alphabetical map list makes the browser response useful without
+    // knowing a resource ID. The terrain filter retrieves any map individually.
+    const rows = await env.DB.prepare(`SELECT lower(json_extract(payload, '$.terrain')) AS terrain,
+      COUNT(*) AS completed, SUM(outcome = 'victory') AS wins, SUM(outcome = 'defeat') AS losses
+      FROM missions WHERE terminal = 1 AND outcome IN ('victory', 'defeat')
+      GROUP BY lower(json_extract(payload, '$.terrain')) ORDER BY terrain LIMIT 101`).all();
+    summary.maps = rows.results.slice(0, 100);
+    summary.mapsTruncated = rows.results.length > 100;
+  }
+  return json(summary);
+}
+
+async function ingest(request, env) {
+  method(request, 'POST');
+  // Cloudflare supplies this header. Hash it for ephemeral rate-limit keys;
+  // neither the original IP nor its hash is persisted in the telemetry database.
+  const source = await hash(request.headers.get('cf-connecting-ip') ?? 'local');
+  await rateLimit(env.REQUEST_LIMITER, source);
+  const envelope = await readJson(request);
+  requireValue(object(envelope) && Object.keys(envelope).every(k => ['token', 'event'].includes(k)), 'Invalid envelope');
+  if (!string(envelope.token, 256, 32)) throw new HttpError(401, 'Unauthorized');
+  requireValue(object(envelope.event), 'Invalid event');
+  const serverId = id(envelope.event.serverId);
+  const server = await env.DB.prepare('SELECT token_hash, active FROM servers WHERE server_id = ?').bind(serverId).first();
+  // Always hash before comparison; never persist/log the incoming token.
+  const digest = await hash(envelope.token);
+  if (server && (!equal(digest, server.token_hash) || !server.active)) throw new HttpError(401, 'Unauthorized');
+  const event = validateEvent(envelope.event);
+  const payload = canonical(event);
+  // Native queue bounds strings, so permit UTF-8 expansion at the HTTP boundary.
+  if (payload.length > 65536) throw new HttpError(413, 'Event too large');
+  if (!server) await rateLimit(env.ENROLLMENT_LIMITER, source);
+  // Include the credential hash so guesses cannot exhaust another server's quota.
+  await rateLimit(env.SERVER_LIMITER, `${serverId}:${digest}`);
+  const payloadHash = await hash(payload);
+  const results = await env.DB.batch([
+    env.DB.prepare(`INSERT INTO servers(server_id, name, token_hash, created_at, registration)
+      VALUES (?, ?, ?, ?, 'automatic') ON CONFLICT(server_id) DO NOTHING`)
+      .bind(serverId, event.serverName, digest, now()),
+    env.DB.prepare(`INSERT INTO events(server_id, event_id, mission_id, received_at, payload_hash, payload)
+      SELECT ?, ?, ?, ?, ?, ? FROM servers WHERE server_id = ? AND token_hash = ? AND active = 1
+      ON CONFLICT(server_id, event_id) DO NOTHING RETURNING event_id`)
+      .bind(serverId, event.eventId, event.missionId, now(), payloadHash, payload, serverId, digest),
+    env.DB.prepare('SELECT payload_hash FROM events WHERE server_id = ? AND event_id = ?').bind(serverId, event.eventId),
+    env.DB.prepare('SELECT token_hash, active FROM servers WHERE server_id = ?').bind(serverId),
+  ]);
+  // Resolve concurrent first claims and revocations inside the same transaction.
+  const owner = results[3].results[0];
+  if (!owner?.active || !equal(owner.token_hash, digest)) throw new HttpError(401, 'Unauthorized');
+  if (results[2].results[0]?.payload_hash !== payloadHash) throw new HttpError(409, 'eventId already exists with different content');
+  const duplicate = results[1].results.length === 0;
+  const reply = {accepted: true, duplicate, eventId: event.eventId};
+  if (event.type === 'mission_finished' && ['victory', 'defeat'].includes(event.outcome)) {
+    // Stats are a post-commit convenience. Failure to read them must not turn a
+    // durably accepted event into a false storage failure or cause double credit.
+    try {
+      const selected = await env.DB.prepare('SELECT event_id FROM missions WHERE server_id = ? AND mission_id = ?')
+        .bind(serverId, event.missionId).first();
+      if (selected?.event_id === event.eventId) {
+        reply.community = {...await mapRecords(env.DB, event.terrain, serverId), serverId,
+          missionId: event.missionId, eventId: event.eventId, outcome: event.outcome};
+      }
+    } catch { /* Acknowledgement remains valid; the game shows unavailable totals. */ }
+  }
+  return json(reply, duplicate ? 200 : 201);
+}
+
+async function servers(request, env, url) {
+  await authorize(request, env);
+  if (request.method === 'GET') {
+    const {limit, before} = paging(url);
+    const result = await env.DB.prepare(`SELECT rowid AS cursor, server_id AS serverId, name, active, registration, created_at AS createdAt
+      FROM servers WHERE rowid < ? ORDER BY rowid DESC LIMIT ?`).bind(before, limit).all();
+    return json({servers: result.results, nextBefore: result.results.length === limit ? result.results.at(-1).cursor : null});
+  }
+  method(request, 'POST');
+  const body = await readJson(request, 4096);
+  requireValue(object(body) && Object.keys(body).every(k => ['serverId', 'name'].includes(k)), 'Invalid server');
+  const serverId = id(body.serverId);
+  requireValue(string(body.name, 120, 1), 'name must be 1–120 characters');
+  const token = newToken();
+  const created = await env.DB.prepare(`INSERT INTO servers(server_id, name, token_hash, created_at)
+    VALUES (?, ?, ?, ?) ON CONFLICT(server_id) DO NOTHING RETURNING server_id`)
+    .bind(serverId, body.name, await hash(token), now()).first();
+  if (!created) throw new HttpError(409, 'Server already exists; rotate its token instead');
+  return json({serverId, token, endpoint: `${url.origin}/api/mission-events`}, 201);
+}
+
+async function serverCredential(request, env, serverId, rotate) {
+  await authorize(request, env);
+  method(request, rotate ? 'POST' : 'DELETE');
+  const token = rotate ? newToken() : null;
+  const statement = rotate
+    ? env.DB.prepare('UPDATE servers SET token_hash = ?, active = 1 WHERE server_id = ? RETURNING server_id').bind(await hash(token), serverId)
+    : env.DB.prepare('UPDATE servers SET active = 0 WHERE server_id = ? RETURNING server_id').bind(serverId);
+  if (!await statement.first()) throw new HttpError(404, 'Server not found');
+  return json(rotate ? {serverId, token} : {serverId, active: false});
+}
+
+async function missions(request, env, url) {
+  method(request, 'GET'); await authorize(request, env, true);
+  const {limit, before} = paging(url), serverId = serverFilter(url);
+  const outcome = url.searchParams.get('outcome');
+  requireValue(outcome === null || [...outcomes, 'unfinished'].includes(outcome), 'Invalid outcome filter');
+  const conditions = ['id < ?'], values = [before];
+  if (serverId) { conditions.push('server_id = ?'); values.push(serverId); }
+  if (outcome) { conditions.push('outcome = ?'); values.push(outcome === 'unfinished' ? '' : outcome); }
+  const rows = await env.DB.prepare(`SELECT id AS cursor, server_id AS serverId, mission_id AS missionId,
+    CASE WHEN terminal = 1 THEN outcome ELSE 'unfinished' END AS outcome,
+    elapsed_seconds AS elapsedSeconds, occurred_at AS occurredAt, received_at AS receivedAt,
+    partial, peak_players AS peakPlayers, dropped_events AS droppedEvents,
+    json_extract(payload, '$.terrain') AS terrain, json_extract(payload, '$.playersTruncated') AS playersTruncated
+    FROM missions WHERE ${conditions.join(' AND ')} ORDER BY id DESC LIMIT ?`).bind(...values, limit).all();
+  return json({missions: rows.results, nextBefore: rows.results.length === limit ? rows.results.at(-1).cursor : null});
+}
+
+async function missionDetails(request, env, serverId, missionId, events, url) {
+  method(request, 'GET'); await authorize(request, env, true);
+  if (events) {
+    const {limit, before} = paging(url);
+    const rows = await env.DB.prepare(`SELECT id, received_at, payload FROM events
+      WHERE server_id = ? AND mission_id = ? AND id < ? ORDER BY id DESC LIMIT ?`)
+      .bind(serverId, missionId, before, limit).all();
+    return json({events: rows.results.map(row => ({cursor: row.id, receivedAt: row.received_at, event: JSON.parse(row.payload)})),
+      nextBefore: rows.results.length === limit ? rows.results.at(-1).id : null});
+  }
+  const row = await env.DB.prepare('SELECT received_at, payload FROM missions WHERE server_id = ? AND mission_id = ?')
+    .bind(serverId, missionId).first();
+  if (!row) throw new HttpError(404, 'Mission not found');
+  return json({receivedAt: row.received_at, event: JSON.parse(row.payload)});
+}
+
+async function stats(request, env, url) {
+  method(request, 'GET'); await authorize(request, env, true);
+  const serverId = serverFilter(url);
+  const result = await env.DB.prepare(`SELECT CASE WHEN terminal = 1 THEN outcome ELSE 'unfinished' END AS outcome,
+    COUNT(*) AS missions, SUM(partial) AS partialMissions FROM missions
+    ${serverId ? 'WHERE server_id = ?' : ''} GROUP BY terminal, outcome`)
+    .bind(...(serverId ? [serverId] : [])).all();
+  return json({serverId, outcomes: result.results});
+}
+
+async function players(request, env, url) {
+  method(request, 'GET'); await authorize(request, env, true);
+  const serverId = serverFilter(url), {limit} = paging(url);
+  const after = url.searchParams.get('after');
+  if (after !== null) id(after);
+  const conditions = [], values = [];
+  if (serverId) { conditions.push('server_id = ?'); values.push(serverId); }
+  if (after) { conditions.push('player_id > ?'); values.push(after.toLowerCase()); }
+  const rows = await env.DB.prepare(`SELECT player_id AS playerId, COUNT(*) AS attempts,
+    SUM(escaped) AS escapes, SUM(1 - escaped) AS didNotEscape,
+    SUM(partial) AS partialMissions, SUM(players_truncated) AS truncatedMissions,
+    MIN(CASE WHEN escaped = 1 THEN elapsed_seconds END) AS bestEscapeSeconds,
+    SUM(kills) AS kills, COUNT(kills) AS missionsWithKills,
+    SUM(friendly_kills) AS friendlyKills, COUNT(friendly_kills) AS missionsWithFriendlyKills,
+    SUM(deaths) AS deaths, COUNT(deaths) AS missionsWithDeaths,
+    SUM(downed) AS downed, COUNT(downed) AS missionsWithDowned,
+    SUM(recoveries) AS recoveries, COUNT(recoveries) AS missionsWithRecoveries
+    FROM final_players ${conditions.length ? `WHERE ${conditions.join(' AND ')}` : ''}
+    GROUP BY player_id ORDER BY player_id LIMIT ?`).bind(...values, limit).all();
+  return json({players: rows.results, nextAfter: rows.results.length === limit ? rows.results.at(-1).playerId : null});
+}
+
+export default {
+  async fetch(request, env) {
+    try {
+      const url = new URL(request.url), path = url.pathname;
+      if (path === '/health') { method(request, 'GET'); return json({ok: true, service: 'co10escape-telemetry'}); }
+      if (path === '/' || path === '/api/summary' || (path === '/api/mission-events' && request.method === 'GET')) {
+        return await publicSummary(request, env, url);
+      }
+      if (path === '/api/mission-events') return await ingest(request, env);
+      if (path === '/api/servers') return await servers(request, env, url);
+      const credential = path.match(/^\/api\/servers\/([^/]+)(\/token)?$/);
+      if (credential) return await serverCredential(request, env, id(credential[1]), !!credential[2]);
+      if (path === '/api/missions') return await missions(request, env, url);
+      const mission = path.match(/^\/api\/missions\/([^/]+)\/([^/]+)(\/events)?$/);
+      if (mission) return await missionDetails(request, env, id(mission[1]), id(mission[2]), !!mission[3], url);
+      if (path === '/api/stats') return await stats(request, env, url);
+      if (path === '/api/players') return await players(request, env, url);
+      throw new HttpError(404, 'Not found');
+    } catch (error) {
+      // Database failures must never acknowledge an event. Do not log request
+      // bodies, player identities, SQL parameters or credentials on failure.
+      if (error instanceof HttpError) return json({error: error.message}, error.status);
+      return json({error: 'Storage temporarily unavailable'}, 503);
+    }
+  },
+};
@@ -0,0 +1,117 @@
+export class HttpError extends Error {
+  constructor(status, message) { super(message); this.status = status; }
+}
+export function requireValue(ok, message) {
+  if (!ok) throw new HttpError(400, message);
+}
+export function object(value) {
+  return value !== null && typeof value === 'object' && !Array.isArray(value);
+}
+export function string(value, max, min = 0) {
+  return typeof value === 'string' && value.length >= min && value.length <= max;
+}
+export function integer(value, max = 2147483647) {
+  return Number.isInteger(value) && value >= 0 && value <= max;
+}
+export function uuid(value) {
+  return typeof value === 'string' && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value)
+    && value !== '00000000-0000-0000-0000-000000000000';
+}
+function keys(value, allowed) {
+  requireValue(object(value) && Object.keys(value).every(k => allowed.includes(k)), 'Unexpected object fields');
+}
+const types = ['mission_started', 'mission_restarted', 'phase_changed', 'heartbeat', 'telemetry_resumed', 'mission_finished'];
+export const outcomes = ['victory', 'defeat', 'preparation_failed', 'restarted_empty', 'interrupted'];
+const counters = ['kills', 'friendlyKills', 'deaths', 'downed', 'recoveries'];
+
+export function validateEvent(event) {
+  keys(event, ['schemaVersion', 'eventId', 'missionId', 'serverId', 'serverName', 'type',
+    'occurredAt', 'elapsedSeconds', 'partial', 'phase', 'outcome', 'previousMissionId',
+    'previousOutcome', 'terrain', 'scenario', 'gameVersion', 'mods', 'modsTruncated',
+    'playersTruncated', 'connectedPlayers', 'peakPlayers', 'players', 'reason',
+    'preparationError', 'settings', 'droppedEvents']);
+  requireValue(event.schemaVersion === 1, 'Unsupported schemaVersion');
+  for (const key of ['serverId', 'missionId', 'eventId']) {
+    requireValue(uuid(event[key]), `Invalid ${key}`);
+    event[key] = event[key].toLowerCase();
+  }
+  requireValue(types.includes(event.type), 'Invalid event type');
+  requireValue(event.type === 'mission_finished' ? outcomes.includes(event.outcome) : event.outcome === '', 'Invalid event outcome');
+  requireValue(integer(event.phase, 6), 'Invalid phase');
+  requireValue(event.outcome !== 'victory' || event.phase === 5, 'Victory requires phase 5');
+  requireValue(event.outcome !== 'defeat' || event.phase === 6, 'Defeat requires phase 6');
+  requireValue(event.previousMissionId === '' || uuid(event.previousMissionId), 'Invalid previousMissionId');
+  event.previousMissionId = event.previousMissionId.toLowerCase();
+  requireValue(['', 'unfinished', ...outcomes].includes(event.previousOutcome), 'Invalid previousOutcome');
+  for (const [key, limit] of Object.entries({serverName: 120, terrain: 1024, scenario: 512,
+    gameVersion: 128, reason: 4096, preparationError: 4096})) {
+    requireValue(string(event[key], limit), `Invalid ${key}`);
+  }
+  for (const key of ['partial', 'modsTruncated', 'playersTruncated']) {
+    requireValue(typeof event[key] === 'boolean', `Invalid ${key}`);
+  }
+  for (const key of ['elapsedSeconds', 'connectedPlayers', 'peakPlayers', 'droppedEvents']) {
+    requireValue(integer(event[key]), `Invalid ${key}`);
+  }
+  requireValue(integer(event.occurredAt, 253402300799), 'Invalid occurredAt');
+  requireValue(event.peakPlayers >= event.connectedPlayers, 'peakPlayers is below connectedPlayers');
+  requireValue(Array.isArray(event.players) && event.players.length <= 128, 'Invalid players');
+  const ids = new Set();
+  for (const p of event.players) {
+    keys(p, ['id', 'name', 'authenticated', 'connected', 'participated', 'escaped', 'connectedSeconds', ...counters]);
+    requireValue(string(p.id, 128, 1) && string(p.name, 128), 'Invalid player identity');
+    for (const key of ['authenticated', 'connected', 'participated', 'escaped']) {
+      requireValue(typeof p[key] === 'boolean', `Invalid player ${key}`);
+    }
+    requireValue(!p.authenticated || uuid(p.id), 'Authenticated player ID must be a UUID');
+    if (p.authenticated) p.id = p.id.toLowerCase();
+    requireValue(!ids.has(p.id), 'Duplicate player identity'); ids.add(p.id);
+    requireValue(!p.escaped || (p.participated && event.outcome === 'victory'), 'Invalid escaped player');
+    requireValue(integer(p.connectedSeconds), 'Invalid connectedSeconds');
+    for (const key of counters) {
+      requireValue(p[key] === undefined || integer(p[key]), `Invalid player ${key}`);
+    }
+  }
+  requireValue(Array.isArray(event.mods) && event.mods.length <= 128, 'Invalid mods');
+  for (const mod of event.mods) {
+    keys(mod, ['id', 'name', 'version']);
+    requireValue(string(mod.id, 128, 1) && string(mod.name, 512) && string(mod.version, 128), 'Invalid mod');
+  }
+  requireValue(object(event.settings) && Object.keys(event.settings).length <= 512, 'Invalid settings');
+  for (const [key, value] of Object.entries(event.settings)) {
+    requireValue(/^[a-zA-Z][a-zA-Z0-9_]{0,127}$/.test(key) && Number.isInteger(value)
+      && value >= -2147483648 && value <= 2147483647, 'Invalid setting');
+  }
+  return event;
+}
+
+// Sort object keys for idempotency even if a proxy reformats JSON. Array order
+// remains meaningful. Only the event is persisted; the envelope token is not.
+export function canonical(value) {
+  if (Array.isArray(value)) return `[${value.map(canonical).join(',')}]`;
+  if (object(value)) return `{${Object.keys(value).sort().map(k => `${JSON.stringify(k)}:${canonical(value[k])}`).join(',')}}`;
+  return JSON.stringify(value);
+}
+
+export async function readJson(request, limit = 270336) {
+  if (request.headers.get('content-type')?.split(';')[0].trim().toLowerCase() !== 'application/json') {
+    throw new HttpError(415, 'Content-Type must be application/json');
+  }
+  if (Number(request.headers.get('content-length')) > limit) throw new HttpError(413, 'Request too large');
+  if (!request.body) throw new HttpError(400, 'JSON body required');
+  const reader = request.body.getReader();
+  const chunks = []; let size = 0;
+  try {
+    while (true) {
+      const {done, value} = await reader.read();
+      if (done) break;
+      size += value.byteLength;
+      if (size > limit) { await reader.cancel(); throw new HttpError(413, 'Request too large'); }
+      chunks.push(value);
+    }
+  } finally { reader.releaseLock(); }
+  const bytes = new Uint8Array(size); let offset = 0;
+  for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; }
+  try { return JSON.parse(new TextDecoder('utf-8', {fatal: true}).decode(bytes)); }
+  catch { throw new HttpError(400, 'Invalid JSON'); }
+}