minecraft-atm10-stats/backend/test/projections.test.ts

import type { Pool } from 'pg';
import { afterAll, beforeEach, describe, expect, it } from 'vitest';
import '../src/pipeline/projections/index.ts';
import { processEvents } from '../src/pipeline/process.ts';
import { closeTestPool, getTestPool, resetDb } from './helpers/db.ts';
import { ALICE, BOB, ev, MIN, T0 } from './helpers/events.ts';

let pool: Pool;
beforeEach(async () => {
  pool = await getTestPool();
  await resetDb(pool);
});
afterAll(closeTestPool);

const push = (...events: Parameters<typeof processEvents>[1]) => processEvents(pool, events, 'push', 'atm10');
const ms = (d: Date) => d.getTime();

describe('sessions', () => {
  it('join then leave creates closed session with duration', async () => {
    await push(ev('player_join', { ...ALICE, ts: T0 }), ev('player_leave', { ...ALICE, ts: T0 + 30 * MIN }));
    const { rows } = await pool.query('SELECT started_at, ended_at, end_reason FROM sessions');
    expect(rows).toHaveLength(1);
    expect(ms(rows[0].ended_at) - ms(rows[0].started_at)).toBe(30 * MIN);
    expect(rows[0].end_reason).toBe('leave');
  });

  it('closes dangling sessions on server_start', async () => {
    await push(
      ev('player_join', { ...ALICE, ts: T0 }),
      ev('health', {
        ts: T0 + 10 * MIN,
        windowStart: T0 + 9 * MIN,
        windowEnd: T0 + 10 * MIN,
        msptAvg: 30,
        msptP95: 40,
        msptMax: 50,
        tps: 20,
        players: 1,
        chunks: 100,
        heapUsedMb: 1000,
        heapMaxMb: 4000,
      }),
    );
    await push(
      ev('server_start', { ts: T0 + 60 * MIN, mcVersion: '1.21.1', neoforgeVersion: '21.1.219', javaVersion: '21', mods: [] }),
    );
    const { rows } = await pool.query('SELECT ended_at, end_reason FROM sessions');
    expect(ms(rows[0].ended_at)).toBe(T0 + 10 * MIN);
    expect(rows[0].end_reason).toBe('crash');
    const info = await pool.query('SELECT mc_version FROM server_info');
    expect(info.rows[0].mc_version).toBe('1.21.1');
  });

  it('server_stop closes open sessions', async () => {
    await push(ev('player_join', { ...ALICE, ts: T0 }), ev('server_stop', { ts: T0 + 5 * MIN }));
    const { rows } = await pool.query('SELECT ended_at, end_reason FROM sessions');
    expect(ms(rows[0].ended_at)).toBe(T0 + 5 * MIN);
    expect(rows[0].end_reason).toBe('stop');
  });

  it('second join closes the previous open session as superseded', async () => {
    await push(ev('player_join', { ...ALICE, ts: T0 }), ev('player_join', { ...ALICE, ts: T0 + 5 * MIN }));
    const { rows } = await pool.query('SELECT end_reason FROM sessions ORDER BY started_at');
    expect(rows.map((r) => r.end_reason)).toEqual(['superseded', null]);
  });
});

describe('players', () => {
  it('rename updates name and history', async () => {
    await push(ev('player_join', { uuid: ALICE.uuid, name: 'OldName', ts: T0 }));
    await push(ev('player_join', { uuid: ALICE.uuid, name: 'NewName', ts: T0 + MIN }));
    const p = await pool.query('SELECT name FROM players WHERE uuid = $1', [ALICE.uuid]);
    expect(p.rows[0].name).toBe('NewName');
    const h = await pool.query('SELECT name FROM name_history WHERE uuid = $1 ORDER BY seen_at', [ALICE.uuid]);
    expect(h.rows.map((r) => r.name)).toEqual(['OldName', 'NewName']);
  });
});

