Fix keyed telemetry match attribution (Refs #144)

src/lib/server/fleet/leaderboard-ingest.ts

@@ -7,10 +7,17 @@ interface Match {
     id: number;
     map_key: string;
     game: string;
-    status: 'active' | 'completed' | 'interrupted';
+    status: 'active' | 'completed' | 'interrupted' | 'end-report-missing';
     highest_round: number;
     last_event: GameTelemetry['event'];
     last_fingerprint: string;
+    start_observed: number;
+    started_at: string;
+    server_deleted_at: string | null;
+}
+
+interface KeyedMatch extends Match {
+    last_seq: number;
 }
 
 interface Participation {
@@ -51,7 +58,7 @@ function mapIdentity(handle: Database.Database, report: GameTelemetry) {
 
 function fingerprint(report: GameTelemetry): string {
     // Receipt timestamps and roster ordering do not identify new gameplay.
-    const { receivedAt: _receivedAt, timing: _timing, ...event } = report;
+    const { receivedAt: _receivedAt, timing: _timing, match: _match, seq: _seq, ...event } = report;
     const players = [...event.players].sort((a, b) => a.guid.localeCompare(b.guid));
     return createHash('sha256')
         .update(JSON.stringify({ ...event, players }))
@@ -59,7 +66,16 @@ function fingerprint(report: GameTelemetry): string {
 }
 
 /** Called inside the same transaction as the latest scoreboard snapshot. */
-export function recordLeaderboardTelemetry(handle: Database.Database, report: GameTelemetry): void {
+export function recordLeaderboardTelemetry(
+    handle: Database.Database,
+    report: GameTelemetry
+): { matchId: number; current: boolean } {
+    if (report.match !== undefined) return recordKeyedTelemetry(handle, report);
+    return { matchId: recordLegacyTelemetry(handle, report), current: true };
+}
+
+// Also used by the version-0 snapshot backfill, before keyed schema exists.
+function recordLegacyTelemetry(handle: Database.Database, report: GameTelemetry): number {
     const identity = mapIdentity(handle, report);
     const signature = fingerprint(report);
     let match = handle
@@ -74,7 +90,7 @@ export function recordLeaderboardTelemetry(handle: Database.Database, report: Ga
             handle
                 .prepare('UPDATE leaderboard_matches SET last_received_at = ? WHERE id = ?')
                 .run(report.receivedAt, match.id);
-        if (match.status !== 'interrupted') return;
+        if (match.status !== 'interrupted') return match.id;
     }
 
     const isStart = report.event === 'game_start';
@@ -95,40 +111,55 @@ export function recordLeaderboardTelemetry(handle: Database.Database, report: Ga
                 )
                 .run(report.receivedAt, match.id);
         }
-        const inserted = handle
-            .prepare(
-                `INSERT INTO leaderboard_matches
-            (server, game, map, map_key, map_name, started_at, last_received_at, start_observed, last_event, last_fingerprint)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
-            )
-            .run(
-                report.server,
-                identity.game,
-                identity.map,
-                identity.key,
-                identity.name,
-                report.receivedAt,
-                report.receivedAt,
-                Number(isStart),
-                report.event,
-                signature
-            );
-        match = {
-            id: Number(inserted.lastInsertRowid),
-            game: identity.game,
-            map_key: identity.key,
-            status: 'active',
-            highest_round: 0,
-            last_event: report.event,
-            last_fingerprint: signature
-        };
+        match = createMatch(handle, report, identity, signature);
     }
-    if (!match) return;
+    // No prior match always creates one above.
+    if (!match) throw new Error('Could not create telemetry match');
+    applyReport(handle, report, match, signature);
+    return match.id;
+}
 
+function createMatch(
+    handle: Database.Database,
+    report: GameTelemetry,
+    identity: ReturnType<typeof mapIdentity>,
+    signature: string
+): Match {
+    const inserted = handle
+        .prepare(
+            `INSERT INTO leaderboard_matches
+        (server, game, map, map_key, map_name, started_at, last_received_at, start_observed, last_event, last_fingerprint)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
+        )
+        .run(
+            report.server,
+            identity.game,
+            identity.map,
+            identity.key,
+            identity.name,
+            report.receivedAt,
+            report.receivedAt,
+            Number(report.event === 'game_start'),
+            report.event,
+            signature
+        );
+    return handle
+        .prepare('SELECT * FROM leaderboard_matches WHERE id = ?')
+        .get(inserted.lastInsertRowid) as Match;
+}
+
+function applyReport(
+    handle: Database.Database,
+    report: GameTelemetry,
+    match: Match,
+    signature: string
+): void {
+    const isStart = report.event === 'game_start';
+    const round = isStart ? 0 : report.round;
     if (isStart) {
         handle
             .prepare(
-                'UPDATE leaderboard_matches SET start_players = COALESCE(start_players, ?) WHERE id = ?'
+                'UPDATE leaderboard_matches SET start_observed = 1, start_players = COALESCE(start_players, ?) WHERE id = ?'
             )
             .run(report.players.length, match.id);
     }
@@ -159,8 +190,10 @@ export function recordLeaderboardTelemetry(handle: Database.Database, report: Ga
             rounds_survived = rounds_survived + excluded.rounds_survived,
             kills = MAX(kills, excluded.kills), headshots = MAX(headshots, excluded.headshots),
             downs = MAX(downs, excluded.downs), revives = MAX(revives, excluded.revives),
-            best_score = MAX(best_score, excluded.best_score), last_round = excluded.last_round,
+            best_score = MAX(best_score, excluded.best_score), last_round = MAX(last_round, excluded.last_round),
             present = 1, has_stats = MAX(has_stats, excluded.has_stats),
+            first_observed_round = CASE WHEN excluded.present_at_start = 1 THEN 0 ELSE first_observed_round END,
+            present_at_start = CASE WHEN excluded.present_at_start = 1 THEN 1 ELSE present_at_start END,
             completed = MAX(completed, excluded.completed), last_seen_at = excluded.last_seen_at`);
 
     // Highest round includes late joins; total rounds only credits progression
@@ -172,7 +205,7 @@ export function recordLeaderboardTelemetry(handle: Database.Database, report: Ga
         const survived =
             !isStart && prior?.present ? Math.max(0, completedRounds - prior.last_round) : 0;
         saveIdentity.run(
-            identity.game,
+            match.game,
             player.guid,
             player.name,
             report.receivedAt,
@@ -217,6 +250,153 @@ export function recordLeaderboardTelemetry(handle: Database.Database, report: Ga
         );
 }
 
+function recordKeyedTelemetry(
+    handle: Database.Database,
+    report: GameTelemetry
+): { matchId: number; current: boolean } {
+    const signature = fingerprint(report);
+    let match = handle
+        .prepare('SELECT * FROM leaderboard_matches WHERE server = ? AND reporter_match_key = ?')
+        .get(report.server, report.match) as KeyedMatch | undefined;
+    if (!match) {
+        // A key is a durable identity. Only a previously unseen key supersedes
+        // the current game; a retry for a known old key cannot close a new one.
+        handle
+            .prepare(
+                `UPDATE leaderboard_matches SET
+            status = CASE WHEN reporter_match_key IS NULL THEN 'interrupted' ELSE 'end-report-missing' END,
+            ended_at = ? WHERE server = ? AND server_deleted_at IS NULL AND status = 'active'`
+            )
+            .run(report.receivedAt, report.server);
+        const created = createMatch(handle, report, mapIdentity(handle, report), signature);
+        handle
+            .prepare('UPDATE leaderboard_matches SET reporter_match_key = ? WHERE id = ?')
+            .run(report.match, created.id);
+        match = { ...created, last_seq: 0 };
+    }
+
+    const result = { matchId: match.id, current: false };
+    if (report.seq !== undefined) {
+        const inserted = handle
+            .prepare(
+                `INSERT INTO leaderboard_match_reports (match_id, seq, applied)
+            VALUES (?, ?, 0) ON CONFLICT DO NOTHING`
+            )
+            .run(match.id, report.seq);
+        if (!inserted.changes) return result;
+        // Count received sequences even when their state is too old to apply.
+        // This represents gaps without allocating a row for each missing value.
+        handle
+            .prepare(
+                `UPDATE leaderboard_matches SET last_seq = MAX(last_seq, ?),
+            report_count = report_count + 1 WHERE id = ?`
+            )
+            .run(report.seq, match.id);
+    }
+
+    const stale =
+        report.seq !== undefined
+            ? report.seq < match.last_seq
+            : report.event !== 'game_start' && report.round < match.highest_round;
+    const delayedStart =
+        report.event === 'game_start' &&
+        (stale || match.highest_round > 0 || match.status !== 'active');
+    let applied = false;
+    if (delayedStart) {
+        if (!match.start_observed) {
+            observeDelayedStart(handle, report, match);
+            applied = true;
+        }
+    } else if (stale) {
+        // Completion is forward progress, even if its sequence trails a report
+        // already seen. Do not rewind the newer roster or counters with it.
+        if (report.event === 'game_end' && match.status !== 'completed') {
+            completeDelayedMatch(handle, report, match);
+            applied = true;
+        }
+    } else if (match.status !== 'completed') {
+        applyReport(handle, report, match, signature);
+        applied = true;
+        const latest = handle
+            .prepare(
+                `SELECT id FROM leaderboard_matches
+            WHERE server = ? AND server_deleted_at IS NULL ORDER BY id DESC LIMIT 1`
+            )
+            .get(report.server) as { id: number } | undefined;
+        result.current = latest?.id === match.id && match.server_deleted_at === null;
+    }
+    if (applied && report.seq !== undefined) {
+        handle
+            .prepare(
+                'UPDATE leaderboard_match_reports SET applied = 1 WHERE match_id = ? AND seq = ?'
+            )
+            .run(match.id, report.seq);
+    }
+    return result;
+}
+
+function completeDelayedMatch(
+    handle: Database.Database,
+    report: Extract<GameTelemetry, { round: number }>,
+    match: Match
+): void {
+    handle
+        .prepare(
+            `UPDATE leaderboard_matches SET status = 'completed', ended_at = ?,
+        highest_round = MAX(highest_round, ?) WHERE id = ?`
+        )
+        .run(report.receivedAt, report.round, match.id);
+    const identity =
+        handle.prepare(`INSERT INTO leaderboard_players (game, guid, name, first_seen_at, last_seen_at)
+        VALUES (?, ?, ?, ?, ?) ON CONFLICT(game, guid) DO NOTHING`);
+    const endPlayer = handle.prepare(`INSERT INTO leaderboard_match_players
+        (match_id, guid, highest_round, kills, headshots, downs, revives, best_score,
+            last_round, present, has_stats, completed, first_seen_at, last_seen_at, first_observed_round, present_at_start)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 1, 1, ?, ?, ?, 0)
+        ON CONFLICT(match_id, guid) DO UPDATE SET
+            highest_round = MAX(highest_round, excluded.highest_round),
+            kills = MAX(kills, excluded.kills), headshots = MAX(headshots, excluded.headshots),
+            downs = MAX(downs, excluded.downs), revives = MAX(revives, excluded.revives),
+            best_score = MAX(best_score, excluded.best_score), has_stats = 1, completed = 1`);
+    for (const player of report.players) {
+        identity.run(match.game, player.guid, player.name, report.receivedAt, report.receivedAt);
+        endPlayer.run(
+            match.id,
+            player.guid,
+            report.round,
+            player.kills,
+            player.headshots,
+            player.downs,
+            player.revives,
+            player.score,
+            Math.max(0, report.round - 1),
+            report.receivedAt,
+            report.receivedAt,
+            report.round
+        );
+    }
+}
+
+/** Recover start evidence without replacing the latest roster or its progress. */
+function observeDelayedStart(handle: Database.Database, report: GameTelemetry, match: Match): void {
+    handle
+        .prepare(
+            `UPDATE leaderboard_matches SET start_observed = 1, start_players = ? WHERE id = ?`
+        )
+        .run(report.players.length, match.id);
+    const identity =
+        handle.prepare(`INSERT INTO leaderboard_players (game, guid, name, first_seen_at, last_seen_at)
+        VALUES (?, ?, ?, ?, ?) ON CONFLICT(game, guid) DO NOTHING`);
+    const playerStart = handle.prepare(`INSERT INTO leaderboard_match_players
+        (match_id, guid, present, first_seen_at, last_seen_at, first_observed_round, present_at_start)
+        VALUES (?, ?, 0, ?, ?, 0, 1) ON CONFLICT(match_id, guid) DO UPDATE SET
+        present_at_start = 1, first_observed_round = 0`);
+    for (const player of report.players) {
+        identity.run(match.game, player.guid, player.name, match.started_at, match.started_at);
+        playerStart.run(match.id, player.guid, match.started_at, match.started_at);
+    }
+}
+
 /** Seed the single surviving scoreboard per server once; older games are gone. */
 export function backfillLeaderboard(handle: Database.Database): void {
     if (handle.prepare("SELECT 1 FROM settings WHERE key = 'leaderboard_v1'").get()) return;
@@ -239,7 +419,7 @@ export function backfillLeaderboard(handle: Database.Database): void {
             } catch {
                 continue;
             } // Legacy malformed snapshots cannot establish player records.
-            recordLeaderboardTelemetry(handle, report);
+            recordLegacyTelemetry(handle, report);
         }
         handle.prepare("INSERT INTO settings (key, value) VALUES ('leaderboard_v1', '1')").run();
     })();