452 lines
18 KiB
TypeScript
452 lines
18 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import { db } from '../db/index.js';
|
|
import {
|
|
enrichGameChatTimeline,
|
|
type ChatMatch,
|
|
type ChatRoundReport,
|
|
type ContextualChatEntry
|
|
} from '../../game-chat.js';
|
|
import { parseGameEvent } from '../../telemetry.js';
|
|
import { telemetryRetention, telemetryCutoff } from './telemetry-requests.js';
|
|
import { getServer, listServers, type ServerRow } from './servers.js';
|
|
import {
|
|
gameLogKey,
|
|
resolveLogPath,
|
|
readLogChunk,
|
|
type LogCursor,
|
|
type TailResult
|
|
} from './logs.js';
|
|
import { listHosts } from '../hosts/index.js';
|
|
import { observationReader } from './status.js';
|
|
|
|
/** Historical context is scoped to this registration, including round-one matches. */
|
|
export function serverChatTimeline(server: ServerRow, tail: TailResult): ContextualChatEntry[] {
|
|
const unscoped = enrichGameChatTimeline(tail.lines, {
|
|
mtime: tail.mtime,
|
|
matches: [],
|
|
reports: []
|
|
});
|
|
const times = unscoped.flatMap((entry) =>
|
|
entry.context.realTime ? [entry.context.realTime] : []
|
|
);
|
|
if (!times.length) return unscoped;
|
|
const since = times.reduce((a, b) => (a < b ? a : b)),
|
|
until = times.reduce((a, b) => (a > b ? a : b));
|
|
const handle = db();
|
|
const matches = handle
|
|
.prepare(
|
|
`SELECT id, started_at AS startedAt, ended_at AS endedAt,
|
|
last_received_at AS lastReceivedAt, status, start_observed AS startObserved
|
|
FROM leaderboard_matches WHERE server = ? AND server_deleted_at IS NULL
|
|
AND started_at <= ? AND (COALESCE(ended_at,last_received_at) >= ? OR status='active')
|
|
ORDER BY id DESC LIMIT 2000`
|
|
)
|
|
.all(server.name, until, since) as Array<
|
|
Omit<ChatMatch, 'startObserved'> & { startObserved: number }
|
|
>;
|
|
|
|
// Diagnostic history survives deletion. Its receipt time must belong to the
|
|
// current registration so a reused server name cannot inherit old rounds.
|
|
const cutoff = telemetryCutoff();
|
|
const roundSince = new Date(Date.parse(since) - 120_000).toISOString();
|
|
const matchSince = matches.reduce(
|
|
(earliest, match) => (match.startedAt < earliest ? match.startedAt : earliest),
|
|
since
|
|
);
|
|
const reports = handle
|
|
.prepare(
|
|
`SELECT received_at AS receivedAt, event, round,
|
|
player_count AS playerCount FROM telemetry_requests
|
|
WHERE server = ? AND status = 200 AND received_at >= ?
|
|
AND julianday(received_at) >= julianday(?)
|
|
AND (received_at >= ? OR (event='game_start' AND received_at >= ?))
|
|
AND received_at <= ?
|
|
AND event IN ('game_start', 'round_change', 'heartbeat', 'game_end')
|
|
AND player_count >= 0 AND (event = 'game_start' OR round >= 0)
|
|
ORDER BY id DESC LIMIT ?`
|
|
)
|
|
.all(
|
|
server.name,
|
|
cutoff,
|
|
server.created_at,
|
|
roundSince,
|
|
matchSince,
|
|
new Date(Date.parse(until) + 120_000).toISOString(),
|
|
telemetryRetention().maxPerServer || 10000
|
|
) as ChatRoundReport[];
|
|
|
|
// Saving gameplay does not depend on diagnostic logging succeeding. Keep the
|
|
// surviving accepted snapshot useful when its request log was not recorded.
|
|
const latest = handle
|
|
.prepare(
|
|
`SELECT received_at AS receivedAt, payload FROM server_telemetry
|
|
WHERE server = ? AND received_at >= ? AND julianday(received_at) >= julianday(?)`
|
|
)
|
|
.get(server.name, cutoff, server.created_at) as
|
|
{ receivedAt: string; payload: string | null } | undefined;
|
|
if (latest?.payload) {
|
|
try {
|
|
const event = parseGameEvent(JSON.parse(latest.payload));
|
|
const round = event.event === 'game_start' ? null : event.round;
|
|
if (
|
|
event.server === server.name &&
|
|
!reports.some(
|
|
(report) =>
|
|
report.receivedAt === latest.receivedAt &&
|
|
report.event === event.event &&
|
|
report.round === round &&
|
|
report.playerCount === event.players.length
|
|
)
|
|
) {
|
|
reports.push({
|
|
receivedAt: latest.receivedAt,
|
|
event: event.event,
|
|
round,
|
|
playerCount: event.players.length
|
|
});
|
|
}
|
|
} catch {
|
|
/* Legacy malformed snapshots cannot establish chat context. */
|
|
}
|
|
}
|
|
|
|
return enrichGameChatTimeline(tail.lines, {
|
|
mtime: tail.mtime,
|
|
matches: matches.map((match) => ({
|
|
...match,
|
|
startObserved: Boolean(match.startObserved)
|
|
})),
|
|
reports
|
|
});
|
|
}
|
|
|
|
export function matchChatTimeline(matchId: number): ContextualChatEntry[] {
|
|
return (
|
|
db()
|
|
.prepare(
|
|
'SELECT message_key,occurrence,entry FROM match_chat WHERE match_id = ? ORDER BY rowid'
|
|
)
|
|
.all(matchId) as { message_key: string; occurrence: number; entry: string }[]
|
|
).map((row) => ({
|
|
...JSON.parse(row.entry),
|
|
matchId,
|
|
messageId: `${matchId}:${row.message_key}:${row.occurrence}`
|
|
}));
|
|
}
|
|
|
|
/** Save before filtering the live view so completed matches retain their messages. */
|
|
function archiveChat(entries: ContextualChatEntry[]): void {
|
|
const handle = db();
|
|
const insert = handle.prepare(`INSERT INTO match_chat (match_id, message_key, occurrence, entry)
|
|
VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING`);
|
|
const occurrences = new Map<string, number>();
|
|
handle.transaction(() => {
|
|
for (const entry of entries) {
|
|
if (entry.type !== 'message' || entry.matchId === undefined) continue;
|
|
// Engine time is stable across reads; estimated wall time is not.
|
|
// Count identical messages separately, including repeats in one second.
|
|
const key = createHash('sha256').update(JSON.stringify(entry.message)).digest('hex');
|
|
const scope = `${entry.matchId}:${key}`;
|
|
const occurrence = (occurrences.get(scope) ?? 0) + 1;
|
|
occurrences.set(scope, occurrence);
|
|
insert.run(entry.matchId, key, occurrence, JSON.stringify(entry));
|
|
}
|
|
})();
|
|
}
|
|
|
|
export function currentMatchChat(
|
|
server: ServerRow,
|
|
tail: TailResult | null
|
|
): ContextualChatEntry[] {
|
|
const entries = tail ? serverChatTimeline(server, tail) : [];
|
|
archiveChat(entries);
|
|
return currentChatEntries(server, entries);
|
|
}
|
|
|
|
function currentChatEntries(
|
|
server: ServerRow,
|
|
entries: ContextualChatEntry[]
|
|
): ContextualChatEntry[] {
|
|
const match = db()
|
|
.prepare(
|
|
`SELECT id, started_at AS startedAt FROM leaderboard_matches
|
|
WHERE server = ? AND server_deleted_at IS NULL ORDER BY id DESC LIMIT 1`
|
|
)
|
|
.get(server.name) as { id: number; startedAt: string } | undefined;
|
|
let startIndex = entries.length - 1;
|
|
while (startIndex >= 0) {
|
|
const entry = entries[startIndex];
|
|
if (entry.type === 'boundary' && entry.event !== 'end') break;
|
|
startIndex--;
|
|
}
|
|
const current = entries.slice(Math.max(0, startIndex));
|
|
if (!match) {
|
|
// Servers without gameplay telemetry still have log-defined sessions.
|
|
return current.filter(
|
|
(entry) =>
|
|
entry.context.realTime !== null &&
|
|
Date.parse(entry.context.realTime) >= Date.parse(server.created_at)
|
|
);
|
|
}
|
|
const start = entries[startIndex];
|
|
// A log restart can arrive before game_start telemetry. Clear immediately
|
|
// instead of falling back to messages from the previous recorded match.
|
|
if (
|
|
start &&
|
|
!current.some((entry) => entry.matchId === match.id) &&
|
|
(!start.context.realTime ||
|
|
Date.parse(start.context.realTime) > Date.parse(match.startedAt) + 1000)
|
|
)
|
|
return [];
|
|
return matchChatTimeline(match.id);
|
|
}
|
|
|
|
const pending = new Map<string, Promise<{ entries: ContextualChatEntry[]; missing: boolean }>>();
|
|
interface ChatReadState {
|
|
key: string;
|
|
cursor?: LogCursor;
|
|
lines: string[];
|
|
partial: Buffer;
|
|
skipLine: boolean;
|
|
signature?: string;
|
|
entries: ContextualChatEntry[];
|
|
contextVersion?: string;
|
|
wasOccupied: boolean;
|
|
needsDrain: boolean;
|
|
bytesRead: number;
|
|
}
|
|
const readers = new Map<string, ChatReadState>();
|
|
const failures = new Map<string, number>();
|
|
const CHAT_READ_MIN_AGE_MS = 15_000;
|
|
const MAX_IDLE_STALLED_READS = 3;
|
|
const idleStalls = new Map<string, number>();
|
|
type LastChatRead = { completedAt: number } & ({ missing: boolean } | { error: unknown });
|
|
const lastReads = new Map<string, LastChatRead>();
|
|
|
|
function chatReadKey(server: ServerRow): string {
|
|
return `${server.name}:${server.created_at}:${server.native_instance_id}:${server.instance_generation}`;
|
|
}
|
|
|
|
/** Live viewers reuse the worker's reader state instead of adding a read per tab. */
|
|
export async function readServerChat(
|
|
server: ServerRow
|
|
): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
|
|
const key = chatReadKey(server);
|
|
const existing = pending.get(key);
|
|
if (existing) return existing;
|
|
const last = lastReads.get(key);
|
|
if (!last || Date.now() - last.completedAt >= CHAT_READ_MIN_AGE_MS)
|
|
return collectServerChat(server);
|
|
if ('error' in last) throw last.error;
|
|
const state = readers.get(server.name);
|
|
return {
|
|
// Telemetry may have started a new match since the last log read.
|
|
entries: currentChatEntries(server, state?.key === key ? state.entries : []),
|
|
missing: last.missing
|
|
};
|
|
}
|
|
|
|
/** Shared by the live API and the worker; collection does not require a viewer. */
|
|
export function collectServerChat(
|
|
server: ServerRow
|
|
): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
|
|
const key = chatReadKey(server);
|
|
const existing = pending.get(key);
|
|
if (existing) return existing;
|
|
let progressed = false;
|
|
const work = (async () => {
|
|
const path = resolveLogPath(server.name, gameLogKey(server.name));
|
|
if (!path) throw new Error('Invalid server log path');
|
|
let saved = readers.get(server.name);
|
|
if (saved?.key !== key) saved = undefined;
|
|
// Publish the new cursor only after its messages have been processed.
|
|
const state: ChatReadState = saved
|
|
? { ...saved }
|
|
: {
|
|
key,
|
|
lines: [],
|
|
partial: Buffer.alloc(0),
|
|
skipLine: false,
|
|
entries: [],
|
|
wasOccupied: false,
|
|
needsDrain: false,
|
|
bytesRead: 0
|
|
};
|
|
const tail = await readLogChunk(path, state.cursor);
|
|
const current = getServer(server.name);
|
|
// A remote read may finish after destruction or replacement.
|
|
if (
|
|
!current ||
|
|
current.destroyed_at ||
|
|
current.created_at !== server.created_at ||
|
|
current.native_instance_id !== server.native_instance_id ||
|
|
current.instance_generation !== server.instance_generation
|
|
) {
|
|
return { entries: [], missing: true };
|
|
}
|
|
if (!tail) {
|
|
readers.delete(server.name);
|
|
return { entries: currentChatEntries(current, []), missing: true };
|
|
}
|
|
let changed = false;
|
|
if ('cursor' in tail) {
|
|
if (tail.reset) {
|
|
state.lines = [];
|
|
state.partial = Buffer.alloc(0);
|
|
state.skipLine = tail.truncated;
|
|
state.entries = [];
|
|
}
|
|
state.cursor = tail.cursor;
|
|
const content = Buffer.from(tail.contentBase64, 'base64');
|
|
progressed = content.length > 0;
|
|
state.bytesRead += content.length;
|
|
const bytes = Buffer.concat([state.partial, content]);
|
|
const end = bytes.lastIndexOf(10) + 1;
|
|
if (end) {
|
|
const lines = bytes
|
|
.subarray(0, end)
|
|
.toString('utf8')
|
|
.split('\n')
|
|
.slice(0, -1)
|
|
.map((line) => line.replace(/\r$/, ''));
|
|
if (state.skipLine) lines.shift();
|
|
state.skipLine = false;
|
|
state.lines = [...state.lines, ...lines].slice(-2000);
|
|
let length = 0,
|
|
start = state.lines.length;
|
|
while (
|
|
start > 0 &&
|
|
(length += Buffer.byteLength(state.lines[start - 1]) + 1) <= 256 * 1024
|
|
)
|
|
start--;
|
|
state.lines = state.lines.slice(start);
|
|
changed = true;
|
|
}
|
|
state.partial = Buffer.from(bytes.subarray(end));
|
|
if (state.partial.length > 256 * 1024) {
|
|
state.partial = Buffer.alloc(0);
|
|
state.skipLine = true;
|
|
}
|
|
// File mtime anchors the final line only after the backlog has been read.
|
|
state.needsDrain = tail.cursor.offset < tail.size || state.partial.length > 0;
|
|
if (state.needsDrain) {
|
|
state.signature = undefined;
|
|
readers.set(server.name, state);
|
|
return { entries: currentChatEntries(current, state.entries), missing: false };
|
|
}
|
|
} else {
|
|
// Older agents remain readable during a rolling update.
|
|
state.needsDrain = false;
|
|
changed = state.signature !== `${tail.size}:${tail.mtime}`;
|
|
progressed = changed;
|
|
if (changed) state.lines = tail.lines;
|
|
}
|
|
const signature = `${tail.size}:${tail.mtime}`;
|
|
const needsContext = state.entries.some(
|
|
(entry) => entry.type === 'message' && entry.matchId === undefined
|
|
);
|
|
const contextVersion = needsContext
|
|
? (
|
|
db()
|
|
.prepare('SELECT received_at FROM server_telemetry WHERE server=?')
|
|
.get(server.name) as { received_at: string } | undefined
|
|
)?.received_at
|
|
: undefined;
|
|
if (
|
|
changed ||
|
|
state.signature === undefined ||
|
|
(needsContext && contextVersion !== state.contextVersion)
|
|
) {
|
|
state.entries = serverChatTimeline(current, { ...tail, lines: state.lines });
|
|
archiveChat(state.entries);
|
|
state.contextVersion = contextVersion;
|
|
}
|
|
state.signature = signature;
|
|
readers.set(server.name, state);
|
|
return { entries: currentChatEntries(current, state.entries), missing: false };
|
|
})()
|
|
.then(
|
|
(result) => {
|
|
const state = readers.get(server.name);
|
|
if (progressed || result.missing || (state?.key === key && !state.needsDrain))
|
|
idleStalls.delete(key);
|
|
lastReads.set(key, { completedAt: Date.now(), missing: result.missing });
|
|
return result;
|
|
},
|
|
(error: unknown) => {
|
|
// Failed reads also observe the cooldown, without advancing the cursor.
|
|
lastReads.set(key, { completedAt: Date.now(), error });
|
|
throw error;
|
|
}
|
|
)
|
|
.finally(() => pending.delete(key));
|
|
pending.set(key, work);
|
|
return work;
|
|
}
|
|
|
|
export async function collectFleetChat(): Promise<void> {
|
|
const hosts = new Map(listHosts().map((host) => [host.id, host]));
|
|
const observe = observationReader();
|
|
const live = listServers().filter((server) => !server.destroyed_at);
|
|
const names = new Set(live.map((server) => server.name));
|
|
const keys = new Set(live.map(chatReadKey));
|
|
for (const name of readers.keys()) if (!names.has(name)) readers.delete(name);
|
|
for (const name of failures.keys()) if (!names.has(name)) failures.delete(name);
|
|
for (const key of lastReads.keys()) if (!keys.has(key)) lastReads.delete(key);
|
|
for (const key of idleStalls.keys()) if (!keys.has(key)) idleStalls.delete(key);
|
|
const servers = live.filter((server) => {
|
|
if (server.host_id && !hosts.get(server.host_id)?.connected) return false;
|
|
if (server.container_state && !['running', 'unknown'].includes(server.container_state))
|
|
return false;
|
|
const players = observe(server).last_players ?? server.last_players;
|
|
const key = chatReadKey(server);
|
|
if (players !== 0) {
|
|
idleStalls.delete(key);
|
|
return true;
|
|
}
|
|
const stalls = idleStalls.get(key) ?? 0;
|
|
if (stalls >= MAX_IDLE_STALLED_READS) return false;
|
|
const last = lastReads.get(key);
|
|
if (stalls > 0 && last && Date.now() - last.completedAt < CHAT_READ_MIN_AGE_MS)
|
|
return false;
|
|
const state = readers.get(server.name);
|
|
// Pick up surviving final logs after a collector restart, then finish any
|
|
// unread chunks even after players leave. Errors and incomplete final lines
|
|
// have bounded idle retries rather than polling an empty server forever.
|
|
return (
|
|
!last ||
|
|
'error' in last ||
|
|
(state?.key === key && (state.wasOccupied || state.needsDrain))
|
|
);
|
|
});
|
|
// Bound agent traffic and isolate failures to their own server.
|
|
async function collect() {
|
|
for (let server = servers.shift(); server; server = servers.shift()) {
|
|
const key = chatReadKey(server);
|
|
const before = readers.get(server.name);
|
|
try {
|
|
await collectServerChat(server);
|
|
const state = readers.get(server.name);
|
|
if (state?.key === key) {
|
|
state.wasOccupied =
|
|
(observe(server).last_players ?? server.last_players ?? 0) > 0;
|
|
if (!state.wasOccupied && state.needsDrain) {
|
|
const progressed =
|
|
before?.key !== key || state.bytesRead > before.bytesRead;
|
|
idleStalls.set(key, progressed ? 0 : (idleStalls.get(key) ?? 0) + 1);
|
|
} else idleStalls.delete(key);
|
|
}
|
|
failures.delete(server.name);
|
|
} catch (error) {
|
|
if ((observe(server).last_players ?? server.last_players) === 0)
|
|
idleStalls.set(key, (idleStalls.get(key) ?? 0) + 1);
|
|
if (Date.now() - (failures.get(server.name) ?? -Infinity) >= 5 * 60_000) {
|
|
console.error(`[chat] Could not collect chat for ${server.name}`, error);
|
|
failures.set(server.name, Date.now());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
await Promise.all(Array.from({ length: Math.min(4, servers.length) }, collect));
|
|
}
|