feat: ATM10-Statistikportal
Statistikportal für den All-the-Mods-10-Server (Minecraft 1.21.1, NeoForge 21.1): - mod/: serverseitige NeoForge-Mod atmstats – Ereignisse, Live-Zähler, Servergesundheit, FTB-Dateien; Outbox auf der Festplatte, signierter gzip-Versand per HTTPS mit Backoff - backend/: Fastify + PostgreSQL – idempotenter Ingest, Projektionen, Parser für SNBT, FTB Quests/Teams, world/stats, world/advancements und latest.log; Ranglisten, Snapshots, Retention, öffentliche API, SSE-Live-Updates, Admin mit argon2/CSRF, SFTP-Worker - web/: Next.js-Dashboard – Übersicht, Ranglisten, Spielerprofile, ATM10, Server, Datenquellen, Verwaltung - shared/: Protokoll v1 und API-Typen; tools/: Simulator, Log-Anonymisierer, RCON-Client - deploy/: Docker Compose hinter Traefik, Backup; docs/: Architektur, Datenquellen, Betrieb Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
lucashahmannandClaude Opus 5.5 committed
1 parent
56d2078e6a
commit
e66f89f49e
214 files changed
+27077
No files matched your search
@@ -0,0 +1,59 @@
|
||||
import cookie from '@fastify/cookie';
|
||||
import helmet from '@fastify/helmet';
|
||||
import rateLimit from '@fastify/rate-limit';
|
||||
import Fastify, { type FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import type { Config } from './config.ts';
|
||||
import { ingestRoutes } from './ingest/route.ts';
|
||||
import { LiveHub } from './live/hub.ts';
|
||||
import { installLiveNotifier } from './live/notifier.ts';
|
||||
import './pipeline/projections/index.ts';
|
||||
import { liveRoutes } from './routes/live.ts';
|
||||
import { adminRoutes } from './routes/admin.ts';
|
||||
import { publicRoutes } from './routes/public/index.ts';
|
||||
|
||||
declare module 'fastify' {
|
||||
interface FastifyInstance {
|
||||
liveHub: LiveHub;
|
||||
}
|
||||
}
|
||||
|
||||
export interface AppDeps {
|
||||
pool: Pool;
|
||||
config: Config;
|
||||
}
|
||||
|
||||
export async function buildApp(deps: AppDeps): Promise<FastifyInstance> {
|
||||
const app = Fastify({
|
||||
logger: {
|
||||
level: deps.config.LOG_LEVEL,
|
||||
redact: ['req.headers.authorization', 'req.headers["x-atm-signature"]', 'req.headers.cookie'],
|
||||
},
|
||||
trustProxy: deps.config.TRUST_PROXY,
|
||||
bodyLimit: 1024 * 1024,
|
||||
});
|
||||
|
||||
await app.register(helmet, { contentSecurityPolicy: false, crossOriginResourcePolicy: { policy: 'same-site' } });
|
||||
await app.register(rateLimit, { global: false });
|
||||
await app.register(cookie);
|
||||
|
||||
const hub = new LiveHub(deps.config.DATABASE_URL, app.log);
|
||||
void hub.start();
|
||||
const uninstallNotifier = installLiveNotifier(deps.pool);
|
||||
app.decorate('liveHub', hub);
|
||||
app.addHook('onClose', async () => {
|
||||
uninstallNotifier();
|
||||
await hub.stop();
|
||||
});
|
||||
|
||||
app.get('/api/health', async () => {
|
||||
await deps.pool.query('SELECT 1');
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
await app.register(ingestRoutes(deps));
|
||||
await app.register(publicRoutes, deps);
|
||||
await app.register(liveRoutes, { hub });
|
||||
await app.register(adminRoutes, deps);
|
||||
return app;
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
import { randomBytes } from 'node:crypto';
|
||||
import argon2 from 'argon2';
|
||||
import type { Pool } from 'pg';
|
||||
|
||||
export const SESSION_TTL_MS = 12 * 3600_000;
|
||||
|
||||
export async function createAdmin(pool: Pool, username: string, password: string): Promise<void> {
|
||||
if (password.length < 12) throw new Error('password must be at least 12 characters');
|
||||
const hash = await argon2.hash(password, { type: argon2.argon2id });
|
||||
await pool.query(
|
||||
`INSERT INTO admin_users(username, password_hash) VALUES ($1, $2)
|
||||
ON CONFLICT (username) DO UPDATE SET password_hash = EXCLUDED.password_hash`,
|
||||
[username, hash],
|
||||
);
|
||||
}
|
||||
|
||||
// Verifying against a dummy hash for unknown users keeps response times uniform (no user enumeration).
|
||||
let dummyHash: Promise<string> | null = null;
|
||||
const getDummyHash = () => (dummyHash ??= argon2.hash(randomBytes(16).toString('hex'), { type: argon2.argon2id }));
|
||||
|
||||
export async function login(pool: Pool, username: string, password: string): Promise<{ sessionId: string; csrf: string } | null> {
|
||||
const { rows } = await pool.query<{ id: number; password_hash: string }>(
|
||||
'SELECT id, password_hash FROM admin_users WHERE username = $1',
|
||||
[username],
|
||||
);
|
||||
const user = rows[0];
|
||||
const ok = await argon2.verify(user?.password_hash ?? (await getDummyHash()), password).catch(() => false);
|
||||
if (!user || !ok) return null;
|
||||
const sessionId = randomBytes(32).toString('base64url');
|
||||
const csrf = randomBytes(24).toString('base64url');
|
||||
await pool.query('DELETE FROM admin_sessions WHERE expires_at < now()');
|
||||
await pool.query(`INSERT INTO admin_sessions(id, user_id, csrf, expires_at) VALUES ($1, $2, $3, now() + make_interval(secs => $4))`, [
|
||||
sessionId,
|
||||
user.id,
|
||||
csrf,
|
||||
SESSION_TTL_MS / 1000,
|
||||
]);
|
||||
return { sessionId, csrf };
|
||||
}
|
||||
|
||||
export async function getSession(pool: Pool, sessionId: string): Promise<{ username: string; csrf: string } | null> {
|
||||
const { rows } = await pool.query<{ username: string; csrf: string }>(
|
||||
`SELECT u.username, s.csrf FROM admin_sessions s JOIN admin_users u ON u.id = s.user_id
|
||||
WHERE s.id = $1 AND s.expires_at > now()`,
|
||||
[sessionId],
|
||||
);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
export async function logout(pool: Pool, sessionId: string): Promise<void> {
|
||||
await pool.query('DELETE FROM admin_sessions WHERE id = $1', [sessionId]);
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
// Verwaltungs-CLI:
|
||||
// cli admin:create <benutzer> Passwort aus ADMIN_PASSWORD oder interaktiv
|
||||
// cli token:create neues Ingest-Token (wird nur einmal angezeigt)
|
||||
// cli migrate Datenbankmigrationen ausführen
|
||||
import { createInterface } from 'node:readline/promises';
|
||||
import { createAdmin } from './auth/admin.ts';
|
||||
import { createPool } from './db/pool.ts';
|
||||
import { migrate } from './db/migrate.ts';
|
||||
import { createIngestToken } from './ingest/tokens.ts';
|
||||
|
||||
async function readPassword(): Promise<string> {
|
||||
if (process.env.ADMIN_PASSWORD) return process.env.ADMIN_PASSWORD;
|
||||
const rl = createInterface({ input: process.stdin, output: process.stdout, terminal: true });
|
||||
// Hide typed characters.
|
||||
(rl as unknown as { _writeToOutput: (s: string) => void })._writeToOutput = (s: string) => {
|
||||
if (s.includes('Passwort')) process.stdout.write(s);
|
||||
};
|
||||
const pw = await rl.question('Passwort (min. 12 Zeichen): ');
|
||||
rl.close();
|
||||
process.stdout.write('\n');
|
||||
return pw;
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const [command, arg] = process.argv.slice(2);
|
||||
const url = process.env.DATABASE_URL;
|
||||
if (!url) throw new Error('DATABASE_URL is required');
|
||||
const pool = createPool(url, 2);
|
||||
try {
|
||||
await migrate(pool);
|
||||
switch (command) {
|
||||
case 'migrate':
|
||||
console.log('Migrationen ausgeführt.');
|
||||
break;
|
||||
case 'admin:create': {
|
||||
if (!arg) throw new Error('usage: admin:create <benutzer>');
|
||||
await createAdmin(pool, arg, await readPassword());
|
||||
console.log(`Admin "${arg}" angelegt bzw. Passwort aktualisiert.`);
|
||||
break;
|
||||
}
|
||||
case 'token:create': {
|
||||
const token = await createIngestToken(pool);
|
||||
console.log('Neues Ingest-Token (in config/atmstats-server.toml als token eintragen):');
|
||||
console.log(token);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
console.log('Befehle: migrate | admin:create <benutzer> | token:create');
|
||||
process.exitCode = 1;
|
||||
}
|
||||
} finally {
|
||||
await pool.end();
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error((err as Error).message);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,46 @@
|
||||
import { z } from 'zod';
|
||||
|
||||
const bool = z
|
||||
.enum(['true', 'false', '1', '0'])
|
||||
.transform((v) => v === 'true' || v === '1');
|
||||
|
||||
const envSchema = z.object({
|
||||
NODE_ENV: z.string().default('production'),
|
||||
DATABASE_URL: z.string({ required_error: 'DATABASE_URL is required' }).min(1, 'DATABASE_URL is required'),
|
||||
PORT: z.coerce.number().int().default(3001),
|
||||
HOST: z.string().default('0.0.0.0'),
|
||||
PUBLIC_ORIGIN: z.string().url().optional(),
|
||||
INGEST_HMAC_SECRET: z.string().min(32, 'INGEST_HMAC_SECRET must be at least 32 characters'),
|
||||
SERVER_ID: z.string().default('atm10'),
|
||||
TRUST_PROXY: bool.default('true'),
|
||||
SFTP_HOST: z.string().optional(),
|
||||
SFTP_PORT: z.coerce.number().int().default(22),
|
||||
SFTP_USER: z.string().optional(),
|
||||
SFTP_PASSWORD: z.string().optional(),
|
||||
SFTP_ROOT: z.string().default('/'),
|
||||
/** level-name of the server (folder containing stats/, advancements/, ftbquests/) */
|
||||
SFTP_WORLD_DIR: z.string().default('world'),
|
||||
SFTP_INTERVAL_MIN: z.coerce.number().int().min(1).default(15),
|
||||
AVATAR_SOURCE: z.string().default('https://mc-heads.net/avatar/{uuid}/64'),
|
||||
/** Time zone of the game server (log timestamps) and of activity heatmaps. */
|
||||
TIMEZONE: z.string().default('Europe/Berlin'),
|
||||
LOG_LEVEL: z.string().default('info'),
|
||||
});
|
||||
|
||||
export type Config = z.infer<typeof envSchema> & {
|
||||
sftp: { host: string; port: number; username: string; password: string; root: string } | null;
|
||||
};
|
||||
|
||||
export function loadConfig(env: Record<string, string | undefined> = process.env): Config {
|
||||
const r = envSchema.safeParse(env);
|
||||
if (!r.success) {
|
||||
const msg = r.error.issues.map((i) => `${i.path.join('.')}: ${i.message}`).join('; ');
|
||||
throw new Error(`Invalid configuration: ${msg}`);
|
||||
}
|
||||
const c = r.data;
|
||||
const sftp =
|
||||
c.SFTP_HOST && c.SFTP_USER && c.SFTP_PASSWORD
|
||||
? { host: c.SFTP_HOST, port: c.SFTP_PORT, username: c.SFTP_USER, password: c.SFTP_PASSWORD, root: c.SFTP_ROOT }
|
||||
: null;
|
||||
return { ...c, sftp };
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import { readdirSync, readFileSync } from 'node:fs';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import type { Pool } from 'pg';
|
||||
|
||||
const MIGRATION_LOCK = 4711;
|
||||
|
||||
function migrationsDir(): string {
|
||||
// src/db -> ../../migrations (dev, tests) ; dist -> ../migrations (bundled build)
|
||||
const here = dirname(fileURLToPath(import.meta.url));
|
||||
for (const candidate of [join(here, '..', '..', 'migrations'), join(here, '..', 'migrations')]) {
|
||||
try {
|
||||
readdirSync(candidate);
|
||||
return candidate;
|
||||
} catch {
|
||||
/* try next */
|
||||
}
|
||||
}
|
||||
throw new Error('migrations directory not found');
|
||||
}
|
||||
|
||||
/** Applies pending `migrations/*.sql` in name order, each in its own transaction, serialized by an advisory lock. */
|
||||
export async function migrate(pool: Pool): Promise<string[]> {
|
||||
const dir = migrationsDir();
|
||||
const files = readdirSync(dir).filter((f) => f.endsWith('.sql')).sort();
|
||||
const client = await pool.connect();
|
||||
const applied: string[] = [];
|
||||
try {
|
||||
await client.query('SELECT pg_advisory_lock($1)', [MIGRATION_LOCK]);
|
||||
await client.query(
|
||||
'CREATE TABLE IF NOT EXISTS schema_migrations (name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())',
|
||||
);
|
||||
const { rows } = await client.query<{ name: string }>('SELECT name FROM schema_migrations');
|
||||
const done = new Set(rows.map((r) => r.name));
|
||||
for (const file of files) {
|
||||
if (done.has(file)) continue;
|
||||
const sql = readFileSync(join(dir, file), 'utf8');
|
||||
await client.query('BEGIN');
|
||||
try {
|
||||
await client.query(sql);
|
||||
await client.query('INSERT INTO schema_migrations(name) VALUES ($1)', [file]);
|
||||
await client.query('COMMIT');
|
||||
applied.push(file);
|
||||
} catch (err) {
|
||||
await client.query('ROLLBACK');
|
||||
throw new Error(`migration ${file} failed: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await client.query('SELECT pg_advisory_unlock($1)', [MIGRATION_LOCK]).catch(() => {});
|
||||
client.release();
|
||||
}
|
||||
return applied;
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import pg from 'pg';
|
||||
|
||||
// bigint (int8) columns hold Minecraft counters; they stay far below 2^53, so parse as number.
|
||||
pg.types.setTypeParser(20, (v) => Number(v));
|
||||
// numeric is used for aggregates (SUM over bigint)
|
||||
pg.types.setTypeParser(1700, (v) => Number(v));
|
||||
|
||||
export type Pool = pg.Pool;
|
||||
export type PoolClient = pg.PoolClient;
|
||||
|
||||
export function createPool(url: string, max = 10): pg.Pool {
|
||||
return new pg.Pool({ connectionString: url, max });
|
||||
}
|
||||
|
||||
export async function withTx<T>(pool: pg.Pool, fn: (c: pg.PoolClient) => Promise<T>): Promise<T> {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
const result = await fn(client);
|
||||
await client.query('COMMIT');
|
||||
return result;
|
||||
} catch (err) {
|
||||
await client.query('ROLLBACK').catch(() => {});
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
import { createHash, createHmac, timingSafeEqual } from 'node:crypto';
|
||||
|
||||
export const MAX_CLOCK_SKEW_MS = 5 * 60_000;
|
||||
|
||||
export type IngestAuthError = 'missing_auth' | 'bad_token' | 'timestamp_out_of_range' | 'bad_signature';
|
||||
export type IngestAuthResult = { ok: true } | { ok: false; status: 401; error: IngestAuthError };
|
||||
|
||||
export interface VerifyOpts {
|
||||
secret: string;
|
||||
now: number;
|
||||
tokenValid: (token: string) => Promise<boolean>;
|
||||
}
|
||||
|
||||
export function computeSignature(secret: string, ts: number | string, body: Buffer): string {
|
||||
const bodyHash = createHash('sha256').update(body).digest('hex');
|
||||
return createHmac('sha256', secret).update(`${ts}.${bodyHash}`).digest('hex');
|
||||
}
|
||||
|
||||
const fail = (error: IngestAuthError): IngestAuthResult => ({ ok: false, status: 401, error });
|
||||
|
||||
/** Verifies bearer token, timestamp window and HMAC over the bytes as transmitted (before gunzip). */
|
||||
export async function verifyIngest(
|
||||
req: { headers: Record<string, string | string[] | undefined>; rawBody: Buffer },
|
||||
opts: VerifyOpts,
|
||||
): Promise<IngestAuthResult> {
|
||||
const header = (n: string) => {
|
||||
const v = req.headers[n];
|
||||
return Array.isArray(v) ? v[0] : v;
|
||||
};
|
||||
const auth = header('authorization');
|
||||
const tsHeader = header('x-atm-timestamp');
|
||||
const sig = header('x-atm-signature');
|
||||
if (!auth?.startsWith('Bearer ') || !tsHeader || !sig) return fail('missing_auth');
|
||||
|
||||
if (!(await opts.tokenValid(auth.slice(7).trim()))) return fail('bad_token');
|
||||
|
||||
const ts = Number(tsHeader);
|
||||
if (!Number.isFinite(ts) || Math.abs(opts.now - ts) > MAX_CLOCK_SKEW_MS) return fail('timestamp_out_of_range');
|
||||
|
||||
const expected = Buffer.from(computeSignature(opts.secret, tsHeader, req.rawBody), 'hex');
|
||||
const given = Buffer.from(sig, 'hex');
|
||||
if (given.length !== expected.length || !timingSafeEqual(given, expected)) return fail('bad_signature');
|
||||
return { ok: true };
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
import { gunzipSync } from 'node:zlib';
|
||||
import { ingestBatchSchema } from '@atm/shared';
|
||||
import type { FastifyInstance, FastifyPluginAsync } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import type { Config } from '../config.ts';
|
||||
import { processEvents, validateEvents } from '../pipeline/process.ts';
|
||||
import { verifyIngest } from './auth.ts';
|
||||
import { isValidIngestToken } from './tokens.ts';
|
||||
|
||||
export const MAX_UNPACKED_BYTES = 2 * 1024 * 1024;
|
||||
|
||||
export function ingestRoutes(deps: { pool: Pool; config: Config }): FastifyPluginAsync {
|
||||
return async (app: FastifyInstance) => {
|
||||
// Raw bytes are needed for the signature; parse JSON ourselves after verification.
|
||||
app.addContentTypeParser('application/json', { parseAs: 'buffer', bodyLimit: MAX_UNPACKED_BYTES }, (_req, body, done) =>
|
||||
done(null, body),
|
||||
);
|
||||
app.removeContentTypeParser(['text/plain']);
|
||||
|
||||
const authenticate = (headers: Record<string, string | string[] | undefined>, rawBody: Buffer) =>
|
||||
verifyIngest(
|
||||
{ headers, rawBody },
|
||||
{
|
||||
secret: deps.config.INGEST_HMAC_SECRET,
|
||||
now: Date.now(),
|
||||
tokenValid: (t) => isValidIngestToken(deps.pool, t),
|
||||
},
|
||||
);
|
||||
|
||||
app.get('/api/ingest/ping', async (req, reply) => {
|
||||
const auth = await authenticate(req.headers, Buffer.alloc(0));
|
||||
if (!auth.ok) return reply.code(auth.status).send({ error: auth.error });
|
||||
return { ok: true, serverTime: Date.now(), protocol: 1 };
|
||||
});
|
||||
|
||||
app.post(
|
||||
'/api/ingest',
|
||||
{ config: { rateLimit: { max: 120, timeWindow: '1 minute' } } },
|
||||
async (req, reply) => {
|
||||
const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
|
||||
const auth = await authenticate(req.headers, raw);
|
||||
if (!auth.ok) {
|
||||
req.log.warn({ error: auth.error }, 'ingest rejected');
|
||||
return reply.code(auth.status).send({ error: auth.error });
|
||||
}
|
||||
|
||||
let text: string;
|
||||
try {
|
||||
const encoding = String(req.headers['content-encoding'] ?? '').toLowerCase();
|
||||
const unpacked = encoding === 'gzip' ? gunzipSync(raw, { maxOutputLength: MAX_UNPACKED_BYTES }) : raw;
|
||||
if (unpacked.length > MAX_UNPACKED_BYTES) return reply.code(413).send({ error: 'payload_too_large' });
|
||||
text = unpacked.toString('utf8');
|
||||
} catch (err) {
|
||||
if ((err as NodeJS.ErrnoException).code === 'ERR_BUFFER_TOO_LARGE') {
|
||||
return reply.code(413).send({ error: 'payload_too_large' });
|
||||
}
|
||||
return reply.code(400).send({ error: 'bad_encoding' });
|
||||
}
|
||||
|
||||
let json: unknown;
|
||||
try {
|
||||
json = JSON.parse(text);
|
||||
} catch {
|
||||
return reply.code(400).send({ error: 'bad_json' });
|
||||
}
|
||||
const batch = ingestBatchSchema.safeParse(json);
|
||||
if (!batch.success) return reply.code(400).send({ error: 'bad_batch', detail: batch.error.issues[0]?.message });
|
||||
|
||||
const { valid, rejected } = validateEvents(batch.data.events);
|
||||
const serverId = String(req.headers['x-atm-server'] ?? deps.config.SERVER_ID).slice(0, 64);
|
||||
const result = valid.length
|
||||
? await processEvents(deps.pool, valid, 'push', serverId)
|
||||
: { accepted: [], duplicate: [], rejected: [] };
|
||||
if (rejected.length) req.log.warn({ rejected: rejected.length }, 'ingest: invalid events rejected');
|
||||
return { ...result, rejected };
|
||||
},
|
||||
);
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
import { createHash, randomBytes } from 'node:crypto';
|
||||
import type { Pool } from 'pg';
|
||||
|
||||
const hash = (token: string) => createHash('sha256').update(token).digest('hex');
|
||||
|
||||
/** Creates a new ingest token. The plain value is returned once and only its hash is stored. */
|
||||
export async function createIngestToken(pool: Pool): Promise<string> {
|
||||
const token = randomBytes(32).toString('base64url');
|
||||
await pool.query('INSERT INTO ingest_tokens(hash) VALUES ($1)', [hash(token)]);
|
||||
return token;
|
||||
}
|
||||
|
||||
export async function isValidIngestToken(pool: Pool, token: string): Promise<boolean> {
|
||||
const { rowCount } = await pool.query('SELECT 1 FROM ingest_tokens WHERE hash = $1 AND revoked_at IS NULL', [
|
||||
hash(token),
|
||||
]);
|
||||
return (rowCount ?? 0) > 0;
|
||||
}
|
||||
|
||||
export async function revokeAllIngestTokens(pool: Pool): Promise<void> {
|
||||
await pool.query('UPDATE ingest_tokens SET revoked_at = now() WHERE revoked_at IS NULL');
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
import type { LiveEvent } from '@atm/shared/api';
|
||||
import pg from 'pg';
|
||||
import { LIVE_CHANNEL, type LiveMessage } from './notifier.ts';
|
||||
|
||||
export const RING_SIZE = 500;
|
||||
|
||||
/** Listens on the Postgres live channel and fans messages out to SSE subscribers, keeping a ring buffer for resume. */
|
||||
export class LiveHub {
|
||||
private ring: LiveEvent[] = [];
|
||||
private nextId = 1;
|
||||
private subscribers = new Set<(e: LiveEvent) => void>();
|
||||
private client: pg.Client | null = null;
|
||||
private stopped = false;
|
||||
private readyPromise: Promise<void>;
|
||||
private resolveReady!: () => void;
|
||||
|
||||
constructor(
|
||||
private readonly connectionString: string,
|
||||
private readonly log: { warn: (o: object, msg: string) => void } = console,
|
||||
) {
|
||||
this.readyPromise = new Promise((r) => (this.resolveReady = r));
|
||||
}
|
||||
|
||||
ready(): Promise<void> {
|
||||
return this.readyPromise;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
let attempt = 0;
|
||||
while (!this.stopped) {
|
||||
try {
|
||||
const client = new pg.Client({ connectionString: this.connectionString });
|
||||
client.on('notification', (n) => {
|
||||
if (n.channel === LIVE_CHANNEL && n.payload) this.publish(JSON.parse(n.payload) as LiveMessage);
|
||||
});
|
||||
await client.connect();
|
||||
await client.query(`LISTEN ${LIVE_CHANNEL}`);
|
||||
this.client = client;
|
||||
attempt = 0;
|
||||
this.resolveReady();
|
||||
await new Promise<void>((resolve) => {
|
||||
client.on('error', () => resolve());
|
||||
client.on('end', () => resolve());
|
||||
});
|
||||
} catch (err) {
|
||||
this.log.warn({ err }, 'live hub: connection failed');
|
||||
}
|
||||
this.client = null;
|
||||
if (this.stopped) return;
|
||||
await new Promise((r) => setTimeout(r, Math.min(30_000, 1000 * 2 ** attempt++)));
|
||||
}
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
await this.client?.end().catch(() => {});
|
||||
this.subscribers.clear();
|
||||
}
|
||||
|
||||
publish(msg: LiveMessage): LiveEvent {
|
||||
const event: LiveEvent = { id: this.nextId++, kind: msg.kind, data: msg.data };
|
||||
this.ring.push(event);
|
||||
if (this.ring.length > RING_SIZE) this.ring.shift();
|
||||
for (const fn of this.subscribers) fn(event);
|
||||
return event;
|
||||
}
|
||||
|
||||
/** Events with id > lastId still in the ring buffer. */
|
||||
since(lastId: number): LiveEvent[] {
|
||||
return this.ring.filter((e) => e.id > lastId);
|
||||
}
|
||||
|
||||
lastId(): number {
|
||||
return this.nextId - 1;
|
||||
}
|
||||
|
||||
subscribe(fn: (e: LiveEvent) => void): () => void {
|
||||
this.subscribers.add(fn);
|
||||
return () => this.subscribers.delete(fn);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
import type { AtmEvent } from '@atm/shared';
|
||||
import type { LiveKind } from '@atm/shared/api';
|
||||
import type { Pool } from 'pg';
|
||||
import { onEventsCommitted } from '../pipeline/process.ts';
|
||||
|
||||
export const LIVE_CHANNEL = 'atm_live';
|
||||
const COUNTERS_THROTTLE_MS = 10_000;
|
||||
|
||||
export interface LiveMessage {
|
||||
kind: LiveKind;
|
||||
data: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Maps committed events to live messages. Pure; hidden players are filtered by the caller. */
|
||||
export function toLiveMessages(events: AtmEvent[]): (LiveMessage & { uuid?: string })[] {
|
||||
const out: (LiveMessage & { uuid?: string })[] = [];
|
||||
for (const e of events) {
|
||||
switch (e.type) {
|
||||
case 'player_join':
|
||||
case 'player_leave':
|
||||
out.push({ kind: 'online', uuid: e.uuid, data: { uuid: e.uuid, name: e.name, online: e.type === 'player_join', ts: e.ts } });
|
||||
break;
|
||||
case 'player_death':
|
||||
out.push({
|
||||
kind: 'death',
|
||||
uuid: e.uuid,
|
||||
data: { uuid: e.uuid, name: e.name, cause: e.cause, killerType: e.killerType, dimension: e.dimension, ts: e.ts },
|
||||
});
|
||||
break;
|
||||
case 'advancement':
|
||||
if (e.title) out.push({ kind: 'advancement', uuid: e.uuid, data: { uuid: e.uuid, name: e.name, title: e.title, ts: e.ts } });
|
||||
break;
|
||||
case 'file_sync':
|
||||
if (e.path.startsWith('world/ftbquests/')) out.push({ kind: 'quest', data: { ts: e.ts } });
|
||||
break;
|
||||
case 'stats_snapshot':
|
||||
out.push({ kind: 'counters', data: { ts: e.ts } });
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Publishes live messages via Postgres NOTIFY after each committed batch, so every API instance
|
||||
* (and events processed by the worker) reaches all SSE clients. Returns an uninstall function.
|
||||
*/
|
||||
export function installLiveNotifier(pool: Pool): () => void {
|
||||
let lastCounters = 0;
|
||||
return onEventsCommitted(async (events) => {
|
||||
let msgs = toLiveMessages(events);
|
||||
if (msgs.length === 0) return;
|
||||
const uuids = [...new Set(msgs.flatMap((m) => (m.uuid ? [m.uuid] : [])))];
|
||||
if (uuids.length) {
|
||||
const { rows } = await pool.query<{ uuid: string }>('SELECT uuid FROM players WHERE hidden AND uuid = ANY($1::uuid[])', [uuids]);
|
||||
const hidden = new Set(rows.map((r) => r.uuid));
|
||||
msgs = msgs.filter((m) => !m.uuid || !hidden.has(m.uuid));
|
||||
}
|
||||
const now = Date.now();
|
||||
let countersSent = false;
|
||||
for (const { kind, data } of msgs) {
|
||||
if (kind === 'counters') {
|
||||
if (countersSent || now - lastCounters < COUNTERS_THROTTLE_MS) continue;
|
||||
countersSent = true;
|
||||
lastCounters = now;
|
||||
}
|
||||
await pool.query('SELECT pg_notify($1, $2)', [LIVE_CHANNEL, JSON.stringify({ kind, data })]);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
import { isCompound, parseSnbt, type SnbtValue } from './snbt.ts';
|
||||
|
||||
/** "308a693c2bd84d6b8790670aef5be601" → "308a693c-2bd8-4d6b-8790-670aef5be601" */
|
||||
export function normalizeUuid(raw: string): string {
|
||||
const hex = raw.replace(/-/g, '').toLowerCase();
|
||||
if (!/^[0-9a-f]{32}$/.test(hex)) throw new Error(`not a uuid: ${raw}`);
|
||||
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`;
|
||||
}
|
||||
|
||||
export interface QuestProgress {
|
||||
teamId: string;
|
||||
name: string | null;
|
||||
/** IDs from FTB's `completed` map (quests and chapters) with completion time in ms. */
|
||||
completed: { questId: string; at: number }[];
|
||||
started: number;
|
||||
}
|
||||
|
||||
/** Parses `world/ftbquests/<team>.snbt` (FTB Quests 2101.x team data). */
|
||||
export function parseQuestProgress(text: string): QuestProgress {
|
||||
const root = parseSnbt(text);
|
||||
if (!isCompound(root) || typeof root.uuid !== 'string') throw new Error('quest progress: missing uuid');
|
||||
const completedMap = isCompound(root.completed) ? root.completed : {};
|
||||
const completed = Object.entries(completedMap)
|
||||
.filter(([id, at]) => /^[0-9A-F]{16}$/i.test(id) && typeof at === 'number')
|
||||
.map(([id, at]) => ({ questId: id.toUpperCase(), at: at as number }));
|
||||
return {
|
||||
teamId: normalizeUuid(root.uuid),
|
||||
name: typeof root.name === 'string' ? root.name : null,
|
||||
completed,
|
||||
started: isCompound(root.started) ? Object.keys(root.started).length : 0,
|
||||
};
|
||||
}
|
||||
|
||||
export interface QuestDef {
|
||||
chapterId: string;
|
||||
chapterTitle: string | null;
|
||||
questId: string;
|
||||
title: string | null;
|
||||
}
|
||||
|
||||
const str = (v: SnbtValue | undefined): string | null => (typeof v === 'string' && v.length > 0 ? v : null);
|
||||
|
||||
/**
|
||||
* Removes Minecraft formatting codes ("&a", "&6&l", "§r") used in ATM10 quest texts. FTB writes a literal
|
||||
* ampersand as "\&" (e.g. "Forbidden \& Arcanus"); that one is kept as "&".
|
||||
*/
|
||||
export function stripFormatting(s: string): string {
|
||||
return s
|
||||
.split('\\&')
|
||||
.map((part) => part.replace(/[&§][0-9a-fk-or]/gi, ''))
|
||||
.join('&')
|
||||
.replace(/\s{2,}/g, ' ')
|
||||
.trim();
|
||||
}
|
||||
|
||||
const strip = (s: string | null) => (s ? stripFormatting(s) : null);
|
||||
|
||||
const EN_LANG = /\/lang\/en_us(\.snbt|\/.+\.snbt_merged)$/;
|
||||
const CHAPTER_FILE = /\/chapters\/[^/]+\.snbt$/;
|
||||
|
||||
/**
|
||||
* Quest config files worth transferring: chapter definitions and the English texts split per chapter by
|
||||
* ftbquestslangsplitter (lang/en_us/**\/*.snbt_merged, verified on the ATM10 server). Skipped: other
|
||||
* languages (~35 MB), reward tables and the combined lang/en_us.snbt (1.5 MB, same content as the split files).
|
||||
*/
|
||||
export function isQuestConfigFile(path: string): boolean {
|
||||
return /^config\/ftbquests\/quests\/(chapters\/[^/]+\.snbt|lang\/en_us\/.+\.snbt_merged)$/.test(path);
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds quest definitions from `config/ftbquests/quests/chapters/*.snbt`. Titles come from the English lang
|
||||
* files (`chapter.<id>.title`, `quest.<id>.title`; split `lang/en_us/**\/*.snbt_merged` or a single
|
||||
* `lang/en_us.snbt`), else from inline `title` fields; chapters fall back to their filename.
|
||||
* Quests without a title key show their task item in game; that name is not available server-side (null).
|
||||
*/
|
||||
export function parseQuestDefs(files: { path: string; content: string }[]): QuestDef[] {
|
||||
const lang = new Map<string, string>();
|
||||
for (const f of files) {
|
||||
if (!EN_LANG.test(f.path)) continue;
|
||||
const root = parseSnbt(f.content);
|
||||
if (!isCompound(root)) continue;
|
||||
for (const [k, v] of Object.entries(root)) {
|
||||
const raw = Array.isArray(v) ? v.filter((x) => typeof x === 'string').join(' ') : str(v);
|
||||
const text = raw ? stripFormatting(raw) : null;
|
||||
if (text) lang.set(k, text);
|
||||
}
|
||||
}
|
||||
const defs: QuestDef[] = [];
|
||||
const chapters = files.filter((f) => CHAPTER_FILE.test(f.path)).sort((a, b) => a.path.localeCompare(b.path));
|
||||
for (const f of chapters) {
|
||||
const root = parseSnbt(f.content);
|
||||
if (!isCompound(root)) continue;
|
||||
const chapterId = str(root.id);
|
||||
if (!chapterId) continue;
|
||||
const chapterTitle = lang.get(`chapter.${chapterId}.title`) ?? strip(str(root.title)) ?? str(root.filename);
|
||||
for (const q of Array.isArray(root.quests) ? root.quests : []) {
|
||||
if (!isCompound(q)) continue;
|
||||
const questId = str(q.id);
|
||||
if (!questId) continue;
|
||||
defs.push({
|
||||
chapterId,
|
||||
chapterTitle,
|
||||
questId: questId.toUpperCase(),
|
||||
title: lang.get(`quest.${questId}.title`) ?? strip(str(q.title)),
|
||||
});
|
||||
}
|
||||
}
|
||||
return defs;
|
||||
}
|
||||
|
||||
export interface TeamInfo {
|
||||
teamId: string;
|
||||
type: 'player' | 'party' | 'server' | string;
|
||||
name: string | null;
|
||||
/** players with rank member, officer or owner (not invited players or allies) */
|
||||
members: string[];
|
||||
}
|
||||
|
||||
const MEMBER_RANKS = new Set(['member', 'officer', 'owner']);
|
||||
|
||||
/**
|
||||
* Parses `world/ftbteams/{player,party}/<id>.snbt` (FTB Teams 2101.x). Verified on the real server for player
|
||||
* teams; party files use the same fields (`id`, `type`, `ranks`, `properties`).
|
||||
*/
|
||||
export function parseTeam(text: string): TeamInfo {
|
||||
const root = parseSnbt(text);
|
||||
if (!isCompound(root) || typeof root.id !== 'string') throw new Error('team: missing id');
|
||||
const props = isCompound(root.properties) ? root.properties : {};
|
||||
const ranks = isCompound(root.ranks) ? root.ranks : {};
|
||||
return {
|
||||
teamId: normalizeUuid(root.id),
|
||||
type: typeof root.type === 'string' ? root.type : 'unknown',
|
||||
name: str(props['ftbteams:display_name']) ?? str(root.player_name),
|
||||
members: Object.entries(ranks)
|
||||
.filter(([, rank]) => typeof rank === 'string' && MEMBER_RANKS.has(rank))
|
||||
.map(([uuid]) => normalizeUuid(uuid)),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { deterministicId, type AtmEvent } from '@atm/shared';
|
||||
|
||||
const MONTHS: Record<string, number> = {
|
||||
Jan: 0, Feb: 1, Mar: 2, Apr: 3, May: 4, Jun: 5, Jul: 6, Aug: 7, Sep: 8, Oct: 9, Nov: 10, Dec: 11,
|
||||
};
|
||||
|
||||
function tzOffsetMs(utcMs: number, timeZone: string): number {
|
||||
const parts = new Intl.DateTimeFormat('en-US', {
|
||||
timeZone,
|
||||
hourCycle: 'h23',
|
||||
year: 'numeric',
|
||||
month: '2-digit',
|
||||
day: '2-digit',
|
||||
hour: '2-digit',
|
||||
minute: '2-digit',
|
||||
second: '2-digit',
|
||||
}).formatToParts(new Date(utcMs));
|
||||
const get = (t: string) => Number(parts.find((p) => p.type === t)!.value);
|
||||
const asUtc = Date.UTC(get('year'), get('month') - 1, get('day'), get('hour'), get('minute'), get('second'));
|
||||
return asUtc - Math.floor(utcMs / 1000) * 1000;
|
||||
}
|
||||
|
||||
/** "12Oct2026 09:03:03.006" in the server's local time zone → epoch ms. */
|
||||
export function parseLogTimestamp(s: string, timeZone: string): number {
|
||||
const m = /^(\d{2})([A-Z][a-z]{2})(\d{4}) (\d{2}):(\d{2}):(\d{2})\.(\d{3})$/.exec(s);
|
||||
if (!m || MONTHS[m[2]!] === undefined) return Number.NaN;
|
||||
const local = Date.UTC(+m[3]!, MONTHS[m[2]!]!, +m[1]!, +m[4]!, +m[5]!, +m[6]!, +m[7]!);
|
||||
let utc = local - tzOffsetMs(local, timeZone);
|
||||
utc = local - tzOffsetMs(utc, timeZone); // second pass settles DST boundaries
|
||||
return utc;
|
||||
}
|
||||
|
||||
const LINE = /^\[([^\]]+)\] \[[^\]]+\] \[([^\]]+)\]: (.*)$/;
|
||||
const NAME = '([A-Za-z0-9_]{1,16})';
|
||||
const UUID_OF = new RegExp(`^UUID of player ${NAME} is ([0-9a-f-]{36})$`);
|
||||
const JOINED = new RegExp(`^${NAME} joined the game$`);
|
||||
const LEFT = new RegExp(`^${NAME} left the game$`);
|
||||
const ADVANCEMENT = new RegExp(`^${NAME} has (?:made the advancement|reached the goal|completed the challenge) \\[(.+)\\]$`);
|
||||
const SYSTEM_LOGGER = 'net.minecraft.server.MinecraftServer/';
|
||||
|
||||
/**
|
||||
* Parses complete lines of `logs/latest.log`. Only bytes up to the last newline are consumed, so a line that is
|
||||
* still being written is read again next time. `names` maps player name → uuid and is updated from
|
||||
* "UUID of player" lines; lines for players without known uuid are skipped.
|
||||
* Event IDs derive from the absolute byte offset and line content, so re-reading yields the same IDs.
|
||||
*/
|
||||
export function parseLogChunk(
|
||||
chunk: Buffer,
|
||||
baseOffset: number,
|
||||
names: Map<string, string>,
|
||||
timeZone: string,
|
||||
): { events: AtmEvent[]; consumed: number } {
|
||||
const end = chunk.lastIndexOf(0x0a);
|
||||
if (end < 0) return { events: [], consumed: 0 };
|
||||
const events: AtmEvent[] = [];
|
||||
let offset = 0;
|
||||
while (offset <= end) {
|
||||
const nl = chunk.indexOf(0x0a, offset);
|
||||
const line = chunk.toString('utf8', offset, nl).replace(/\r$/, '');
|
||||
const lineOffset = baseOffset + offset;
|
||||
offset = nl + 1;
|
||||
|
||||
const m = LINE.exec(line);
|
||||
if (!m) continue;
|
||||
const [, time, logger, msg] = m as unknown as [string, string, string, string];
|
||||
|
||||
const uuidOf = UUID_OF.exec(msg);
|
||||
if (uuidOf) {
|
||||
names.set(uuidOf[1]!, uuidOf[2]!);
|
||||
continue;
|
||||
}
|
||||
if (logger !== SYSTEM_LOGGER) continue;
|
||||
|
||||
const id = () => deterministicId(`log:${lineOffset}:${createHash('sha256').update(line).digest('hex')}`);
|
||||
const ts = parseLogTimestamp(time, timeZone);
|
||||
if (!Number.isFinite(ts)) continue;
|
||||
|
||||
let hit: RegExpExecArray | null;
|
||||
if ((hit = JOINED.exec(msg)) || (hit = LEFT.exec(msg))) {
|
||||
const name = hit[1]!;
|
||||
const uuid = names.get(name);
|
||||
if (!uuid) continue;
|
||||
events.push({ id: id(), type: JOINED.test(msg) ? 'player_join' : 'player_leave', ts, uuid, name });
|
||||
} else if ((hit = ADVANCEMENT.exec(msg))) {
|
||||
const name = hit[1]!;
|
||||
const uuid = names.get(name);
|
||||
if (!uuid) continue;
|
||||
events.push({ id: id(), type: 'advancement', ts, uuid, name, advancementId: `log:${hit[2]}`, title: hit[2]! });
|
||||
}
|
||||
}
|
||||
return { events, consumed: end + 1 };
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
/**
|
||||
* Tolerant SNBT parser for FTB files. FTB writes "pretty" SNBT: entries separated by newlines instead of commas,
|
||||
* unquoted keys that may start with digits (team/quest IDs) or contain dots (lang keys).
|
||||
* Numbers lose their type suffix (b, s, L, f, d) and become JS numbers.
|
||||
*/
|
||||
export type SnbtValue = string | number | boolean | SnbtValue[] | { [key: string]: SnbtValue };
|
||||
|
||||
export class SnbtError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
readonly line: number,
|
||||
readonly column: number,
|
||||
) {
|
||||
super(`${message} at line ${line}, column ${column}`);
|
||||
}
|
||||
}
|
||||
|
||||
const NUMBER = /^[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?[bBsSlLfFdD]?$/;
|
||||
const BARE = /[A-Za-z0-9_\-.+]/;
|
||||
|
||||
export function parseSnbt(text: string): SnbtValue {
|
||||
let pos = 0;
|
||||
|
||||
const error = (msg: string): never => {
|
||||
const before = text.slice(0, pos);
|
||||
const line = before.split('\n').length;
|
||||
const column = pos - before.lastIndexOf('\n');
|
||||
throw new SnbtError(msg, line, column);
|
||||
};
|
||||
|
||||
/** Skips whitespace and commas (commas are optional separators in FTB SNBT). */
|
||||
const skipSeparators = () => {
|
||||
while (pos < text.length && /[\s,]/.test(text[pos]!)) pos++;
|
||||
};
|
||||
const skipWs = () => {
|
||||
while (pos < text.length && /\s/.test(text[pos]!)) pos++;
|
||||
};
|
||||
|
||||
const quoted = (): string => {
|
||||
const q = text[pos++];
|
||||
let out = '';
|
||||
while (pos < text.length) {
|
||||
const ch = text[pos++]!;
|
||||
if (ch === '\\') {
|
||||
if (pos >= text.length) break;
|
||||
const next = text[pos++]!;
|
||||
out += next === 'n' ? '\n' : next === 't' ? '\t' : next;
|
||||
} else if (ch === q) {
|
||||
return out;
|
||||
} else {
|
||||
out += ch;
|
||||
}
|
||||
}
|
||||
return error('unterminated string');
|
||||
};
|
||||
|
||||
const bare = (): string => {
|
||||
const start = pos;
|
||||
while (pos < text.length && BARE.test(text[pos]!)) pos++;
|
||||
if (pos === start) error(`unexpected character '${text[pos] ?? 'EOF'}'`);
|
||||
return text.slice(start, pos);
|
||||
};
|
||||
|
||||
const key = (): string => (text[pos] === '"' || text[pos] === "'" ? quoted() : bare());
|
||||
|
||||
const compound = (): SnbtValue => {
|
||||
pos++; // {
|
||||
const obj: { [k: string]: SnbtValue } = {};
|
||||
for (;;) {
|
||||
skipSeparators();
|
||||
if (pos >= text.length) error('unterminated compound');
|
||||
if (text[pos] === '}') {
|
||||
pos++;
|
||||
return obj;
|
||||
}
|
||||
const k = key();
|
||||
skipWs();
|
||||
if (text[pos] !== ':') error(`expected ':' after key '${k}'`);
|
||||
pos++;
|
||||
obj[k] = value();
|
||||
}
|
||||
};
|
||||
|
||||
const list = (): SnbtValue => {
|
||||
pos++; // [
|
||||
skipWs();
|
||||
// typed arrays: [I; 1, 2]
|
||||
if (/[IBL]/.test(text[pos] ?? '') && text[pos + 1] === ';') pos += 2;
|
||||
const arr: SnbtValue[] = [];
|
||||
for (;;) {
|
||||
skipSeparators();
|
||||
if (pos >= text.length) error('unterminated list');
|
||||
if (text[pos] === ']') {
|
||||
pos++;
|
||||
return arr;
|
||||
}
|
||||
arr.push(value());
|
||||
}
|
||||
};
|
||||
|
||||
const value = (): SnbtValue => {
|
||||
skipWs();
|
||||
const ch = text[pos];
|
||||
if (ch === undefined) return error('unexpected end of input');
|
||||
if (ch === '{') return compound();
|
||||
if (ch === '[') return list();
|
||||
if (ch === '"' || ch === "'") return quoted();
|
||||
const word = bare();
|
||||
if (word === 'true') return true;
|
||||
if (word === 'false') return false;
|
||||
if (NUMBER.test(word)) return Number(/[bBsSlLfFdD]$/.test(word) ? word.slice(0, -1) : word);
|
||||
return word;
|
||||
};
|
||||
|
||||
const result = value();
|
||||
skipWs();
|
||||
if (pos < text.length) error('trailing content');
|
||||
return result;
|
||||
}
|
||||
|
||||
export const isCompound = (v: SnbtValue | undefined): v is { [key: string]: SnbtValue } =>
|
||||
typeof v === 'object' && v !== null && !Array.isArray(v);
|
||||
@@ -0,0 +1,56 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { deterministicId, type AtmEvent } from '@atm/shared';
|
||||
|
||||
const sha256 = (s: string) => createHash('sha256').update(s).digest('hex');
|
||||
|
||||
/** `world/stats/<uuid>.json` → absolute stats_snapshot observed at the file's mtime. */
|
||||
export function statsFileToEvent(uuid: string, json: string, mtimeMs: number): AtmEvent {
|
||||
const root = JSON.parse(json) as { stats?: Record<string, Record<string, number>> };
|
||||
const stats: Record<string, Record<string, number>> = {};
|
||||
for (const [cat, entries] of Object.entries(root.stats ?? {})) {
|
||||
const clean: Record<string, number> = {};
|
||||
for (const [k, v] of Object.entries(entries)) if (Number.isInteger(v) && v >= 0) clean[k] = v;
|
||||
stats[cat] = clean;
|
||||
}
|
||||
return {
|
||||
id: deterministicId(`stats:${uuid}:${mtimeMs}:${sha256(json)}`),
|
||||
type: 'stats_snapshot',
|
||||
ts: mtimeMs,
|
||||
uuid,
|
||||
name: null,
|
||||
full: true,
|
||||
stats,
|
||||
};
|
||||
}
|
||||
|
||||
/** "2026-10-12 12:42:23 +0200" → epoch ms */
|
||||
export function parseAdvancementTime(s: string): number {
|
||||
const m = /^(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}) ([+-])(\d{2})(\d{2})$/.exec(s);
|
||||
if (!m) return Number.NaN;
|
||||
return Date.parse(`${m[1]}T${m[2]}${m[3]}${m[4]}:${m[5]}`);
|
||||
}
|
||||
|
||||
/** `world/advancements/<uuid>.json` → one advancement event per completed, non-recipe advancement. */
|
||||
export function advancementsFileToEvents(uuid: string, json: string): AtmEvent[] {
|
||||
const root = JSON.parse(json) as Record<string, unknown>;
|
||||
const events: AtmEvent[] = [];
|
||||
for (const [id, entry] of Object.entries(root)) {
|
||||
if (id === 'DataVersion' || id.includes(':recipes/')) continue;
|
||||
const adv = entry as { done?: boolean; criteria?: Record<string, string> };
|
||||
if (!adv?.done) continue;
|
||||
const times = Object.values(adv.criteria ?? {})
|
||||
.map(parseAdvancementTime)
|
||||
.filter(Number.isFinite);
|
||||
if (times.length === 0) continue;
|
||||
events.push({
|
||||
id: deterministicId(`adv:${uuid}:${id}`),
|
||||
type: 'advancement',
|
||||
ts: Math.min(...times),
|
||||
uuid,
|
||||
name: null,
|
||||
advancementId: id,
|
||||
title: null,
|
||||
});
|
||||
}
|
||||
return events;
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
import { atmEventSchema, type AtmEvent, type EventOf, type EventType, type IngestResponse } from '@atm/shared';
|
||||
import type { Pool, PoolClient } from 'pg';
|
||||
import { withTx } from '../db/pool.ts';
|
||||
|
||||
export type Source = 'push' | 'sftp';
|
||||
export type Projection<T extends EventType = EventType> = (c: PoolClient, e: EventOf<T>, source: Source) => Promise<void>;
|
||||
|
||||
const projections = new Map<EventType, Projection[]>();
|
||||
|
||||
export function registerProjection<T extends EventType>(type: T, fn: Projection<T>): void {
|
||||
const list = projections.get(type) ?? [];
|
||||
list.push(fn as unknown as Projection);
|
||||
projections.set(type, list);
|
||||
}
|
||||
|
||||
const batchEnd = new WeakMap<PoolClient, Map<string, () => Promise<void>>>();
|
||||
|
||||
/**
|
||||
* Runs `fn` once at the end of the current batch (still inside its transaction), however many events asked
|
||||
* for it under the same key. Used for derived tables rebuilt from many files (quest definitions).
|
||||
*/
|
||||
export function deferToBatchEnd(c: PoolClient, key: string, fn: () => Promise<void>): void {
|
||||
let tasks = batchEnd.get(c);
|
||||
if (!tasks) batchEnd.set(c, (tasks = new Map()));
|
||||
tasks.set(key, fn);
|
||||
}
|
||||
|
||||
export type AfterCommit =(events: AtmEvent[], source: Source) => void | Promise<void>;
|
||||
const afterCommitHooks: AfterCommit[] = [];
|
||||
|
||||
/** Called with all newly inserted events after the transaction committed (live updates, cache invalidation). */
|
||||
export function onEventsCommitted(fn: AfterCommit): () => void {
|
||||
afterCommitHooks.push(fn);
|
||||
return () => {
|
||||
const i = afterCommitHooks.indexOf(fn);
|
||||
if (i >= 0) afterCommitHooks.splice(i, 1);
|
||||
};
|
||||
}
|
||||
|
||||
/** Validates each raw event on its own; one bad event never discards the batch. */
|
||||
export function validateEvents(raw: unknown[]): { valid: AtmEvent[]; rejected: IngestResponse['rejected'] } {
|
||||
const valid: AtmEvent[] = [];
|
||||
const rejected: IngestResponse['rejected'] = [];
|
||||
for (const item of raw) {
|
||||
const r = atmEventSchema.safeParse(item);
|
||||
if (r.success) {
|
||||
valid.push(r.data);
|
||||
} else {
|
||||
const id = typeof (item as { id?: unknown })?.id === 'string' ? (item as { id: string }).id : null;
|
||||
const reason = r.error.issues.map((i) => `${i.path.join('.') || '(root)'}: ${i.message}`).join('; ');
|
||||
rejected.push({ id, reason: reason.slice(0, 500) });
|
||||
}
|
||||
}
|
||||
return { valid, rejected };
|
||||
}
|
||||
|
||||
/**
|
||||
* Stores events idempotently (event_id is the primary key) and runs projections only for newly inserted ones,
|
||||
* all inside one transaction. Events are projected in timestamp order.
|
||||
*/
|
||||
export async function processEvents(
|
||||
pool: Pool,
|
||||
events: AtmEvent[],
|
||||
source: Source,
|
||||
serverId: string,
|
||||
): Promise<IngestResponse> {
|
||||
const sorted = [...events].sort((a, b) => a.ts - b.ts);
|
||||
const inserted = await withTx(pool, async (c) => {
|
||||
batchEnd.delete(c); // pooled clients are reused; drop leftovers of a failed batch
|
||||
const fresh: AtmEvent[] = [];
|
||||
for (const e of sorted) {
|
||||
const { rowCount } = await c.query(
|
||||
`INSERT INTO events(event_id, server_id, type, ts, source, payload)
|
||||
VALUES ($1, $2, $3, to_timestamp($4 / 1000.0), $5, $6)
|
||||
ON CONFLICT (event_id) DO NOTHING`,
|
||||
[e.id, serverId, e.type, e.ts, source, e],
|
||||
);
|
||||
if (!rowCount) continue;
|
||||
fresh.push(e);
|
||||
for (const p of projections.get(e.type) ?? []) await p(c, e, source);
|
||||
}
|
||||
const tasks = batchEnd.get(c);
|
||||
batchEnd.delete(c);
|
||||
for (const task of tasks?.values() ?? []) await task();
|
||||
return fresh;
|
||||
});
|
||||
for (const hook of afterCommitHooks) {
|
||||
try {
|
||||
await hook(inserted, source);
|
||||
} catch {
|
||||
/* hooks must not fail ingest */
|
||||
}
|
||||
}
|
||||
const freshIds = new Set(inserted.map((e) => e.id));
|
||||
return {
|
||||
accepted: events.filter((e) => freshIds.has(e.id)).map((e) => e.id),
|
||||
duplicate: events.filter((e) => !freshIds.has(e.id)).map((e) => e.id),
|
||||
rejected: [],
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
import { registerProjection } from '../process.ts';
|
||||
import { ensurePlayer } from './players.ts';
|
||||
|
||||
registerProjection('advancement', async (c, e) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, false);
|
||||
await c.query(
|
||||
`INSERT INTO advancements(uuid, advancement_id, achieved_at, title)
|
||||
VALUES ($1, $2, to_timestamp($3 / 1000.0), $4)
|
||||
ON CONFLICT (uuid, advancement_id) DO UPDATE SET
|
||||
achieved_at = LEAST(advancements.achieved_at, EXCLUDED.achieved_at),
|
||||
title = COALESCE(EXCLUDED.title, advancements.title)`,
|
||||
[e.uuid, e.advancementId, e.ts, e.title],
|
||||
);
|
||||
});
|
||||
|
||||
registerProjection('dimension_change', async (c, e) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, true);
|
||||
for (const dim of [e.from, e.to]) {
|
||||
await c.query(
|
||||
`INSERT INTO dimension_visits(uuid, dimension, first_at) VALUES ($1, $2, to_timestamp($3 / 1000.0))
|
||||
ON CONFLICT (uuid, dimension) DO UPDATE SET first_at = LEAST(dimension_visits.first_at, EXCLUDED.first_at)`,
|
||||
[e.uuid, dim, e.ts],
|
||||
);
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,20 @@
|
||||
import { registerProjection } from '../process.ts';
|
||||
import { ensurePlayer } from './players.ts';
|
||||
|
||||
registerProjection('player_death', async (c, e) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, true);
|
||||
await c.query(
|
||||
`INSERT INTO deaths(event_id, uuid, ts, dimension, cause, killer_type, killer_name)
|
||||
VALUES ($1, $2, to_timestamp($3 / 1000.0), $4, $5, $6, $7) ON CONFLICT DO NOTHING`,
|
||||
[e.id, e.uuid, e.ts, e.dimension, e.cause, e.killerType, e.killerName],
|
||||
);
|
||||
});
|
||||
|
||||
registerProjection('entity_kill', async (c, e) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, true);
|
||||
await c.query(
|
||||
`INSERT INTO kills(event_id, uuid, ts, dimension, victim_type, is_boss, weapon)
|
||||
VALUES ($1, $2, to_timestamp($3 / 1000.0), $4, $5, $6, $7) ON CONFLICT DO NOTHING`,
|
||||
[e.id, e.uuid, e.ts, e.dimension, e.victimType, e.isBoss, e.weapon],
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,88 @@
|
||||
import type { PoolClient } from 'pg';
|
||||
import { isQuestConfigFile, parseQuestDefs, parseQuestProgress, parseTeam } from '../../parsers/ftbquests.ts';
|
||||
import { deferToBatchEnd, registerProjection } from '../process.ts';
|
||||
|
||||
const QUEST_PROGRESS = /^world\/ftbquests\/[0-9a-f-]{32,36}\.snbt$/i;
|
||||
const TEAM_FILE = /^world\/ftbteams\/(party|player)\/[0-9a-f-]{32,36}\.snbt$/i;
|
||||
|
||||
export async function recordError(c: PoolClient, source: string, message: string): Promise<void> {
|
||||
await c.query('INSERT INTO app_errors(source, message) VALUES ($1, $2)', [source, message.slice(0, 2000)]);
|
||||
}
|
||||
|
||||
/** Rebuilds quest_defs from all cached chapter and English lang files (once per batch). */
|
||||
async function rebuildQuestDefs(c: PoolClient): Promise<void> {
|
||||
const files = await c.query<{ path: string; content: string }>(
|
||||
`SELECT path, content FROM file_cache WHERE path LIKE 'config/ftbquests/quests/%'`,
|
||||
);
|
||||
let defs;
|
||||
try {
|
||||
defs = parseQuestDefs(files.rows.filter((f) => isQuestConfigFile(f.path)));
|
||||
} catch (err) {
|
||||
await recordError(c, 'file_sync', `quest definitions: ${(err as Error).message}`);
|
||||
return;
|
||||
}
|
||||
await c.query('DELETE FROM quest_defs');
|
||||
if (defs.length) {
|
||||
await c.query(
|
||||
`INSERT INTO quest_defs(quest_id, chapter_id, chapter_title, title)
|
||||
SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[])
|
||||
ON CONFLICT (quest_id) DO NOTHING`,
|
||||
[defs.map((d) => d.questId), defs.map((d) => d.chapterId), defs.map((d) => d.chapterTitle), defs.map((d) => d.title)],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
registerProjection('file_sync', async (c, e) => {
|
||||
const { rows } = await c.query<{ sha256: string }>('SELECT sha256 FROM file_cache WHERE path = $1', [e.path]);
|
||||
if (rows[0]?.sha256 === e.sha256) return; // unchanged content
|
||||
await c.query(
|
||||
`INSERT INTO file_cache(path, sha256, content, updated_at) VALUES ($1, $2, $3, to_timestamp($4 / 1000.0))
|
||||
ON CONFLICT (path) DO UPDATE SET sha256 = EXCLUDED.sha256, content = EXCLUDED.content, updated_at = EXCLUDED.updated_at`,
|
||||
[e.path, e.sha256, e.content, e.ts],
|
||||
);
|
||||
|
||||
try {
|
||||
if (QUEST_PROGRESS.test(e.path)) {
|
||||
const p = parseQuestProgress(e.content);
|
||||
// The file is the team's complete state: quests no longer listed were reset in FTB Quests.
|
||||
await c.query(`DELETE FROM quest_progress WHERE team_id = $1 AND NOT (quest_id = ANY($2::text[]))`, [
|
||||
p.teamId,
|
||||
p.completed.map((q) => q.questId),
|
||||
]);
|
||||
if (p.completed.length) {
|
||||
await c.query(
|
||||
`INSERT INTO quest_progress(team_id, quest_id, completed_at)
|
||||
SELECT $1, t.id, to_timestamp(t.at / 1000.0) FROM unnest($2::text[], $3::bigint[]) AS t(id, at)
|
||||
ON CONFLICT (team_id, quest_id) DO UPDATE SET completed_at = EXCLUDED.completed_at`,
|
||||
[p.teamId, p.completed.map((q) => q.questId), p.completed.map((q) => q.at)],
|
||||
);
|
||||
}
|
||||
// Team names from ftbteams win; the progress file only provides a fallback ("Name#uuidprefix").
|
||||
await c.query(`INSERT INTO teams(team_id, name, type) VALUES ($1, $2, NULL) ON CONFLICT (team_id) DO NOTHING`, [
|
||||
p.teamId,
|
||||
p.name,
|
||||
]);
|
||||
} else if (isQuestConfigFile(e.path)) {
|
||||
deferToBatchEnd(c, 'quest_defs', () => rebuildQuestDefs(c));
|
||||
} else if (TEAM_FILE.test(e.path)) {
|
||||
const t = parseTeam(e.content);
|
||||
await c.query(
|
||||
`INSERT INTO teams(team_id, name, type) VALUES ($1, $2, $3)
|
||||
ON CONFLICT (team_id) DO UPDATE SET name = COALESCE(EXCLUDED.name, teams.name), type = EXCLUDED.type`,
|
||||
[t.teamId, t.name, t.type],
|
||||
);
|
||||
// Only party membership is stored: a player outside a party plays as their own team (team id = uuid).
|
||||
if (t.type === 'party') {
|
||||
await c.query('DELETE FROM team_members WHERE team_id = $1', [t.teamId]);
|
||||
if (t.members.length) {
|
||||
// a player belongs to one party at a time
|
||||
await c.query('DELETE FROM team_members WHERE uuid = ANY($1::uuid[])', [t.members]);
|
||||
await c.query(`INSERT INTO team_members(team_id, uuid) SELECT $1, unnest($2::uuid[])`, [t.teamId, t.members]);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
// A malformed file must not reject the batch; it stays cached and is reported to the admin.
|
||||
await recordError(c, 'file_sync', `${e.path}: ${(err as Error).message}`);
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,7 @@
|
||||
// Side-effect imports register all projections with the pipeline.
|
||||
import './sessions.ts';
|
||||
import './combat.ts';
|
||||
import './advancements.ts';
|
||||
import './stats.ts';
|
||||
import './server.ts';
|
||||
import './files.ts';
|
||||
@@ -0,0 +1,29 @@
|
||||
import type { PoolClient } from 'pg';
|
||||
|
||||
/**
|
||||
* Upserts a player. A non-null name replaces the stored one only if it was observed at least as late as the
|
||||
* current name (late replays never revert a rename) and is recorded in name_history;
|
||||
* `seen` updates last_seen (only for events proving the player was online).
|
||||
*/
|
||||
export async function ensurePlayer(c: PoolClient, uuid: string, name: string | null, ts: number, seen: boolean): Promise<void> {
|
||||
await c.query(
|
||||
`INSERT INTO players(uuid, name, name_seen_at, first_seen, last_seen)
|
||||
VALUES ($1, $2, CASE WHEN $2::text IS NOT NULL THEN to_timestamp($3 / 1000.0) END,
|
||||
to_timestamp($3 / 1000.0), CASE WHEN $4 THEN to_timestamp($3 / 1000.0) END)
|
||||
ON CONFLICT (uuid) DO UPDATE SET
|
||||
name = CASE WHEN EXCLUDED.name IS NOT NULL AND (players.name_seen_at IS NULL OR EXCLUDED.name_seen_at >= players.name_seen_at)
|
||||
THEN EXCLUDED.name ELSE players.name END,
|
||||
name_seen_at = CASE WHEN EXCLUDED.name IS NOT NULL AND (players.name_seen_at IS NULL OR EXCLUDED.name_seen_at >= players.name_seen_at)
|
||||
THEN EXCLUDED.name_seen_at ELSE players.name_seen_at END,
|
||||
first_seen = LEAST(players.first_seen, EXCLUDED.first_seen),
|
||||
last_seen = CASE WHEN $4 THEN GREATEST(players.last_seen, EXCLUDED.last_seen) ELSE players.last_seen END`,
|
||||
[uuid, name, ts, seen],
|
||||
);
|
||||
if (name) {
|
||||
await c.query(
|
||||
`INSERT INTO name_history(uuid, name, seen_at) VALUES ($1, $2, to_timestamp($3 / 1000.0))
|
||||
ON CONFLICT (uuid, name) DO UPDATE SET seen_at = GREATEST(name_history.seen_at, EXCLUDED.seen_at)`,
|
||||
[uuid, name, ts],
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import { registerProjection } from '../process.ts';
|
||||
|
||||
registerProjection('health', async (c, e) => {
|
||||
await c.query(
|
||||
`INSERT INTO server_health(bucket, mspt_avg, mspt_p95, mspt_max, tps, players, chunks, heap_used_mb, heap_max_mb)
|
||||
VALUES (date_trunc('minute', to_timestamp($1 / 1000.0)), $2, $3, $4, $5, $6, $7, $8, $9)
|
||||
ON CONFLICT (bucket) DO UPDATE SET
|
||||
mspt_avg = EXCLUDED.mspt_avg, mspt_p95 = EXCLUDED.mspt_p95, mspt_max = EXCLUDED.mspt_max, tps = EXCLUDED.tps,
|
||||
players = EXCLUDED.players, chunks = EXCLUDED.chunks,
|
||||
heap_used_mb = EXCLUDED.heap_used_mb, heap_max_mb = EXCLUDED.heap_max_mb`,
|
||||
[e.windowStart, e.msptAvg, e.msptP95, e.msptMax, e.tps, e.players, e.chunks, e.heapUsedMb, e.heapMaxMb],
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,65 @@
|
||||
import { registerProjection } from '../process.ts';
|
||||
import { ensurePlayer } from './players.ts';
|
||||
|
||||
/**
|
||||
* Push and log (sftp) both report joins. A join that falls inside a session already recorded by the other
|
||||
* source is the same visit seen twice and is skipped; only an open session of the same source is superseded.
|
||||
*/
|
||||
registerProjection('player_join', async (c, e, source) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, true);
|
||||
const { rowCount: covered } = await c.query(
|
||||
`SELECT 1 FROM sessions
|
||||
WHERE uuid = $1 AND source <> $3 AND started_at <= to_timestamp($2 / 1000.0)
|
||||
AND (ended_at IS NULL OR ended_at >= to_timestamp($2 / 1000.0))`,
|
||||
[e.uuid, e.ts, source],
|
||||
);
|
||||
if (covered) return;
|
||||
await c.query(
|
||||
`UPDATE sessions SET ended_at = GREATEST(started_at, to_timestamp($2 / 1000.0)), end_reason = 'superseded'
|
||||
WHERE uuid = $1 AND ended_at IS NULL AND source = $3`,
|
||||
[e.uuid, e.ts, source],
|
||||
);
|
||||
await c.query(`INSERT INTO sessions(uuid, started_at, source) VALUES ($1, to_timestamp($2 / 1000.0), $3)`, [e.uuid, e.ts, source]);
|
||||
});
|
||||
|
||||
registerProjection('player_leave', async (c, e) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, true);
|
||||
await c.query(
|
||||
`UPDATE sessions SET ended_at = GREATEST(started_at, to_timestamp($2 / 1000.0)), end_reason = 'leave'
|
||||
WHERE uuid = $1 AND ended_at IS NULL`,
|
||||
[e.uuid, e.ts],
|
||||
);
|
||||
});
|
||||
|
||||
registerProjection('server_stop', async (c, e) => {
|
||||
await c.query(
|
||||
`UPDATE sessions SET ended_at = GREATEST(started_at, to_timestamp($1 / 1000.0)), end_reason = 'stop'
|
||||
WHERE ended_at IS NULL`,
|
||||
[e.ts],
|
||||
);
|
||||
});
|
||||
|
||||
/**
|
||||
* A server_start while sessions are still open means the server went down without server_stop (crash, kill).
|
||||
* Each open session ends at the last sign of life before the start: the player's last event or the last
|
||||
* health window, whichever is later — never later than the start itself.
|
||||
*/
|
||||
registerProjection('server_start', async (c, e) => {
|
||||
await c.query(
|
||||
`UPDATE sessions s SET
|
||||
end_reason = 'crash',
|
||||
ended_at = GREATEST(s.started_at, LEAST(to_timestamp($1 / 1000.0), COALESCE(GREATEST(
|
||||
(SELECT max(ev.ts) FROM events ev
|
||||
WHERE ev.payload->>'uuid' = s.uuid::text AND ev.ts >= s.started_at AND ev.ts < to_timestamp($1 / 1000.0)),
|
||||
(SELECT to_timestamp(max((ev.payload->>'windowEnd')::bigint) / 1000.0) FROM events ev
|
||||
WHERE ev.type = 'health' AND ev.ts >= s.started_at AND ev.ts < to_timestamp($1 / 1000.0))
|
||||
), s.started_at)))
|
||||
WHERE s.ended_at IS NULL`,
|
||||
[e.ts],
|
||||
);
|
||||
await c.query(
|
||||
`INSERT INTO server_info(started_at, mc_version, neoforge_version, java_version, mods)
|
||||
VALUES (to_timestamp($1 / 1000.0), $2, $3, $4, $5)`,
|
||||
[e.ts, e.mcVersion, e.neoforgeVersion, e.javaVersion, JSON.stringify(e.mods)],
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,31 @@
|
||||
import { registerProjection } from '../process.ts';
|
||||
import { ensurePlayer } from './players.ts';
|
||||
|
||||
/**
|
||||
* Counters are absolute. Per (uuid, category, key) the observation with the newest timestamp wins,
|
||||
* so late or repeated deliveries (push vs. SFTP) never move a value backwards in time.
|
||||
* Keys missing from a snapshot are left untouched, even for `full` snapshots.
|
||||
*/
|
||||
registerProjection('stats_snapshot', async (c, e, source) => {
|
||||
await ensurePlayer(c, e.uuid, e.name, e.ts, false);
|
||||
const cats: string[] = [];
|
||||
const keys: string[] = [];
|
||||
const vals: number[] = [];
|
||||
for (const [cat, entries] of Object.entries(e.stats)) {
|
||||
for (const [key, value] of Object.entries(entries)) {
|
||||
cats.push(cat);
|
||||
keys.push(key);
|
||||
vals.push(value);
|
||||
}
|
||||
}
|
||||
if (cats.length === 0) return;
|
||||
await c.query(
|
||||
`INSERT INTO stat_current(uuid, category, key, value, observed_at, source)
|
||||
SELECT $1, t.cat, t.key, t.val, to_timestamp($5 / 1000.0), $6
|
||||
FROM unnest($2::text[], $3::text[], $4::bigint[]) AS t(cat, key, val)
|
||||
ON CONFLICT (uuid, category, key) DO UPDATE SET
|
||||
value = EXCLUDED.value, observed_at = EXCLUDED.observed_at, source = EXCLUDED.source
|
||||
WHERE stat_current.observed_at < EXCLUDED.observed_at`,
|
||||
[e.uuid, cats, keys, vals, e.ts, source],
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,124 @@
|
||||
import { timingSafeEqual } from 'node:crypto';
|
||||
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import { z } from 'zod';
|
||||
import { getSession, login, logout, SESSION_TTL_MS } from '../auth/admin.ts';
|
||||
import type { Config } from '../config.ts';
|
||||
import { createIngestToken, revokeAllIngestTokens } from '../ingest/tokens.ts';
|
||||
import { UUID_RE } from './public/shared.ts';
|
||||
|
||||
const COOKIE = 'atm_admin';
|
||||
|
||||
declare module 'fastify' {
|
||||
interface FastifyRequest {
|
||||
admin?: { username: string; csrf: string; sessionId: string };
|
||||
}
|
||||
}
|
||||
|
||||
const sameToken = (a: string, b: string) => a.length === b.length && timingSafeEqual(Buffer.from(a), Buffer.from(b));
|
||||
|
||||
export async function adminRoutes(app: FastifyInstance, { pool, config }: { pool: Pool; config: Config }) {
|
||||
const secure = config.NODE_ENV !== 'test' && config.NODE_ENV !== 'development';
|
||||
|
||||
app.addHook('onSend', async (_req, reply) => {
|
||||
reply.header('cache-control', 'no-store');
|
||||
});
|
||||
|
||||
const requireAdmin = async (req: FastifyRequest, reply: FastifyReply) => {
|
||||
const sessionId = req.cookies[COOKIE];
|
||||
const session = sessionId ? await getSession(pool, sessionId) : null;
|
||||
if (!session || !sessionId) return reply.code(401).send({ error: 'unauthorized' });
|
||||
if (req.method !== 'GET') {
|
||||
const header = req.headers['x-csrf-token'];
|
||||
if (typeof header !== 'string' || !sameToken(header, session.csrf)) return reply.code(403).send({ error: 'csrf' });
|
||||
}
|
||||
req.admin = { ...session, sessionId };
|
||||
};
|
||||
|
||||
app.post(
|
||||
'/api/admin/login',
|
||||
{ config: { rateLimit: { max: 5, timeWindow: '1 minute' } } },
|
||||
async (req, reply) => {
|
||||
const body = z.object({ username: z.string().min(1).max(64), password: z.string().min(1).max(256) }).safeParse(req.body);
|
||||
if (!body.success) return reply.code(400).send({ error: 'bad_request' });
|
||||
const s = await login(pool, body.data.username, body.data.password);
|
||||
if (!s) {
|
||||
req.log.warn({ username: body.data.username }, 'admin login failed');
|
||||
return reply.code(401).send({ error: 'invalid_credentials' });
|
||||
}
|
||||
reply.setCookie(COOKIE, s.sessionId, {
|
||||
httpOnly: true,
|
||||
secure,
|
||||
sameSite: 'strict',
|
||||
path: '/api/admin',
|
||||
maxAge: SESSION_TTL_MS / 1000,
|
||||
});
|
||||
return { csrf: s.csrf, username: body.data.username };
|
||||
},
|
||||
);
|
||||
|
||||
app.register(async (protectedApp) => {
|
||||
protectedApp.addHook('preHandler', requireAdmin);
|
||||
|
||||
protectedApp.get('/api/admin/session', async (req) => ({ username: req.admin!.username, csrf: req.admin!.csrf }));
|
||||
|
||||
protectedApp.post('/api/admin/logout', async (req, reply) => {
|
||||
await logout(pool, req.admin!.sessionId);
|
||||
reply.clearCookie(COOKIE, { path: '/api/admin' });
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
protectedApp.get('/api/admin/status', async () => {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT (SELECT max(received_at) FROM events WHERE source = 'push') AS last_push,
|
||||
(SELECT max(received_at) FROM events WHERE source = 'sftp') AS last_sftp_event,
|
||||
(SELECT max(synced_at) FROM sync_state) AS last_sync,
|
||||
(SELECT GREATEST(reltuples, 0)::bigint FROM pg_class WHERE relname = 'events') AS events, -- estimate, no table scan
|
||||
(SELECT count(*) FROM ingest_tokens WHERE revoked_at IS NULL)::int AS active_tokens,
|
||||
(SELECT count(*) FROM sync_requests WHERE done_at IS NULL)::int AS pending_sync`,
|
||||
);
|
||||
const errors = await pool.query(`SELECT at, source, message FROM app_errors ORDER BY at DESC LIMIT 20`);
|
||||
const r = rows[0];
|
||||
return {
|
||||
push: { lastEventAt: r.last_push?.getTime() ?? null },
|
||||
sftp: {
|
||||
enabled: config.sftp !== null,
|
||||
lastSyncAt: r.last_sync?.getTime() ?? null,
|
||||
lastEventAt: r.last_sftp_event?.getTime() ?? null,
|
||||
pendingRequests: r.pending_sync,
|
||||
},
|
||||
events: r.events,
|
||||
activeTokens: r.active_tokens,
|
||||
lastErrors: errors.rows.map((e) => ({ at: e.at.getTime(), source: e.source, message: e.message })),
|
||||
};
|
||||
});
|
||||
|
||||
protectedApp.get('/api/admin/players', async () => {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT uuid, name, hidden, last_seen FROM players ORDER BY last_seen DESC NULLS LAST, name`,
|
||||
);
|
||||
return { players: rows.map((r) => ({ uuid: r.uuid, name: r.name, hidden: r.hidden, lastSeen: r.last_seen?.getTime() ?? null })) };
|
||||
});
|
||||
|
||||
protectedApp.post<{ Params: { uuid: string } }>('/api/admin/players/:uuid/hidden', async (req, reply) => {
|
||||
const body = z.object({ hidden: z.boolean() }).safeParse(req.body);
|
||||
if (!UUID_RE.test(req.params.uuid) || !body.success) return reply.code(400).send({ error: 'bad_request' });
|
||||
const { rowCount } = await pool.query('UPDATE players SET hidden = $2 WHERE uuid = $1', [req.params.uuid, body.data.hidden]);
|
||||
if (!rowCount) return reply.code(404).send({ error: 'not_found' });
|
||||
req.log.info({ admin: req.admin!.username, uuid: req.params.uuid, hidden: body.data.hidden }, 'player visibility changed');
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
protectedApp.post('/api/admin/sync', async (_req, reply) => {
|
||||
await pool.query('INSERT INTO sync_requests DEFAULT VALUES');
|
||||
return reply.code(202).send({ queued: true });
|
||||
});
|
||||
|
||||
protectedApp.post('/api/admin/ingest-token/rotate', async (req) => {
|
||||
await revokeAllIngestTokens(pool);
|
||||
const token = await createIngestToken(pool);
|
||||
req.log.info({ admin: req.admin!.username }, 'ingest token rotated');
|
||||
return { token };
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
import type { LiveEvent } from '@atm/shared/api';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { LiveHub } from '../live/hub.ts';
|
||||
|
||||
export const HEARTBEAT_MS = 25_000;
|
||||
const MAX_CLIENTS = 500;
|
||||
|
||||
const frame = (e: LiveEvent) => `id: ${e.id}\nevent: ${e.kind}\ndata: ${JSON.stringify(e.data)}\n\n`;
|
||||
|
||||
export async function liveRoutes(app: FastifyInstance, { hub }: { hub: LiveHub }) {
|
||||
let clients = 0;
|
||||
const open = new Set<{ end: () => void }>();
|
||||
// Hijacked streams are not tracked by Fastify; end them so shutdown does not hang.
|
||||
app.addHook('onClose', async () => {
|
||||
for (const res of open) res.end();
|
||||
});
|
||||
|
||||
app.get<{ Headers: { 'last-event-id'?: string } }>('/api/live', async (req, reply) => {
|
||||
if (clients >= MAX_CLIENTS) return reply.code(503).send({ error: 'too_many_clients' });
|
||||
clients++;
|
||||
reply.hijack();
|
||||
const res = reply.raw;
|
||||
res.writeHead(200, {
|
||||
'content-type': 'text/event-stream; charset=utf-8',
|
||||
'cache-control': 'no-cache, no-transform',
|
||||
connection: 'keep-alive',
|
||||
'x-accel-buffering': 'no',
|
||||
});
|
||||
res.write(`retry: 5000\n: connected\n\n`);
|
||||
|
||||
const lastId = Number(req.headers['last-event-id']);
|
||||
if (Number.isFinite(lastId) && lastId > 0) for (const e of hub.since(lastId)) res.write(frame(e));
|
||||
|
||||
const unsubscribe = hub.subscribe((e) => res.write(frame(e)));
|
||||
const heartbeat = setInterval(() => res.write(': ping\n\n'), HEARTBEAT_MS);
|
||||
open.add(res);
|
||||
const close = () => {
|
||||
clearInterval(heartbeat);
|
||||
unsubscribe();
|
||||
open.delete(res);
|
||||
clients--;
|
||||
};
|
||||
req.raw.on('close', close);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import { UUID_RE } from './shared.ts';
|
||||
|
||||
const TTL_MS = 24 * 3600_000;
|
||||
const MAX_ENTRIES = 500;
|
||||
const TIMEOUT_MS = 5000;
|
||||
|
||||
// 8×8 neutral grey PNG, served when the avatar service is unreachable.
|
||||
const PLACEHOLDER = Buffer.from(
|
||||
'iVBORw0KGgoAAAANSUhEUgAAAAgAAAAICAIAAABLbSncAAAAEUlEQVR42mN4+vQpTsQwciUBADoNkKUl+DzzAAAAAElFTkSuQmCC',
|
||||
'base64',
|
||||
);
|
||||
|
||||
/**
|
||||
* Player heads are fetched server-side and cached, so visitors' browsers never contact third parties
|
||||
* (no visitor IPs leave the portal).
|
||||
*/
|
||||
export async function avatarRoutes(app: FastifyInstance, { source }: { source: string }) {
|
||||
const cache = new Map<string, { png: Buffer; at: number }>();
|
||||
|
||||
app.get<{ Params: { uuid: string } }>('/api/avatar/:uuid', async (req, reply) => {
|
||||
const uuid = req.params.uuid.toLowerCase();
|
||||
if (!UUID_RE.test(uuid)) return reply.code(400).send({ error: 'bad_uuid' });
|
||||
reply.header('content-type', 'image/png').header('cache-control', 'public, max-age=86400');
|
||||
|
||||
const hit = cache.get(uuid);
|
||||
if (hit && Date.now() - hit.at < TTL_MS) return reply.send(hit.png);
|
||||
try {
|
||||
const res = await fetch(source.replace('{uuid}', uuid), { signal: AbortSignal.timeout(TIMEOUT_MS) });
|
||||
const type = res.headers.get('content-type') ?? '';
|
||||
if (!res.ok || !type.startsWith('image/png')) throw new Error(`avatar ${res.status}`);
|
||||
const png = Buffer.from(await res.arrayBuffer());
|
||||
if (png.length > 256 * 1024) throw new Error('avatar too large');
|
||||
if (cache.size >= MAX_ENTRIES) cache.delete(cache.keys().next().value!);
|
||||
cache.set(uuid, { png, at: Date.now() });
|
||||
return reply.send(png);
|
||||
} catch (err) {
|
||||
req.log.debug({ err }, 'avatar fetch failed');
|
||||
return reply.header('cache-control', 'public, max-age=300').send(hit?.png ?? PLACEHOLDER);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import type { Config } from '../../config.ts';
|
||||
import { avatarRoutes } from './avatar.ts';
|
||||
import { leaderboardRoutes } from './leaderboards.ts';
|
||||
import { overviewRoutes } from './overview.ts';
|
||||
import { playerRoutes } from './players.ts';
|
||||
import { serverRoutes } from './server.ts';
|
||||
|
||||
export async function publicRoutes(app: FastifyInstance, { pool, config }: { pool: Pool; config: Config }) {
|
||||
// Short shared cache for all public read endpoints (Traefik/browsers); avatar sets its own.
|
||||
app.addHook('onSend', async (req, reply) => {
|
||||
if (req.method === 'GET' && !reply.hasHeader('cache-control')) reply.header('cache-control', 'public, max-age=15');
|
||||
});
|
||||
await overviewRoutes(app, { pool });
|
||||
await leaderboardRoutes(app, { pool });
|
||||
await playerRoutes(app, { pool, timeZone: config.TIMEZONE });
|
||||
await serverRoutes(app, { pool, config });
|
||||
await avatarRoutes(app, { source: config.AVATAR_SOURCE });
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import type { LeaderboardDto, LeaderboardsDto, Period } from '@atm/shared/api';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import { getLeaderboard } from '../../stats/leaderboards.ts';
|
||||
import { BOARD_BY_ID, BOARDS, type Board } from '../../stats/metrics.ts';
|
||||
|
||||
const PERIODS: Period[] = ['all', '30d', '7d'];
|
||||
const meta = (b: Board) => ({ id: b.id, title: b.title, description: b.description, unit: b.unit });
|
||||
|
||||
export async function leaderboardRoutes(app: FastifyInstance, { pool }: { pool: Pool }) {
|
||||
app.get('/api/leaderboards', async (): Promise<LeaderboardsDto> => ({
|
||||
boards: await Promise.all(BOARDS.map(async (b) => ({ ...meta(b), top: await getLeaderboard(pool, b.id, 'all', 10) }))),
|
||||
}));
|
||||
|
||||
app.get<{ Params: { id: string }; Querystring: { period?: string } }>('/api/leaderboards/:id', async (req, reply) => {
|
||||
const board = BOARD_BY_ID.get(req.params.id);
|
||||
if (!board) return reply.code(404).send({ error: 'unknown_board' });
|
||||
const period = (req.query.period ?? 'all') as Period;
|
||||
if (!PERIODS.includes(period)) return reply.code(400).send({ error: 'bad_period' });
|
||||
const dto: LeaderboardDto = { board: meta(board), period, entries: await getLeaderboard(pool, board.id, period, 100) };
|
||||
return dto;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
import type { OnlineDto, OverviewDto } from '@atm/shared/api';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import { hours, ms, QUEST_FILTER } from './shared.ts';
|
||||
|
||||
export async function overviewRoutes(app: FastifyInstance, { pool }: { pool: Pool }) {
|
||||
app.get('/api/overview', async (): Promise<OverviewDto> => {
|
||||
const { rows } = await pool.query(
|
||||
`WITH visible AS (SELECT uuid FROM players WHERE NOT hidden),
|
||||
stat AS (
|
||||
SELECT key, sum(value) AS total FROM stat_current
|
||||
WHERE category = 'minecraft:custom' AND key IN ('minecraft:play_time', 'minecraft:deaths', 'minecraft:mob_kills')
|
||||
AND uuid IN (SELECT uuid FROM visible)
|
||||
GROUP BY key)
|
||||
SELECT
|
||||
(SELECT count(*) FROM sessions WHERE ended_at IS NULL AND uuid IN (SELECT uuid FROM visible))::int AS online,
|
||||
(SELECT count(*) FROM visible)::int AS total_players,
|
||||
COALESCE((SELECT total FROM stat WHERE key = 'minecraft:play_time'), 0) AS play_ticks,
|
||||
COALESCE((SELECT total FROM stat WHERE key = 'minecraft:deaths'), 0) AS deaths,
|
||||
COALESCE((SELECT total FROM stat WHERE key = 'minecraft:mob_kills'), 0) AS mob_kills,
|
||||
(SELECT count(*) FROM quest_progress qp WHERE ${QUEST_FILTER})::int AS quests,
|
||||
(SELECT count(*) FROM advancements WHERE uuid IN (SELECT uuid FROM visible))::int AS advancements,
|
||||
(SELECT received_at FROM events ORDER BY received_at DESC LIMIT 1) AS last_update`,
|
||||
);
|
||||
const r = rows[0];
|
||||
return {
|
||||
online: r.online,
|
||||
totalPlayers: r.total_players,
|
||||
totalPlayHours: hours(r.play_ticks),
|
||||
totalDeaths: r.deaths,
|
||||
totalMobKills: r.mob_kills,
|
||||
questsCompleted: r.quests,
|
||||
advancements: r.advancements,
|
||||
lastUpdate: ms(r.last_update),
|
||||
};
|
||||
});
|
||||
|
||||
app.get('/api/online', async (): Promise<OnlineDto> => {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT s.uuid, p.name, s.started_at,
|
||||
(SELECT e.payload->>'to' FROM events e
|
||||
WHERE e.type = 'dimension_change' AND e.payload->>'uuid' = s.uuid::text AND e.ts >= s.started_at
|
||||
ORDER BY e.ts DESC LIMIT 1) AS dimension
|
||||
FROM sessions s JOIN players p ON p.uuid = s.uuid AND NOT p.hidden
|
||||
WHERE s.ended_at IS NULL
|
||||
ORDER BY s.started_at`,
|
||||
);
|
||||
return { players: rows.map((r) => ({ uuid: r.uuid, name: r.name, since: r.started_at.getTime(), dimension: r.dimension })) };
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,193 @@
|
||||
import { ITEM_CATEGORIES, type HistoryDto, type ItemCategory, type ItemsDto, type PlayerListDto, type PlayerProfileDto } from '@atm/shared/api';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import { getPlayerRanks } from '../../stats/leaderboards.ts';
|
||||
import { BOARD_BY_ID, metricSql, toDisplay } from '../../stats/metrics.ts';
|
||||
import { hours, ms, QUEST_FILTER, resolvePlayer } from './shared.ts';
|
||||
|
||||
const PAGE_SIZE = 50;
|
||||
const RANGES = { '7d': { days: 7, granularity: 'hour' }, '30d': { days: 30, granularity: 'day' }, all: { days: null, granularity: 'day' } } as const;
|
||||
|
||||
export async function playerRoutes(app: FastifyInstance, { pool, timeZone }: { pool: Pool; timeZone: string }) {
|
||||
app.get<{ Querystring: { search?: string; page?: string } }>('/api/players', async (req): Promise<PlayerListDto> => {
|
||||
const search = (req.query.search ?? '').trim().slice(0, 16);
|
||||
const page = Math.max(0, Number.parseInt(req.query.page ?? '0', 10) || 0);
|
||||
const like = `%${search.replace(/[\\%_]/g, (c) => `\\${c}`)}%`;
|
||||
const { rows } = await pool.query(
|
||||
`SELECT p.uuid, p.name, p.last_seen,
|
||||
COALESCE(sc.value, 0) AS play_ticks,
|
||||
EXISTS (SELECT 1 FROM sessions s WHERE s.uuid = p.uuid AND s.ended_at IS NULL) AS online,
|
||||
count(*) OVER ()::int AS total
|
||||
FROM players p
|
||||
LEFT JOIN stat_current sc ON sc.uuid = p.uuid AND sc.category = 'minecraft:custom' AND sc.key = 'minecraft:play_time'
|
||||
WHERE NOT p.hidden AND p.name IS NOT NULL AND ($1 = '' OR p.name ILIKE $2)
|
||||
ORDER BY online DESC, p.last_seen DESC NULLS LAST, p.name
|
||||
LIMIT ${PAGE_SIZE} OFFSET $3`,
|
||||
[search, like, page * PAGE_SIZE],
|
||||
);
|
||||
return {
|
||||
total: rows[0]?.total ?? 0,
|
||||
players: rows.map((r) => ({ uuid: r.uuid, name: r.name, lastSeen: ms(r.last_seen), playHours: hours(r.play_ticks), online: r.online })),
|
||||
};
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: string } }>('/api/players/:id', async (req, reply) => {
|
||||
const uuid = await resolvePlayer(pool, req.params.id);
|
||||
if (!uuid) return reply.code(404).send({ error: 'not_found' });
|
||||
return getProfile(pool, uuid, timeZone);
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: string }; Querystring: { metric?: string; range?: string } }>(
|
||||
'/api/players/:id/history',
|
||||
async (req, reply) => {
|
||||
const uuid = await resolvePlayer(pool, req.params.id);
|
||||
if (!uuid) return reply.code(404).send({ error: 'not_found' });
|
||||
const board = BOARD_BY_ID.get(req.query.metric ?? 'playtime');
|
||||
const range = RANGES[(req.query.range ?? '30d') as keyof typeof RANGES];
|
||||
if (!board || !range) return reply.code(400).send({ error: 'bad_query' });
|
||||
const { rows } = await pool.query<{ bucket: Date; value: number }>(
|
||||
`SELECT bucket, value FROM stat_snapshots
|
||||
WHERE uuid = $1 AND metric = $2 AND granularity = $3 AND ($4::int IS NULL OR bucket >= now() - make_interval(days => $4::int))
|
||||
ORDER BY bucket`,
|
||||
[uuid, board.id, range.granularity, range.days],
|
||||
);
|
||||
const current = await pool.query<{ value: number }>(`SELECT value FROM (${metricSql(board.metric)}) v WHERE v.uuid = $1`, [uuid]);
|
||||
const points = rows.map((r) => ({ t: r.bucket.getTime(), v: toDisplay(board.unit, r.value) }));
|
||||
if (current.rows[0]) points.push({ t: Date.now(), v: toDisplay(board.unit, current.rows[0].value) });
|
||||
const dto: HistoryDto = { metric: board.id, unit: board.unit, points };
|
||||
return dto;
|
||||
},
|
||||
);
|
||||
|
||||
app.get<{ Params: { id: string }; Querystring: { category?: string } }>('/api/players/:id/items', async (req, reply) => {
|
||||
const uuid = await resolvePlayer(pool, req.params.id);
|
||||
if (!uuid) return reply.code(404).send({ error: 'not_found' });
|
||||
const category = (req.query.category ?? 'mined') as ItemCategory;
|
||||
if (!ITEM_CATEGORIES.includes(category)) return reply.code(400).send({ error: 'bad_category' });
|
||||
const { rows } = await pool.query<{ ns: string; key: string; value: number; ns_total: number }>(
|
||||
`SELECT ns, key, value, ns_total FROM (
|
||||
SELECT split_part(key, ':', 1) AS ns, key, value,
|
||||
sum(value) OVER (PARTITION BY split_part(key, ':', 1)) AS ns_total,
|
||||
row_number() OVER (PARTITION BY split_part(key, ':', 1) ORDER BY value DESC, key) AS rn
|
||||
FROM stat_current WHERE uuid = $1 AND category = $2 AND value > 0) t
|
||||
WHERE rn <= 20 ORDER BY ns_total DESC, ns, value DESC`,
|
||||
[uuid, `minecraft:${category}`],
|
||||
);
|
||||
const mods: ItemsDto['mods'] = [];
|
||||
for (const r of rows) {
|
||||
let mod = mods.at(-1);
|
||||
if (mod?.namespace !== r.ns) {
|
||||
mod = { namespace: r.ns, total: r.ns_total, items: [] };
|
||||
mods.push(mod);
|
||||
}
|
||||
mod.items.push({ key: r.key, value: r.value });
|
||||
}
|
||||
const dto: ItemsDto = { category, mods };
|
||||
return dto;
|
||||
});
|
||||
}
|
||||
|
||||
async function getProfile(pool: Pool, uuid: string, timeZone: string): Promise<PlayerProfileDto> {
|
||||
const q = async <T extends Record<string, unknown>>(sql: string, params: unknown[] = [uuid]) => (await pool.query<T>(sql, params)).rows;
|
||||
|
||||
const [player] = await q<{ name: string | null; first_seen: Date; last_seen: Date | null }>(
|
||||
'SELECT name, first_seen, last_seen FROM players WHERE uuid = $1',
|
||||
);
|
||||
const [sess] = await q<{ n: number; online: boolean; minutes: number }>(
|
||||
`SELECT count(*)::int AS n, bool_or(ended_at IS NULL) AS online,
|
||||
COALESCE(sum(extract(epoch FROM COALESCE(ended_at, now()) - started_at) / 60), 0) AS minutes
|
||||
FROM sessions WHERE uuid = $1`,
|
||||
);
|
||||
const custom = await q<{ key: string; value: number }>(
|
||||
`SELECT key, value FROM stat_current WHERE uuid = $1 AND category = 'minecraft:custom' ORDER BY key`,
|
||||
);
|
||||
const general: Record<string, number> = {};
|
||||
const movement: Record<string, number> = {};
|
||||
for (const r of custom) (r.key.endsWith('_one_cm') ? movement : general)[r.key] = r.value;
|
||||
const playTicks = general['minecraft:play_time'];
|
||||
|
||||
const counted = (category: string) =>
|
||||
q<{ key: string; value: number }>(
|
||||
`SELECT key, value FROM stat_current WHERE uuid = $1 AND category = $2 AND value > 0 ORDER BY value DESC, key LIMIT 25`,
|
||||
[uuid, category],
|
||||
);
|
||||
|
||||
const team = await q<{ team_id: string; name: string | null }>(
|
||||
`SELECT COALESCE(tm.team_id, $1::uuid) AS team_id, t.name
|
||||
FROM (SELECT $1::uuid AS uuid) p
|
||||
LEFT JOIN team_members tm ON tm.uuid = p.uuid
|
||||
LEFT JOIN teams t ON t.team_id = COALESCE(tm.team_id, p.uuid)
|
||||
LIMIT 1`,
|
||||
);
|
||||
const teamId = team[0]!.team_id;
|
||||
const [questCount] = await q<{ completed: number; total: number }>(
|
||||
`SELECT (SELECT count(*) FROM quest_progress qp WHERE qp.team_id = $1 AND ${QUEST_FILTER})::int AS completed,
|
||||
(SELECT count(*) FROM quest_defs)::int AS total`,
|
||||
[teamId],
|
||||
);
|
||||
const chapters = await q<{ chapter_id: string; title: string | null; completed: number; total: number }>(
|
||||
`SELECT d.chapter_id, max(d.chapter_title) AS title, count(qp.quest_id)::int AS completed, count(*)::int AS total
|
||||
FROM quest_defs d LEFT JOIN quest_progress qp ON qp.quest_id = d.quest_id AND qp.team_id = $1
|
||||
GROUP BY d.chapter_id ORDER BY max(d.chapter_title)`,
|
||||
[teamId],
|
||||
);
|
||||
const recentQuests = await q<{ quest_id: string; title: string; chapter_title: string | null; completed_at: Date }>(
|
||||
`SELECT d.quest_id, d.title, d.chapter_title, qp.completed_at
|
||||
FROM quest_progress qp JOIN quest_defs d ON d.quest_id = qp.quest_id
|
||||
WHERE qp.team_id = $1 AND d.title IS NOT NULL
|
||||
ORDER BY qp.completed_at DESC LIMIT 10`,
|
||||
[teamId],
|
||||
);
|
||||
|
||||
const heat = await q<{ dow: number; hour: number; minutes: number }>(
|
||||
`SELECT (extract(isodow FROM h AT TIME ZONE $2)::int - 1) AS dow, extract(hour FROM h AT TIME ZONE $2)::int AS hour,
|
||||
sum(extract(epoch FROM LEAST(COALESCE(s.ended_at, now()), h + interval '1 hour') - GREATEST(s.started_at, h)) / 60) AS minutes
|
||||
FROM sessions s,
|
||||
generate_series(date_trunc('hour', s.started_at), COALESCE(s.ended_at, now()), interval '1 hour') AS h
|
||||
WHERE s.uuid = $1 AND s.started_at > now() - interval '90 days'
|
||||
GROUP BY 1, 2`,
|
||||
[uuid, timeZone],
|
||||
);
|
||||
const activityHeatmap = Array.from({ length: 7 }, () => Array.from({ length: 24 }, () => 0));
|
||||
for (const r of heat) activityHeatmap[r.dow]![r.hour] = Math.round(r.minutes);
|
||||
|
||||
return {
|
||||
uuid,
|
||||
name: player!.name,
|
||||
firstSeen: player!.first_seen.getTime(),
|
||||
lastSeen: ms(player!.last_seen),
|
||||
online: sess?.online ?? false,
|
||||
playHours: playTicks !== undefined ? hours(playTicks) : Math.round(((sess?.minutes ?? 0) / 60) * 100) / 100,
|
||||
sessions: sess?.n ?? 0,
|
||||
ranks: await getPlayerRanks(pool, uuid),
|
||||
general,
|
||||
movement,
|
||||
deathsByCause: (
|
||||
await q<{ cause: string; count: number }>(
|
||||
'SELECT cause, count(*)::int AS count FROM deaths WHERE uuid = $1 GROUP BY cause ORDER BY count DESC, cause',
|
||||
)
|
||||
).map((r) => ({ cause: r.cause, count: r.count })),
|
||||
killedBy: (await counted('minecraft:killed_by')).map((r) => ({ entity: r.key, count: r.value })),
|
||||
killsByType: (await counted('minecraft:killed')).map((r) => ({ entity: r.key, count: r.value })),
|
||||
bossKills: (
|
||||
await q<{ victim_type: string; ts: Date }>('SELECT victim_type, ts FROM kills WHERE uuid = $1 AND is_boss ORDER BY ts DESC LIMIT 50')
|
||||
).map((r) => ({ entity: r.victim_type, at: r.ts.getTime() })),
|
||||
dimensions: (
|
||||
await q<{ dimension: string; first_at: Date }>('SELECT dimension, first_at FROM dimension_visits WHERE uuid = $1 ORDER BY first_at')
|
||||
).map((r) => ({ dimension: r.dimension, firstAt: r.first_at.getTime() })),
|
||||
advancements: (
|
||||
await q<{ advancement_id: string; title: string | null; achieved_at: Date }>(
|
||||
'SELECT advancement_id, title, achieved_at FROM advancements WHERE uuid = $1 ORDER BY achieved_at DESC LIMIT 200',
|
||||
)
|
||||
).map((r) => ({ id: r.advancement_id, title: r.title, at: r.achieved_at.getTime() })),
|
||||
quests: {
|
||||
teamId,
|
||||
teamName: team[0]!.name,
|
||||
completed: questCount!.completed,
|
||||
total: questCount!.total > 0 ? questCount!.total : null,
|
||||
chapters: chapters.map((c) => ({ chapterId: c.chapter_id, title: c.title, completed: c.completed, total: c.total })),
|
||||
recent: recentQuests.map((r) => ({ questId: r.quest_id, title: r.title, chapterTitle: r.chapter_title, at: r.completed_at.getTime() })),
|
||||
},
|
||||
activityHeatmap,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
import type { ModsDto, QuestsDto, ServerDto, ServerInfoDto, SourcesDto } from '@atm/shared/api';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Pool } from 'pg';
|
||||
import type { Config } from '../../config.ts';
|
||||
import { MOD_CATEGORIES } from '../../stats/metrics.ts';
|
||||
import { LIMITATIONS, ms, QUEST_FILTER } from './shared.ts';
|
||||
|
||||
const RANGE_HOURS = { '24h': 24, '7d': 168, '30d': 720 } as const;
|
||||
const PUSH_FRESH_MS = 10 * 60_000;
|
||||
|
||||
const categoryOf = (ns: string): string | null =>
|
||||
(Object.entries(MOD_CATEGORIES).find(([, list]) => list.includes(ns))?.[0] as string | undefined) ?? null;
|
||||
|
||||
async function latestServerInfo(pool: Pool): Promise<{ info: ServerInfoDto; mods: { id: string; name: string; version: string }[] } | null> {
|
||||
const { rows } = await pool.query('SELECT * FROM server_info ORDER BY started_at DESC LIMIT 1');
|
||||
const r = rows[0];
|
||||
if (!r) return null;
|
||||
const mods = r.mods as { id: string; name: string; version: string }[];
|
||||
return {
|
||||
info: { startedAt: r.started_at.getTime(), mcVersion: r.mc_version, neoforgeVersion: r.neoforge_version, javaVersion: r.java_version, modCount: mods.length },
|
||||
mods,
|
||||
};
|
||||
}
|
||||
|
||||
export async function serverRoutes(app: FastifyInstance, { pool, config }: { pool: Pool; config: Config }) {
|
||||
app.get('/api/mods', async (): Promise<ModsDto> => {
|
||||
const latest = await latestServerInfo(pool);
|
||||
const { rows } = await pool.query<{ ns: string; activity: number }>(
|
||||
`SELECT split_part(key, ':', 1) AS ns, sum(value) AS activity
|
||||
FROM stat_current sc JOIN players p ON p.uuid = sc.uuid AND NOT p.hidden
|
||||
WHERE category IN ('minecraft:mined', 'minecraft:crafted', 'minecraft:used', 'minecraft:picked_up')
|
||||
GROUP BY 1`,
|
||||
);
|
||||
const activity = new Map(rows.map((r) => [r.ns, r.activity]));
|
||||
const known = latest?.mods ?? [];
|
||||
const mods = known
|
||||
.map((m) => ({ id: m.id, name: m.name, version: m.version, activity: activity.get(m.id) ?? 0, category: categoryOf(m.id) }))
|
||||
.sort((a, b) => b.activity - a.activity || a.name.localeCompare(b.name));
|
||||
const byCategory = new Map<string, number>();
|
||||
for (const [ns, value] of activity) {
|
||||
if (ns === 'minecraft') continue;
|
||||
const cat = categoryOf(ns) ?? 'other';
|
||||
byCategory.set(cat, (byCategory.get(cat) ?? 0) + value);
|
||||
}
|
||||
return {
|
||||
server: latest?.info ?? null,
|
||||
mods,
|
||||
categories: [...byCategory].map(([category, a]) => ({ category, activity: a })).sort((a, b) => b.activity - a.activity),
|
||||
};
|
||||
});
|
||||
|
||||
app.get<{ Querystring: { range?: string } }>('/api/server', async (req, reply): Promise<ServerDto | undefined> => {
|
||||
const hours = RANGE_HOURS[(req.query.range ?? '24h') as keyof typeof RANGE_HOURS];
|
||||
if (!hours) return reply.code(400).send({ error: 'bad_range' });
|
||||
// Minute values for the last 14 days, hourly aggregates beyond; long ranges are bucketed to keep payloads small.
|
||||
const step = hours <= 24 ? 'minute' : 'hour';
|
||||
const { rows } = await pool.query(
|
||||
`SELECT date_trunc($2, bucket) AS t, avg(mspt_avg) AS mspt_avg, max(mspt_p95) AS mspt_p95, avg(tps) AS tps,
|
||||
max(players) AS players, avg(heap_used_mb) AS heap_used_mb, max(heap_max_mb) AS heap_max_mb
|
||||
FROM (SELECT bucket, mspt_avg, mspt_p95, tps, players, heap_used_mb, heap_max_mb FROM server_health
|
||||
UNION ALL
|
||||
SELECT bucket, mspt_avg, mspt_p95, tps, players, heap_used_mb, heap_max_mb FROM server_health_hourly) h
|
||||
WHERE bucket >= now() - make_interval(hours => $1)
|
||||
GROUP BY 1 ORDER BY 1`,
|
||||
[hours, step],
|
||||
);
|
||||
return {
|
||||
info: (await latestServerInfo(pool))?.info ?? null,
|
||||
health: rows.map((r) => ({
|
||||
t: r.t.getTime(),
|
||||
msptAvg: Math.round(r.mspt_avg * 10) / 10,
|
||||
msptP95: Math.round(r.mspt_p95 * 10) / 10,
|
||||
tps: Math.round(r.tps * 10) / 10,
|
||||
players: r.players,
|
||||
heapUsedMb: Math.round(r.heap_used_mb),
|
||||
heapMaxMb: Math.round(r.heap_max_mb),
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
app.get('/api/quests', async (): Promise<QuestsDto> => {
|
||||
const defs = await pool.query<{ n: number }>('SELECT count(*)::int AS n FROM quest_defs');
|
||||
const chapters = await pool.query(
|
||||
`WITH per_team AS (
|
||||
SELECT d.chapter_id, qp.team_id, count(*) AS done
|
||||
FROM quest_defs d JOIN quest_progress qp ON qp.quest_id = d.quest_id GROUP BY 1, 2),
|
||||
totals AS (SELECT chapter_id, max(chapter_title) AS title, count(*)::int AS total FROM quest_defs GROUP BY 1)
|
||||
SELECT t.chapter_id, t.title, t.total,
|
||||
COALESCE(max(pt.done), 0)::int AS max_completed,
|
||||
count(pt.team_id) FILTER (WHERE pt.done >= t.total)::int AS teams_finished
|
||||
FROM totals t LEFT JOIN per_team pt ON pt.chapter_id = t.chapter_id
|
||||
GROUP BY t.chapter_id, t.title, t.total ORDER BY t.title NULLS LAST`,
|
||||
);
|
||||
// Teams: explicit members, or the solo player whose uuid is the team id. Teams of hidden players are omitted.
|
||||
const teams = await pool.query(
|
||||
`SELECT qp.team_id, t.name, count(*)::int AS completed,
|
||||
COALESCE(
|
||||
(SELECT json_agg(json_build_object('uuid', p.uuid, 'name', p.name)) FROM team_members tm
|
||||
JOIN players p ON p.uuid = tm.uuid AND NOT p.hidden WHERE tm.team_id = qp.team_id),
|
||||
(SELECT json_agg(json_build_object('uuid', p.uuid, 'name', p.name)) FROM players p
|
||||
WHERE p.uuid = qp.team_id AND NOT p.hidden)) AS members
|
||||
FROM quest_progress qp LEFT JOIN teams t ON t.team_id = qp.team_id
|
||||
WHERE ${QUEST_FILTER}
|
||||
GROUP BY qp.team_id, t.name ORDER BY completed DESC`,
|
||||
);
|
||||
return {
|
||||
defsKnown: defs.rows[0]!.n > 0,
|
||||
chapters: chapters.rows.map((r) => ({
|
||||
chapterId: r.chapter_id,
|
||||
title: r.title,
|
||||
total: r.total,
|
||||
maxCompleted: r.max_completed,
|
||||
teamsFinished: r.teams_finished,
|
||||
})),
|
||||
teams: teams.rows
|
||||
.filter((r) => r.members)
|
||||
.map((r) => ({ teamId: r.team_id, name: r.name, members: r.members, completed: r.completed })),
|
||||
};
|
||||
});
|
||||
|
||||
app.get('/api/sources', async (): Promise<SourcesDto> => {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT (SELECT max(received_at) FROM events WHERE source = 'push') AS last_push,
|
||||
(SELECT max(started_at) FROM server_info) AS last_start,
|
||||
(SELECT max(synced_at) FROM sync_state) AS last_sync,
|
||||
(SELECT message FROM app_errors WHERE source = 'sftp' AND at > now() - interval '1 day' ORDER BY at DESC LIMIT 1) AS sftp_error`,
|
||||
);
|
||||
const r = rows[0];
|
||||
const lastPush = ms(r.last_push);
|
||||
return {
|
||||
push: { lastEventAt: lastPush, lastServerStart: ms(r.last_start), reachable: lastPush !== null && Date.now() - lastPush < PUSH_FRESH_MS },
|
||||
sftp: { enabled: config.sftp !== null, lastSyncAt: ms(r.last_sync), lastError: r.sftp_error },
|
||||
limitations: LIMITATIONS,
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import type { Pool } from 'pg';
|
||||
|
||||
export const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
|
||||
export const PLAYTIME_TICKS_PER_HOUR = 72_000;
|
||||
|
||||
export const ms = (d: Date | null | undefined): number | null => (d ? d.getTime() : null);
|
||||
export const hours = (ticks: number) => Math.round((ticks / PLAYTIME_TICKS_PER_HOUR) * 100) / 100;
|
||||
|
||||
/** Resolves uuid, current name or a former name (newest first) to a visible player's uuid. */
|
||||
export async function resolvePlayer(pool: Pool, idOrName: string): Promise<string | null> {
|
||||
if (UUID_RE.test(idOrName)) {
|
||||
const { rows } = await pool.query('SELECT uuid FROM players WHERE uuid = $1 AND NOT hidden', [idOrName]);
|
||||
return rows[0]?.uuid ?? null;
|
||||
}
|
||||
if (!/^[A-Za-z0-9_]{1,16}$/.test(idOrName)) return null;
|
||||
const { rows } = await pool.query(
|
||||
`SELECT uuid FROM (
|
||||
SELECT uuid, 0 AS prio, last_seen AS seen FROM players WHERE lower(name) = lower($1)
|
||||
UNION ALL
|
||||
SELECT uuid, 1, seen_at FROM name_history WHERE lower(name) = lower($1)
|
||||
) m JOIN players p USING (uuid)
|
||||
WHERE NOT p.hidden
|
||||
ORDER BY prio, seen DESC NULLS LAST LIMIT 1`,
|
||||
[idOrName],
|
||||
);
|
||||
return rows[0]?.uuid ?? null;
|
||||
}
|
||||
|
||||
/** Quest IDs that count: all completed IDs, or only known quests once definitions exist. */
|
||||
export const QUEST_FILTER = `(NOT EXISTS (SELECT 1 FROM quest_defs) OR qp.quest_id IN (SELECT quest_id FROM quest_defs))`;
|
||||
|
||||
export const LIMITATIONS: string[] = [
|
||||
'Energieproduktion und -verbrauch von Maschinen (Mekanism, Powah …) werden nicht erfasst: Es gibt dafür keine verlässliche serverseitige Quelle ohne ungeprüfte Mod-Interna.',
|
||||
'Inhalte von AE2- und Refined-Storage-Netzwerken werden nicht ausgewertet.',
|
||||
'Mana, Zauber und Rituale (Ars Nouveau, Iron’s Spellbooks, Occultism) sind nicht verfügbar; Magie-Aktivität stammt aus hergestellten und benutzten Items.',
|
||||
'MineColonies-Kolonien werden nicht ausgewertet.',
|
||||
'Die Diamantenbank zählt abgebaute Diamanterze, nicht den Besitz.',
|
||||
'Quest-Fortschritt wird pro FTB-Team gespeichert: Mitglieder einer Party teilen sich ihren Fortschritt, wie im Spiel.',
|
||||
'Rund ein Drittel der ATM10-Quests hat keinen eigenen Titel (im Spiel heißen sie nach ihrem Aufgaben-Item). Sie zählen mit, erscheinen aber nicht namentlich.',
|
||||
'Zähler aus world/stats werden vom Server nur beim Speichern geschrieben; dazwischen liefert die Mod alle 5 Minuten Live-Werte.',
|
||||
'Todesorte werden aus Datenschutzgründen nicht gespeichert.',
|
||||
];
|
||||
@@ -0,0 +1,19 @@
|
||||
import { buildApp } from './app.ts';
|
||||
import { loadConfig } from './config.ts';
|
||||
import { createPool } from './db/pool.ts';
|
||||
import { migrate } from './db/migrate.ts';
|
||||
|
||||
const config = loadConfig();
|
||||
const pool = createPool(config.DATABASE_URL);
|
||||
await migrate(pool);
|
||||
const app = await buildApp({ pool, config });
|
||||
|
||||
const shutdown = async () => {
|
||||
await app.close();
|
||||
await pool.end();
|
||||
process.exit(0);
|
||||
};
|
||||
process.on('SIGTERM', shutdown);
|
||||
process.on('SIGINT', shutdown);
|
||||
|
||||
await app.listen({ port: config.PORT, host: config.HOST });
|
||||
@@ -0,0 +1,71 @@
|
||||
import type { Pool } from 'pg';
|
||||
import { BOARD_BY_ID, BOARDS, metricSql, toDisplay, type Board } from './metrics.ts';
|
||||
|
||||
export type Period = 'all' | '30d' | '7d';
|
||||
export const PERIOD_DAYS: Record<Exclude<Period, 'all'>, number> = { '30d': 30, '7d': 7 };
|
||||
|
||||
export interface LeaderboardEntry {
|
||||
rank: number;
|
||||
uuid: string;
|
||||
name: string | null;
|
||||
value: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Value per visible player for a board. For a period the value is the growth since the period start:
|
||||
* current value minus the last daily snapshot at or before the start (or the earliest snapshot if none).
|
||||
*/
|
||||
function boardValuesSql(board: Board, period: Period): string {
|
||||
const cur = metricSql(board.metric);
|
||||
if (period === 'all') {
|
||||
return `SELECT c.uuid, c.value FROM (${cur}) c`;
|
||||
}
|
||||
return `SELECT c.uuid, c.value - COALESCE(
|
||||
(SELECT s.value FROM stat_snapshots s WHERE s.uuid = c.uuid AND s.metric = $1 AND s.granularity = 'day'
|
||||
AND s.bucket <= $2 ORDER BY s.bucket DESC LIMIT 1),
|
||||
(SELECT s.value FROM stat_snapshots s WHERE s.uuid = c.uuid AND s.metric = $1 AND s.granularity = 'day'
|
||||
ORDER BY s.bucket ASC LIMIT 1),
|
||||
0) AS value
|
||||
FROM (${cur}) c`;
|
||||
}
|
||||
|
||||
export async function getLeaderboard(
|
||||
pool: Pool,
|
||||
boardId: string,
|
||||
period: Period,
|
||||
limit = 10,
|
||||
now: Date = new Date(),
|
||||
): Promise<LeaderboardEntry[]> {
|
||||
const board = BOARD_BY_ID.get(boardId);
|
||||
if (!board) throw new Error(`unknown board ${boardId}`);
|
||||
const params: unknown[] = [];
|
||||
if (period !== 'all') params.push(board.id, new Date(now.getTime() - PERIOD_DAYS[period] * 86_400_000));
|
||||
params.push(limit);
|
||||
const { rows } = await pool.query<{ rank: number; uuid: string; name: string | null; value: number }>(
|
||||
`SELECT rank() OVER (ORDER BY v.value DESC)::int AS rank, v.uuid, p.name, v.value
|
||||
FROM (${boardValuesSql(board, period)}) v
|
||||
JOIN players p ON p.uuid = v.uuid AND NOT p.hidden
|
||||
WHERE v.value > 0
|
||||
ORDER BY v.value DESC, p.name ASC NULLS LAST
|
||||
LIMIT $${params.length}`,
|
||||
params,
|
||||
);
|
||||
return rows.map((r) => ({ ...r, value: toDisplay(board.unit, r.value) }));
|
||||
}
|
||||
|
||||
/** Rank and display value of one player on every board where the player has a positive value. */
|
||||
export async function getPlayerRanks(pool: Pool, uuid: string): Promise<Record<string, { rank: number; value: number }>> {
|
||||
const out: Record<string, { rank: number; value: number }> = {};
|
||||
for (const board of BOARDS) {
|
||||
const { rows } = await pool.query<{ rank: number; value: number }>(
|
||||
`SELECT rank, value FROM (
|
||||
SELECT v.uuid, v.value, rank() OVER (ORDER BY v.value DESC)::int AS rank
|
||||
FROM (${metricSql(board.metric)}) v JOIN players p ON p.uuid = v.uuid AND NOT p.hidden
|
||||
WHERE v.value > 0) r
|
||||
WHERE r.uuid = $1`,
|
||||
[uuid],
|
||||
);
|
||||
if (rows[0]) out[board.id] = { rank: rows[0].rank, value: toDisplay(board.unit, rows[0].value) };
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
export type MetricDef =
|
||||
| { kind: 'stat'; category: string; key: string }
|
||||
| { kind: 'sum'; categories: string[]; namespaces?: string[]; keySuffix?: string; excludeKeys?: string[] }
|
||||
| { kind: 'sumKeys'; category: string; keys: string[] }
|
||||
| { kind: 'quests' }
|
||||
| { kind: 'advancements' };
|
||||
|
||||
export type Unit = 'hours' | 'count' | 'km' | 'blocks';
|
||||
|
||||
export interface Board {
|
||||
id: string;
|
||||
title: string;
|
||||
description: string;
|
||||
unit: Unit;
|
||||
metric: MetricDef;
|
||||
}
|
||||
|
||||
export type ModCategories = Record<'tech' | 'magic' | 'farming', string[]>;
|
||||
|
||||
function loadCategories(): ModCategories {
|
||||
const here = dirname(fileURLToPath(import.meta.url));
|
||||
for (const p of [join(here, '..', '..', 'config', 'mod-categories.json'), join(here, '..', 'config', 'mod-categories.json')]) {
|
||||
try {
|
||||
return JSON.parse(readFileSync(p, 'utf8')) as ModCategories;
|
||||
} catch {
|
||||
/* try next */
|
||||
}
|
||||
}
|
||||
throw new Error('config/mod-categories.json not found');
|
||||
}
|
||||
|
||||
export const MOD_CATEGORIES: ModCategories = loadCategories();
|
||||
|
||||
const CUSTOM = 'minecraft:custom';
|
||||
const ACTIVITY = ['minecraft:crafted', 'minecraft:used'];
|
||||
|
||||
export const BOARDS: Board[] = [
|
||||
{ id: 'playtime', title: 'Dauerzocker', description: 'Meiste Spielzeit', unit: 'hours', metric: { kind: 'stat', category: CUSTOM, key: 'minecraft:play_time' } },
|
||||
{ id: 'deaths', title: 'Friedhof', description: 'Häufigste Tode', unit: 'count', metric: { kind: 'stat', category: CUSTOM, key: 'minecraft:deaths' } },
|
||||
{ id: 'mobkills', title: 'Monsterjäger', description: 'Meiste getötete Mobs', unit: 'count', metric: { kind: 'stat', category: CUSTOM, key: 'minecraft:mob_kills' } },
|
||||
{ id: 'mined', title: 'Abrissbirne', description: 'Meiste abgebaute Blöcke', unit: 'blocks', metric: { kind: 'sum', categories: ['minecraft:mined'] } },
|
||||
{
|
||||
id: 'distance',
|
||||
title: 'Weltenbummler',
|
||||
description: 'Weiteste zurückgelegte Strecke (ohne Fallen)',
|
||||
unit: 'km',
|
||||
metric: { kind: 'sum', categories: [CUSTOM], keySuffix: '_one_cm', excludeKeys: ['minecraft:fall_one_cm'] },
|
||||
},
|
||||
{ id: 'flight', title: 'Flieger', description: 'Weiteste Strecke im Flug', unit: 'km', metric: { kind: 'stat', category: CUSTOM, key: 'minecraft:fly_one_cm' } },
|
||||
{
|
||||
id: 'diamonds',
|
||||
title: 'Diamantenbank',
|
||||
description: 'Meiste abgebaute Diamanterze (Abbau, nicht Besitz)',
|
||||
unit: 'count',
|
||||
metric: { kind: 'sumKeys', category: 'minecraft:mined', keys: ['minecraft:diamond_ore', 'minecraft:deepslate_diamond_ore'] },
|
||||
},
|
||||
{ id: 'tech', title: 'Techniker', description: 'Hergestellte und benutzte Tech-Items', unit: 'count', metric: { kind: 'sum', categories: ACTIVITY, namespaces: MOD_CATEGORIES.tech } },
|
||||
{ id: 'magic', title: 'Magier', description: 'Hergestellte und benutzte Magie-Items', unit: 'count', metric: { kind: 'sum', categories: ACTIVITY, namespaces: MOD_CATEGORIES.magic } },
|
||||
{ id: 'farming', title: 'Farmer', description: 'Hergestellte und benutzte Farm-Items', unit: 'count', metric: { kind: 'sum', categories: ACTIVITY, namespaces: MOD_CATEGORIES.farming } },
|
||||
{ id: 'quests', title: 'Questmeister', description: 'Abgeschlossene FTB-Quests des Teams', unit: 'count', metric: { kind: 'quests' } },
|
||||
{ id: 'advancements', title: 'Abenteurer', description: 'Erreichte Advancements', unit: 'count', metric: { kind: 'advancements' } },
|
||||
{ id: 'loot', title: 'Schatzjäger', description: 'Geplünderte Lootr-Truhen', unit: 'count', metric: { kind: 'stat', category: CUSTOM, key: 'lootr:looted_stat' } },
|
||||
{ id: 'worldtier', title: 'Weltenstufe', description: 'Aktivierte Apotheosis-Weltenstufen', unit: 'count', metric: { kind: 'stat', category: CUSTOM, key: 'apotheosis:world_tiers_activated' } },
|
||||
];
|
||||
|
||||
export const BOARD_BY_ID = new Map(BOARDS.map((b) => [b.id, b]));
|
||||
|
||||
/** Board definitions are code constants, never user input; literals are still escaped. */
|
||||
const lit = (s: string) => `'${s.replace(/'/g, "''")}'`;
|
||||
const list = (xs: string[]) => xs.map(lit).join(', ');
|
||||
|
||||
/** SQL yielding `(uuid, value)` with the raw metric value for every player that has data. */
|
||||
export function metricSql(def: MetricDef): string {
|
||||
switch (def.kind) {
|
||||
case 'stat':
|
||||
return `SELECT uuid, value FROM stat_current WHERE category = ${lit(def.category)} AND key = ${lit(def.key)}`;
|
||||
case 'sumKeys':
|
||||
return `SELECT uuid, sum(value) AS value FROM stat_current
|
||||
WHERE category = ${lit(def.category)} AND key IN (${list(def.keys)}) GROUP BY uuid`;
|
||||
case 'sum': {
|
||||
const where = [`category IN (${list(def.categories)})`];
|
||||
if (def.namespaces?.length) where.push(`split_part(key, ':', 1) IN (${list(def.namespaces)})`);
|
||||
if (def.keySuffix) where.push(`key LIKE ${lit(`%${def.keySuffix}`)}`);
|
||||
if (def.excludeKeys?.length) where.push(`key NOT IN (${list(def.excludeKeys)})`);
|
||||
return `SELECT uuid, sum(value) AS value FROM stat_current WHERE ${where.join(' AND ')} GROUP BY uuid`;
|
||||
}
|
||||
case 'quests':
|
||||
// FTB stores progress per team; players without team membership form a solo team with their own uuid.
|
||||
// The `completed` map also lists chapters, so once quest definitions are known only real quests count.
|
||||
return `SELECT p.uuid, count(DISTINCT qp.quest_id) AS value
|
||||
FROM players p
|
||||
LEFT JOIN team_members tm ON tm.uuid = p.uuid
|
||||
JOIN quest_progress qp ON qp.team_id = COALESCE(tm.team_id, p.uuid)
|
||||
WHERE NOT EXISTS (SELECT 1 FROM quest_defs) OR qp.quest_id IN (SELECT quest_id FROM quest_defs)
|
||||
GROUP BY p.uuid`;
|
||||
case 'advancements':
|
||||
return `SELECT uuid, count(*) AS value FROM advancements GROUP BY uuid`;
|
||||
}
|
||||
}
|
||||
|
||||
/** Raw counter → display value for the board's unit. */
|
||||
export function toDisplay(unit: Unit, raw: number): number {
|
||||
switch (unit) {
|
||||
case 'hours':
|
||||
return Math.round((raw / 72_000) * 100) / 100;
|
||||
case 'km':
|
||||
return Math.round((raw / 100_000) * 100) / 100;
|
||||
default:
|
||||
return raw;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
import type { Pool } from 'pg';
|
||||
import { BOARDS, metricSql } from './metrics.ts';
|
||||
|
||||
export const HOURLY_RETENTION_DAYS = 30;
|
||||
export const HEALTH_MINUTE_RETENTION_DAYS = 14;
|
||||
|
||||
/** Stores every board's raw value per player for the hour/day bucket containing `at`. Re-running overwrites. */
|
||||
export async function takeSnapshots(pool: Pool, granularity: 'hour' | 'day', at: Date): Promise<number> {
|
||||
let written = 0;
|
||||
for (const board of BOARDS) {
|
||||
const { rowCount } = await pool.query(
|
||||
`INSERT INTO stat_snapshots(uuid, metric, bucket, granularity, value)
|
||||
SELECT v.uuid, $1, date_trunc($2, $3::timestamptz), $2, v.value FROM (${metricSql(board.metric)}) v
|
||||
ON CONFLICT (uuid, metric, granularity, bucket) DO UPDATE SET value = EXCLUDED.value`,
|
||||
[board.id, granularity, at],
|
||||
);
|
||||
written += rowCount ?? 0;
|
||||
}
|
||||
return written;
|
||||
}
|
||||
|
||||
export const BULKY_EVENT_RETENTION_DAYS = 30;
|
||||
/** Already projected, large or frequent; their content lives on in stat_current, server_health and file_cache. */
|
||||
const BULKY_EVENT_TYPES = ['health', 'stats_snapshot', 'file_sync'];
|
||||
|
||||
export async function applyRetention(pool: Pool, now: Date): Promise<void> {
|
||||
await pool.query(`DELETE FROM stat_snapshots WHERE granularity = 'hour' AND bucket < $1`, [
|
||||
new Date(now.getTime() - HOURLY_RETENTION_DAYS * 86_400_000),
|
||||
]);
|
||||
// Keep the raw log of player events (joins, deaths, advancements …); drop bulky ones after 30 days.
|
||||
// Re-delivery after that window is not expected (the mod's outbox expires after 7 days).
|
||||
await pool.query(`DELETE FROM events WHERE type = ANY($1::text[]) AND received_at < $2`, [
|
||||
BULKY_EVENT_TYPES,
|
||||
new Date(now.getTime() - BULKY_EVENT_RETENTION_DAYS * 86_400_000),
|
||||
]);
|
||||
const cutoff = new Date(now.getTime() - HEALTH_MINUTE_RETENTION_DAYS * 86_400_000);
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
await client.query(
|
||||
`INSERT INTO server_health_hourly(bucket, mspt_avg, mspt_p95, mspt_max, tps, players, chunks, heap_used_mb, heap_max_mb, samples)
|
||||
SELECT date_trunc('hour', bucket), avg(mspt_avg), max(mspt_p95), max(mspt_max), avg(tps), max(players), max(chunks),
|
||||
avg(heap_used_mb), max(heap_max_mb), count(*)
|
||||
FROM server_health WHERE bucket < $1 GROUP BY 1
|
||||
ON CONFLICT (bucket) DO UPDATE SET
|
||||
mspt_avg = (server_health_hourly.mspt_avg * server_health_hourly.samples + EXCLUDED.mspt_avg * EXCLUDED.samples)
|
||||
/ (server_health_hourly.samples + EXCLUDED.samples),
|
||||
mspt_p95 = GREATEST(server_health_hourly.mspt_p95, EXCLUDED.mspt_p95),
|
||||
mspt_max = GREATEST(server_health_hourly.mspt_max, EXCLUDED.mspt_max),
|
||||
tps = (server_health_hourly.tps * server_health_hourly.samples + EXCLUDED.tps * EXCLUDED.samples)
|
||||
/ (server_health_hourly.samples + EXCLUDED.samples),
|
||||
players = GREATEST(server_health_hourly.players, EXCLUDED.players),
|
||||
chunks = GREATEST(server_health_hourly.chunks, EXCLUDED.chunks),
|
||||
heap_used_mb = EXCLUDED.heap_used_mb,
|
||||
heap_max_mb = GREATEST(server_health_hourly.heap_max_mb, EXCLUDED.heap_max_mb),
|
||||
samples = server_health_hourly.samples + EXCLUDED.samples`,
|
||||
[cutoff],
|
||||
);
|
||||
await client.query('DELETE FROM server_health WHERE bucket < $1', [cutoff]);
|
||||
await client.query('COMMIT');
|
||||
} catch (err) {
|
||||
await client.query('ROLLBACK');
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
import pino from 'pino';
|
||||
import { loadConfig } from './config.ts';
|
||||
import { createPool } from './db/pool.ts';
|
||||
import { migrate } from './db/migrate.ts';
|
||||
import { installLiveNotifier } from './live/notifier.ts';
|
||||
import './pipeline/projections/index.ts';
|
||||
import { applyRetention, takeSnapshots } from './stats/snapshots.ts';
|
||||
import { SftpFileSource } from './worker/fileSource.ts';
|
||||
import { LOCKS, runJob } from './worker/scheduler.ts';
|
||||
import { closeStaleSessions, syncOnce } from './worker/sync.ts';
|
||||
|
||||
const TICK_MS = 30_000;
|
||||
|
||||
const config = loadConfig();
|
||||
const log = pino({ level: config.LOG_LEVEL });
|
||||
const pool = createPool(config.DATABASE_URL, 4);
|
||||
await migrate(pool);
|
||||
installLiveNotifier(pool);
|
||||
|
||||
let lastSftp = 0;
|
||||
let lastHour = '';
|
||||
let lastRetentionDay = '';
|
||||
let stopping = false;
|
||||
|
||||
async function sftpDue(now: number): Promise<{ due: boolean; requestIds: number[] }> {
|
||||
if (!config.sftp) return { due: false, requestIds: [] };
|
||||
const { rows } = await pool.query<{ id: number }>('SELECT id FROM sync_requests WHERE done_at IS NULL');
|
||||
const requestIds = rows.map((r) => r.id);
|
||||
return { due: requestIds.length > 0 || now - lastSftp >= config.SFTP_INTERVAL_MIN * 60_000, requestIds };
|
||||
}
|
||||
|
||||
async function tick() {
|
||||
const now = new Date();
|
||||
|
||||
const sftp = await sftpDue(now.getTime());
|
||||
if (sftp.due && config.sftp) {
|
||||
lastSftp = now.getTime();
|
||||
const src = new SftpFileSource(config.sftp);
|
||||
const result = await runJob(pool, 'sftp', LOCKS.sftp, async () => {
|
||||
try {
|
||||
const r = await syncOnce(pool, src, { timeZone: config.TIMEZONE, serverId: config.SERVER_ID, worldDir: config.SFTP_WORLD_DIR });
|
||||
log.info(r, 'sftp sync done');
|
||||
} finally {
|
||||
await src.close();
|
||||
}
|
||||
});
|
||||
if (result === 'error') log.warn('sftp sync failed, see admin status');
|
||||
if (sftp.requestIds.length) {
|
||||
await pool.query('UPDATE sync_requests SET done_at = now() WHERE id = ANY($1::bigint[])', [sftp.requestIds]);
|
||||
}
|
||||
}
|
||||
|
||||
// Hourly and daily snapshots are idempotent per bucket; the daily bucket ends up holding the day's last value.
|
||||
const hour = now.toISOString().slice(0, 13);
|
||||
if (hour !== lastHour) {
|
||||
lastHour = hour;
|
||||
await runJob(pool, 'snapshots', LOCKS.snapshots, async () => {
|
||||
await takeSnapshots(pool, 'hour', now);
|
||||
await takeSnapshots(pool, 'day', now);
|
||||
});
|
||||
}
|
||||
|
||||
const day = now.toISOString().slice(0, 10);
|
||||
if (day !== lastRetentionDay && now.getUTCHours() >= 3) {
|
||||
lastRetentionDay = day;
|
||||
await runJob(pool, 'retention', LOCKS.retention, () => applyRetention(pool, now));
|
||||
}
|
||||
|
||||
await runJob(pool, 'staleSessions', LOCKS.staleSessions, async () => {
|
||||
const closed = await closeStaleSessions(pool, now);
|
||||
if (closed) log.info({ closed }, 'closed stale sessions');
|
||||
});
|
||||
}
|
||||
|
||||
async function loop() {
|
||||
log.info({ sftp: config.sftp ? `${config.sftp.host}:${config.sftp.port}` : 'disabled' }, 'worker started');
|
||||
while (!stopping) {
|
||||
try {
|
||||
await tick();
|
||||
} catch (err) {
|
||||
log.error({ err }, 'worker tick failed');
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, TICK_MS));
|
||||
}
|
||||
await pool.end();
|
||||
}
|
||||
|
||||
const stop = () => {
|
||||
stopping = true;
|
||||
setTimeout(() => process.exit(0), 5000).unref();
|
||||
};
|
||||
process.on('SIGTERM', stop);
|
||||
process.on('SIGINT', stop);
|
||||
|
||||
await loop();
|
||||
@@ -0,0 +1,104 @@
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { readdir, stat } from 'node:fs/promises';
|
||||
import { join, posix } from 'node:path';
|
||||
import SftpClient from 'ssh2-sftp-client';
|
||||
|
||||
export interface FileEntry {
|
||||
name: string;
|
||||
size: number;
|
||||
mtimeMs: number;
|
||||
isDir: boolean;
|
||||
}
|
||||
|
||||
/** Read-only view of the game server's directory tree. Paths are relative and '/'-separated. */
|
||||
export interface FileSource {
|
||||
list(dir: string): Promise<FileEntry[]>;
|
||||
/** Reads `length` bytes from `start` (default: to end of file). */
|
||||
read(path: string, start?: number, length?: number): Promise<Buffer>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
async function collect(stream: NodeJS.ReadableStream): Promise<Buffer> {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const c of stream) chunks.push(Buffer.isBuffer(c) ? c : Buffer.from(c as string));
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
export class LocalFileSource implements FileSource {
|
||||
constructor(private readonly root: string) {}
|
||||
|
||||
async list(dir: string): Promise<FileEntry[]> {
|
||||
const abs = join(this.root, dir);
|
||||
let names: string[];
|
||||
try {
|
||||
names = await readdir(abs);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
return Promise.all(
|
||||
names.map(async (name) => {
|
||||
const s = await stat(join(abs, name));
|
||||
return { name, size: s.size, mtimeMs: Math.floor(s.mtimeMs), isDir: s.isDirectory() };
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async read(path: string, start = 0, length?: number): Promise<Buffer> {
|
||||
const end = length !== undefined ? start + length - 1 : undefined;
|
||||
return collect(createReadStream(join(this.root, path), { start, end }));
|
||||
}
|
||||
|
||||
async close(): Promise<void> {}
|
||||
}
|
||||
|
||||
export interface SftpOptions {
|
||||
host: string;
|
||||
port: number;
|
||||
username: string;
|
||||
password: string;
|
||||
root: string;
|
||||
}
|
||||
|
||||
/** SFTP access to the Mine-Hoster server; one connection per sync run. */
|
||||
export class SftpFileSource implements FileSource {
|
||||
private client = new SftpClient('atm-stats');
|
||||
private connected: Promise<unknown> | null = null;
|
||||
|
||||
constructor(private readonly opts: SftpOptions) {}
|
||||
|
||||
private abs(p: string) {
|
||||
return posix.join(this.opts.root || '/', p);
|
||||
}
|
||||
|
||||
private connect() {
|
||||
this.connected ??= this.client.connect({
|
||||
host: this.opts.host,
|
||||
port: this.opts.port,
|
||||
username: this.opts.username,
|
||||
password: this.opts.password,
|
||||
readyTimeout: 20_000,
|
||||
retries: 1,
|
||||
});
|
||||
return this.connected;
|
||||
}
|
||||
|
||||
async list(dir: string): Promise<FileEntry[]> {
|
||||
await this.connect();
|
||||
const path = this.abs(dir);
|
||||
if (!(await this.client.exists(path))) return [];
|
||||
const entries = await this.client.list(path);
|
||||
return entries.map((e) => ({ name: e.name, size: e.size, mtimeMs: e.modifyTime, isDir: e.type === 'd' }));
|
||||
}
|
||||
|
||||
async read(path: string, start = 0, length?: number): Promise<Buffer> {
|
||||
await this.connect();
|
||||
const end = length !== undefined ? start + length - 1 : undefined;
|
||||
const stream = this.client.createReadStream(this.abs(path), { start, end });
|
||||
return collect(stream);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.connected) await this.client.end().catch(() => {});
|
||||
this.connected = null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
import type { Pool } from 'pg';
|
||||
|
||||
export type JobResult = 'ok' | 'locked' | 'error';
|
||||
|
||||
/**
|
||||
* Runs `fn` while holding a session-level advisory lock, so a job never runs twice at once
|
||||
* (e.g. two worker containers). Failures are logged to app_errors with the job name as source.
|
||||
*/
|
||||
export async function runJob(pool: Pool, name: string, lockId: number, fn: () => Promise<unknown>): Promise<JobResult> {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
const { rows } = await client.query<{ locked: boolean }>('SELECT pg_try_advisory_lock($1) AS locked', [lockId]);
|
||||
if (!rows[0]!.locked) return 'locked';
|
||||
try {
|
||||
await fn();
|
||||
return 'ok';
|
||||
} catch (err) {
|
||||
await pool
|
||||
.query('INSERT INTO app_errors(source, message) VALUES ($1, $2)', [name, String((err as Error).message ?? err).slice(0, 2000)])
|
||||
.catch(() => {});
|
||||
return 'error';
|
||||
} finally {
|
||||
await client.query('SELECT pg_advisory_unlock($1)', [lockId]);
|
||||
}
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
export const LOCKS = {
|
||||
sftp: 820_001,
|
||||
snapshots: 820_002,
|
||||
retention: 820_003,
|
||||
staleSessions: 820_004,
|
||||
} as const;
|
||||
@@ -0,0 +1,172 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { deterministicId, type AtmEvent } from '@atm/shared';
|
||||
import type { Pool } from 'pg';
|
||||
import { isQuestConfigFile } from '../parsers/ftbquests.ts';
|
||||
import { parseLogChunk } from '../parsers/log.ts';
|
||||
import { advancementsFileToEvents, statsFileToEvent } from '../parsers/vanilla.ts';
|
||||
import { processEvents } from '../pipeline/process.ts';
|
||||
import type { FileEntry, FileSource } from './fileSource.ts';
|
||||
|
||||
export interface SyncOptions {
|
||||
timeZone: string;
|
||||
serverId: string;
|
||||
worldDir: string;
|
||||
}
|
||||
|
||||
const UUID_FILE = /^([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.(json|snbt)$/i;
|
||||
const PUSH_FRESH_MS = 10 * 60_000;
|
||||
const LOG_PATH = 'logs/latest.log';
|
||||
const LOG_HEAD_BYTES = 256;
|
||||
const MAX_FILE_BYTES = 1_500_000;
|
||||
const BATCH = 200;
|
||||
|
||||
const sha256 = (b: Buffer | string) => createHash('sha256').update(b).digest('hex');
|
||||
|
||||
interface State {
|
||||
size: number;
|
||||
mtime: number;
|
||||
offset: number;
|
||||
hash: string | null;
|
||||
}
|
||||
|
||||
async function listRecursive(src: FileSource, dir: string): Promise<(FileEntry & { path: string })[]> {
|
||||
const out: (FileEntry & { path: string })[] = [];
|
||||
for (const e of await src.list(dir)) {
|
||||
const path = `${dir}/${e.name}`;
|
||||
if (e.isDir) out.push(...(await listRecursive(src, path)));
|
||||
else out.push({ ...e, path });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pulls changed files from the game server and feeds them through the same pipeline as the mod's pushes.
|
||||
* Unchanged files (same size and mtime) are not read; latest.log is read incrementally from the stored offset.
|
||||
*/
|
||||
export async function syncOnce(pool: Pool, src: FileSource, opts: SyncOptions): Promise<{ files: number; events: number }> {
|
||||
const stateRows = await pool.query<State & { path: string }>('SELECT path, size, mtime, "offset", hash FROM sync_state');
|
||||
const state = new Map(stateRows.rows.map((r) => [r.path, r as State]));
|
||||
const changed = (path: string, e: FileEntry) => {
|
||||
const s = state.get(path);
|
||||
return !s || s.size !== e.size || s.mtime !== e.mtimeMs;
|
||||
};
|
||||
const updates: { path: string; size: number; mtime: number; offset: number; hash: string | null }[] = [];
|
||||
const events: AtmEvent[] = [];
|
||||
const w = opts.worldDir;
|
||||
// Players whose advancements come from world/advancements: their log advancements would be counted twice.
|
||||
const withAdvancementFile = new Set<string>();
|
||||
|
||||
// Vanilla stats and advancements: one JSON per player uuid.
|
||||
for (const [dir, convert] of [
|
||||
[`${w}/stats`, (uuid: string, text: string, e: FileEntry) => [statsFileToEvent(uuid, text, e.mtimeMs)]],
|
||||
[`${w}/advancements`, (uuid: string, text: string) => advancementsFileToEvents(uuid, text)],
|
||||
] as const) {
|
||||
for (const e of await src.list(dir)) {
|
||||
const m = UUID_FILE.exec(e.name);
|
||||
const path = `${dir}/${e.name}`;
|
||||
if (m && dir.endsWith('/advancements')) withAdvancementFile.add(m[1]!.toLowerCase());
|
||||
if (!m || e.isDir || !changed(path, e)) continue;
|
||||
const text = (await src.read(path)).toString('utf8');
|
||||
try {
|
||||
events.push(...convert(m[1]!.toLowerCase(), text, e));
|
||||
} catch (err) {
|
||||
await pool.query('INSERT INTO app_errors(source, message) VALUES ($1, $2)', ['sftp', `${path}: ${(err as Error).message}`]);
|
||||
}
|
||||
updates.push({ path, size: e.size, mtime: e.mtimeMs, offset: 0, hash: null });
|
||||
}
|
||||
}
|
||||
|
||||
// FTB files are forwarded verbatim as file_sync, exactly like the mod does.
|
||||
// Quest config: only chapters and the English lang files (other languages are ~35 MB and unused).
|
||||
const ftbFiles = [
|
||||
...(await listRecursive(src, `${w}/ftbquests`)).filter((f) => f.name.endsWith('.snbt')),
|
||||
...(await listRecursive(src, `${w}/ftbteams`)).filter((f) => f.name.endsWith('.snbt')),
|
||||
...(await listRecursive(src, 'config/ftbquests/quests/chapters')),
|
||||
...(await listRecursive(src, 'config/ftbquests/quests/lang/en_us')),
|
||||
].filter((f) => !f.path.startsWith('config/') || isQuestConfigFile(f.path));
|
||||
for (const f of ftbFiles) {
|
||||
if (!changed(f.path, f) || f.size > MAX_FILE_BYTES) continue;
|
||||
const content = (await src.read(f.path)).toString('utf8');
|
||||
const hash = sha256(content);
|
||||
// The pipeline expects world-relative paths to start with "world/"; keep that contract for custom world dirs.
|
||||
const path = f.path.startsWith(`${w}/`) ? `world/${f.path.slice(w.length + 1)}` : f.path;
|
||||
events.push({ id: deterministicId(`file:${path}:${hash}`), type: 'file_sync', ts: f.mtimeMs, path, sha256: hash, content });
|
||||
updates.push({ path: f.path, size: f.size, mtime: f.mtimeMs, offset: 0, hash });
|
||||
}
|
||||
|
||||
// latest.log, incrementally; a different head means the file was rotated.
|
||||
const log = (await src.list('logs')).find((e) => e.name === 'latest.log');
|
||||
if (log) {
|
||||
const prev = state.get(LOG_PATH);
|
||||
const head = sha256(await src.read(LOG_PATH, 0, LOG_HEAD_BYTES));
|
||||
let offset = prev && prev.hash === head && log.size >= prev.offset ? prev.offset : 0;
|
||||
if (log.size > offset) {
|
||||
const chunk = await src.read(LOG_PATH, offset);
|
||||
const names = new Map(
|
||||
(await pool.query<{ name: string; uuid: string }>('SELECT name, uuid FROM players WHERE name IS NOT NULL')).rows.map((r) => [
|
||||
r.name,
|
||||
r.uuid,
|
||||
]),
|
||||
);
|
||||
const parsed = parseLogChunk(chunk, offset, names, opts.timeZone);
|
||||
events.push(...(await filterLogEvents(pool, parsed.events, withAdvancementFile)));
|
||||
offset += parsed.consumed;
|
||||
}
|
||||
if (!prev || prev.offset !== offset || prev.hash !== head || prev.size !== log.size) {
|
||||
updates.push({ path: LOG_PATH, size: log.size, mtime: log.mtimeMs, offset, hash: head });
|
||||
}
|
||||
}
|
||||
|
||||
for (let i = 0; i < events.length; i += BATCH) {
|
||||
await processEvents(pool, events.slice(i, i + BATCH), 'sftp', opts.serverId);
|
||||
}
|
||||
for (const u of updates) {
|
||||
await pool.query(
|
||||
`INSERT INTO sync_state(path, size, mtime, "offset", hash, synced_at) VALUES ($1, $2, $3, $4, $5, now())
|
||||
ON CONFLICT (path) DO UPDATE SET size = $2, mtime = $3, "offset" = $4, hash = $5, synced_at = now()`,
|
||||
[u.path, u.size, u.mtime, u.offset, u.hash],
|
||||
);
|
||||
}
|
||||
// Marks the time of the last successful run for the status pages, even when no file changed.
|
||||
await pool.query(`INSERT INTO sync_state(path, size, mtime) VALUES ('__last_run__', 0, 0)
|
||||
ON CONFLICT (path) DO UPDATE SET synced_at = now()`);
|
||||
return { files: updates.length, events: events.length };
|
||||
}
|
||||
|
||||
/**
|
||||
* Log-derived events are a fallback: sessions only while the mod is not pushing (otherwise they would be
|
||||
* counted twice), advancements only for players without a synced advancements file (log titles cannot be
|
||||
* mapped to advancement ids) and only if the player has none with the same title yet.
|
||||
*/
|
||||
async function filterLogEvents(pool: Pool, events: AtmEvent[], withAdvancementFile: Set<string>): Promise<AtmEvent[]> {
|
||||
const { rows } = await pool.query<{ fresh: boolean }>(
|
||||
`SELECT COALESCE(max(received_at) > now() - make_interval(secs => $1), false) AS fresh FROM events WHERE source = 'push'`,
|
||||
[PUSH_FRESH_MS / 1000],
|
||||
);
|
||||
const pushFresh = rows[0]!.fresh;
|
||||
const titled = await pool.query<{ uuid: string; title: string }>('SELECT uuid, title FROM advancements WHERE title IS NOT NULL');
|
||||
const known = new Set(titled.rows.map((r) => `${r.uuid}|${r.title}`));
|
||||
return events.filter((e) => {
|
||||
if (e.type === 'player_join' || e.type === 'player_leave') return !pushFresh;
|
||||
if (e.type === 'advancement') return !withAdvancementFile.has(e.uuid) && !known.has(`${e.uuid}|${e.title}`);
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* If the mod reported health before but has been silent for 10 minutes, the server (or the mod) is down:
|
||||
* open sessions that began while the mod was still reporting end at the last health window. Sessions started
|
||||
* later come from the log fallback and stay open. Does nothing when push was never used.
|
||||
*/
|
||||
export async function closeStaleSessions(pool: Pool, now: Date): Promise<number> {
|
||||
const { rowCount } = await pool.query(
|
||||
`WITH last AS (
|
||||
SELECT to_timestamp((payload->>'windowEnd')::bigint / 1000.0) AS at FROM events
|
||||
WHERE type = 'health' ORDER BY ts DESC LIMIT 1)
|
||||
UPDATE sessions s SET ended_at = GREATEST(s.started_at, last.at), end_reason = 'stale'
|
||||
FROM last
|
||||
WHERE s.ended_at IS NULL AND s.started_at <= last.at AND last.at < $1::timestamptz - interval '10 minutes'`,
|
||||
[now],
|
||||
);
|
||||
return rowCount ?? 0;
|
||||
}
|
||||
Reference in new issue
Block a user