describe('stats', () => {
  const snap = (ts: number, value: number, source: 'push' | 'sftp' = 'push') =>
    processEvents(
      pool,
      [ev('stats_snapshot', { uuid: ALICE.uuid, name: ALICE.name, full: false, ts, stats: { 'minecraft:custom': { 'minecraft:deaths': value } } })],
      source,
      'atm10',
    );
  const deaths = async () =>
    (await pool.query(`SELECT value FROM stat_current WHERE key = 'minecraft:deaths'`)).rows[0]?.value;

  it('newer observation wins even if smaller', async () => {
    await snap(T0 + 2 * MIN, 5);
    await snap(T0 + 1 * MIN, 9, 'sftp');
    expect(await deaths()).toBe(5);
    await snap(T0 + 3 * MIN, 3);
    expect(await deaths()).toBe(3);
  });

  it('full snapshot does not zero missing keys', async () => {
    await processEvents(
      pool,
      [ev('stats_snapshot', { uuid: ALICE.uuid, name: null, full: false, ts: T0, stats: { 'minecraft:custom': { 'minecraft:jump': 7 } } })],
      'push',
      'atm10',
    );
    await processEvents(
      pool,
      [ev('stats_snapshot', { uuid: ALICE.uuid, name: null, full: true, ts: T0 + MIN, stats: { 'minecraft:custom': { 'minecraft:deaths': 1 } } })],
      'sftp',
      'atm10',
    );
    const { rows } = await pool.query(`SELECT key, value FROM stat_current ORDER BY key`);
    expect(rows).toEqual([
      { key: 'minecraft:deaths', value: 1 },
      { key: 'minecraft:jump', value: 7 },
    ]);
  });

  it('creates unknown player without name and fills the name later', async () => {
    await processEvents(
      pool,
      [ev('stats_snapshot', { uuid: BOB.uuid, name: null, full: true, ts: T0, stats: { 'minecraft:custom': { 'minecraft:jump': 1 } } })],
      'sftp',
      'atm10',
    );
    expect((await pool.query('SELECT name FROM players')).rows[0].name).toBeNull();
    await push(ev('player_join', { ...BOB, ts: T0 + MIN }));
    expect((await pool.query('SELECT name FROM players')).rows[0].name).toBe('Bob');
  });
});

describe('advancements, combat, dimensions, health', () => {
  it('advancement keeps earliest timestamp', async () => {
    await push(ev('advancement', { ...ALICE, ts: T0 + MIN, advancementId: 'a:b', title: null }));
    await push(ev('advancement', { ...ALICE, ts: T0, advancementId: 'a:b', title: 'Title' }));
    const { rows } = await pool.query('SELECT achieved_at, title FROM advancements');
    expect(ms(rows[0].achieved_at)).toBe(T0);
    expect(rows[0].title).toBe('Title');
  });

  it('death and boss kill are stored', async () => {
    await push(
      ev('player_death', { ...ALICE, ts: T0, dimension: 'minecraft:the_nether', cause: 'minecraft:lava', killerType: null, killerName: null }),
      ev('entity_kill', { ...ALICE, ts: T0, dimension: 'minecraft:the_end', victimType: 'minecraft:ender_dragon', isBoss: true, weapon: null }),
    );
    expect((await pool.query('SELECT cause FROM deaths')).rows[0].cause).toBe('minecraft:lava');
    expect((await pool.query('SELECT is_boss FROM kills')).rows[0].is_boss).toBe(true);
  });

  it('dimension visits keep first visit', async () => {
    await push(ev('dimension_change', { ...ALICE, ts: T0, from: 'minecraft:overworld', to: 'minecraft:the_nether' }));
    await push(ev('dimension_change', { ...ALICE, ts: T0 + MIN, from: 'minecraft:overworld', to: 'minecraft:the_nether' }));
    const { rows } = await pool.query('SELECT dimension, first_at FROM dimension_visits ORDER BY dimension');
    expect(rows.map((r) => r.dimension)).toEqual(['minecraft:overworld', 'minecraft:the_nether']);
    expect(ms(rows[1].first_at)).toBe(T0);
  });

  it('health stored per minute bucket', async () => {
    const h = { msptAvg: 30, msptP95: 40, msptMax: 50, tps: 20, players: 1, chunks: 100, heapUsedMb: 1000, heapMaxMb: 4000 };
    await push(ev('health', { ...h, ts: T0 + MIN, windowStart: T0 + 5_000, windowEnd: T0 + MIN }));
    const { rows } = await pool.query('SELECT bucket, tps FROM server_health');
    expect(ms(rows[0].bucket)).toBe(T0);
    expect(rows[0].tps).toBe(20);
  });
});

describe('review fixes', () => {
  it('a late event with an older name does not revert a rename', async () => {
    await push(ev('player_join', { uuid: ALICE.uuid, name: 'NewName', ts: T0 + 10 * MIN }));
    await push(ev('player_leave', { uuid: ALICE.uuid, name: 'OldName', ts: T0 }));
    expect((await pool.query('SELECT name FROM players WHERE uuid = $1', [ALICE.uuid])).rows[0].name).toBe('NewName');
  });
});