import { appendFileSync, cpSync, mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { Pool } from 'pg'; import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'; import '../src/pipeline/projections/index.ts'; import { processEvents } from '../src/pipeline/process.ts'; import { LocalFileSource } from '../src/worker/fileSource.ts'; import { closeStaleSessions, syncOnce } from '../src/worker/sync.ts'; import { closeTestPool, getTestPool, resetDb } from './helpers/db.ts'; import { ALICE, ev, MIN } from './helpers/events.ts'; const FX = join(import.meta.dirname, 'fixtures'); const OPTS = { timeZone: 'Europe/Berlin', serverId: 'atm10', worldDir: 'world' }; let pool: Pool; let root: string; function makeServerDir(): string { const dir = mkdtempSync(join(tmpdir(), 'atm-sftp-')); mkdirSync(join(dir, 'world', 'stats'), { recursive: true }); mkdirSync(join(dir, 'world', 'advancements'), { recursive: true }); mkdirSync(join(dir, 'world', 'ftbquests'), { recursive: true }); mkdirSync(join(dir, 'config', 'ftbquests', 'quests', 'chapters'), { recursive: true }); mkdirSync(join(dir, 'logs'), { recursive: true }); cpSync(join(FX, 'stats'), join(dir, 'world', 'stats'), { recursive: true }); cpSync(join(FX, 'advancements'), join(dir, 'world', 'advancements'), { recursive: true }); cpSync(join(FX, 'ftbquests', '308a693c-2bd8-4d6b-8790-670aef5be601.snbt'), join(dir, 'world', 'ftbquests', '308a693c-2bd8-4d6b-8790-670aef5be601.snbt')); cpSync(join(FX, 'ftbquests', 'config', 'chapters', 'getting_started.snbt'), join(dir, 'config', 'ftbquests', 'quests', 'chapters', 'getting_started.snbt')); cpSync(join(FX, 'latest.anon.log'), join(dir, 'logs', 'latest.log')); return dir; } const LINE = (t: string, msg: string) => `[12Oct2026 ${t}] [Server thread/INFO] [net.minecraft.server.MinecraftServer/]: ${msg}\r\n`; beforeEach(async () => { pool = await getTestPool(); await resetDb(pool); root = makeServerDir(); }); afterAll(closeTestPool); const count = async (sql: string) => (await pool.query(sql)).rows[0].n as number; describe('syncOnce', () => { it('first sync imports stats, advancements, quests, quest config and log', async () => { const r = await syncOnce(pool, new LocalFileSource(root), OPTS); expect(r.files).toBe(6); expect(await count(`SELECT count(*)::int AS n FROM stat_current WHERE key = 'minecraft:play_time'`)).toBe(2); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE advancement_id = 'minecraft:story/root'`)).toBe(1); expect(await count(`SELECT count(*)::int AS n FROM quest_progress`)).toBeGreaterThan(0); expect(await count(`SELECT count(*)::int AS n FROM quest_defs`)).toBe(2); expect(await count(`SELECT count(*)::int AS n FROM sessions`)).toBeGreaterThan(0); expect((await pool.query(`SELECT name FROM players WHERE uuid = $1`, [ALICE.uuid])).rows[0].name).toBe('CDOGamer'); expect(await count(`SELECT count(*)::int AS n FROM events WHERE source = 'sftp'`)).toBeGreaterThan(0); }); it('second sync without changes reads nothing', async () => { const src = new LocalFileSource(root); await syncOnce(pool, src, OPTS); const spy = vi.spyOn(src, 'read'); const r = await syncOnce(pool, src, OPTS); expect(r.files).toBe(0); // only the 256-byte head of latest.log is read to detect rotation expect(spy.mock.calls.every(([path, start, length]) => path === 'logs/latest.log' && start === 0 && length === 256)).toBe(true); }); it('appended log lines are imported incrementally', async () => { // log advancements only count for players without a synced advancements file rmSync(join(root, 'world', 'advancements', `${ALICE.uuid}.json`)); const src = new LocalFileSource(root); await syncOnce(pool, src, OPTS); const before = await count(`SELECT count(*)::int AS n FROM advancements WHERE advancement_id LIKE 'log:%'`); appendFileSync(join(root, 'logs', 'latest.log'), LINE('23:50:00.000', 'CDOGamer has made the advancement [Brand New Title]')); const r = await syncOnce(pool, src, OPTS); expect(r.files).toBe(1); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE advancement_id LIKE 'log:%'`)).toBe(before + 1); }); it('rotated log resets offset', async () => { rmSync(join(root, 'world', 'advancements', `${ALICE.uuid}.json`)); const src = new LocalFileSource(root); await syncOnce(pool, src, OPTS); writeFileSync( join(root, 'logs', 'latest.log'), LINE('23:58:00.000', 'UUID of player CDOGamer is 308a693c-2bd8-4d6b-8790-670aef5be601').replace('MinecraftServer', 'ServerLoginPacketListenerImpl') + LINE('23:58:01.000', 'CDOGamer has made the advancement [After Rotation]'), ); await syncOnce(pool, src, OPTS); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE title = 'After Rotation'`)).toBe(1); const { rows } = await pool.query(`SELECT "offset", size FROM sync_state WHERE path = 'logs/latest.log'`); expect(rows[0].offset).toBe(rows[0].size); }); it('log joins skipped while push is fresh', async () => { await processEvents(pool, [ev('health', { msptAvg: 1, msptP95: 1, msptMax: 1, tps: 20, players: 0, chunks: 0, heapUsedMb: 1, heapMaxMb: 1, windowStart: Date.now() - MIN, windowEnd: Date.now() })], 'push', 'atm10'); await syncOnce(pool, new LocalFileSource(root), OPTS); expect(await count(`SELECT count(*)::int AS n FROM sessions`)).toBe(0); }); it('log advancement skipped when the player already has it with that title', async () => { await processEvents(pool, [ev('advancement', { ...ALICE, advancementId: 'mysticalagriculture:infusion', title: 'Infusing Efficiency' })], 'push', 'atm10'); await syncOnce(pool, new LocalFileSource(root), OPTS); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE title = 'Infusing Efficiency'`)).toBe(1); }); it('changed stats file is re-imported with its new mtime', async () => { const src = new LocalFileSource(root); await syncOnce(pool, src, OPTS); const f = join(root, 'world', 'stats', '308a693c-2bd8-4d6b-8790-670aef5be601.json'); writeFileSync(f, JSON.stringify({ stats: { 'minecraft:custom': { 'minecraft:play_time': 2_000_000 } }, DataVersion: 3955 })); const future = new Date(Date.now() + 60_000); utimesSync(f, future, future); await syncOnce(pool, src, OPTS); const { rows } = await pool.query(`SELECT value FROM stat_current WHERE uuid = $1 AND key = 'minecraft:play_time'`, [ALICE.uuid]); expect(rows[0].value).toBe(2_000_000); }); }); describe('closeStaleSessions', () => { it('closes open sessions when the mod stopped reporting health', async () => { const now = Date.now(); await processEvents( pool, [ ev('player_join', { ...ALICE, ts: now - 60 * MIN }), ev('health', { msptAvg: 1, msptP95: 1, msptMax: 1, tps: 20, players: 1, chunks: 0, heapUsedMb: 1, heapMaxMb: 1, windowStart: now - 31 * MIN, windowEnd: now - 30 * MIN, ts: now - 30 * MIN }), ], 'push', 'atm10', ); expect(await closeStaleSessions(pool, new Date(now))).toBe(1); const { rows } = await pool.query('SELECT ended_at, end_reason FROM sessions'); expect(rows[0].end_reason).toBe('stale'); expect(rows[0].ended_at.getTime()).toBe(now - 30 * MIN); }); it('keeps sessions while health is fresh or when push was never used', async () => { const now = Date.now(); await processEvents(pool, [ev('player_join', { ...ALICE, ts: now - 60 * MIN })], 'sftp', 'atm10'); expect(await closeStaleSessions(pool, new Date(now))).toBe(0); }); }); describe('review fixes: log fallback', () => { const health = (end: number) => ev('health', { msptAvg: 1, msptP95: 1, msptMax: 1, tps: 20, players: 1, chunks: 0, heapUsedMb: 1, heapMaxMb: 1, windowStart: end - MIN, windowEnd: end, ts: end }); it('a log session that started after the mod went silent stays open', async () => { const now = Date.now(); await processEvents(pool, [health(now - 60 * MIN)], 'push', 'atm10'); await processEvents(pool, [ev('player_join', { ...ALICE, ts: now - 20 * MIN })], 'sftp', 'atm10'); expect(await closeStaleSessions(pool, new Date(now))).toBe(0); expect(await count(`SELECT count(*)::int AS n FROM sessions WHERE ended_at IS NULL`)).toBe(1); }); it('late push join/leave inside an existing log session do not create a second session', async () => { const t = Date.now() - 120 * MIN; await processEvents(pool, [ev('player_join', { ...ALICE, ts: t }), ev('player_leave', { ...ALICE, ts: t + 60 * MIN })], 'sftp', 'atm10'); await processEvents(pool, [ev('player_join', { ...ALICE, ts: t + 1000 }), ev('player_leave', { ...ALICE, ts: t + 60 * MIN + 1000 })], 'push', 'atm10'); const { rows } = await pool.query(`SELECT started_at, ended_at FROM sessions`); expect(rows).toHaveLength(1); expect(rows[0].ended_at.getTime() - rows[0].started_at.getTime()).toBeGreaterThanOrEqual(60 * MIN); }); it('pull-only: advancements from file and log are counted once', async () => { await syncOnce(pool, new LocalFileSource(root), OPTS); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE uuid = '${ALICE.uuid}' AND advancement_id LIKE 'log:%'`)).toBe(0); expect(await count(`SELECT count(*)::int AS n FROM advancements WHERE uuid = '${ALICE.uuid}'`)).toBe(2); }); }); describe('quest config and teams via SFTP', () => { it('reads chapters, English split lang files and team files; skips other languages', async () => { const lang = join(root, 'config', 'ftbquests', 'quests', 'lang'); mkdirSync(join(lang, 'en_us', 'chapters'), { recursive: true }); mkdirSync(join(lang, 'de_de', 'chapters'), { recursive: true }); cpSync(join(FX, 'ftbquests', 'config', 'lang', 'en_us'), join(lang, 'en_us'), { recursive: true }); writeFileSync(join(lang, 'de_de', 'chapters', 'getting_started.snbt_merged'), '{ quest.3BC0A50886A3222B.title: "Baum schlagen" }'); mkdirSync(join(root, 'world', 'ftbteams'), { recursive: true }); cpSync(join(FX, 'ftbteams'), join(root, 'world', 'ftbteams'), { recursive: true }); const src = new LocalFileSource(root); const spy = vi.spyOn(src, 'read'); await syncOnce(pool, src, OPTS); expect(spy.mock.calls.some(([p]) => String(p).includes('de_de'))).toBe(false); const { rows } = await pool.query(`SELECT title, chapter_title FROM quest_defs WHERE quest_id = '3BC0A50886A3222B'`); expect(rows).toEqual([{ title: 'Punch a Tree', chapter_title: 'Getting Started' }]); expect(await count(`SELECT count(*)::int AS n FROM team_members`)).toBe(2); }); });