adding more logging for reconcile

svc/fullhistory.ts

@@ -142,8 +142,10 @@ async function processFullHistory(job: FullHistoryJob) {
       const row = { account_id: player.account_id, match_id: match.match_id, player_slot: playerSlot };
       // Note: If an account ID shows up here that player is not anonymous anymore (could update fh_unavailable for other players)
       // Log the match IDs we should reconcile, we can query our own DB for data
-      await db.raw('INSERT INTO player_match_history(account_id, match_id, player_slot) VALUES (?, ?, ?) ON CONFLICT DO NOTHING', [row.account_id, row.match_id, row.player_slot]);
-      await redisCount('pmh_fullhistory');
+      const { rows } = await db.raw('INSERT INTO player_match_history(account_id, match_id, player_slot) VALUES (?, ?, ?) ON CONFLICT DO NOTHING RETURNING *', [row.account_id, row.match_id, row.player_slot]);
+      if (rows?.length) {
+        await redisCount('pmh_fullhistory');
+      }
     }
   }
   if (isMatchDataDisabled != null) {

svc/parser.ts

@@ -125,13 +125,13 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
     }
 
     // Reconcile anonymous players now that we have parsed data
-    // await Promise.all(gcMatch.players
-    //   .filter(p => !Boolean(pgroup[p.player_slot]?.account_id))
-    //   .map(async (p) => {
-    //     await db.raw('INSERT INTO player_match_history(account_id, match_id, player_slot) VALUES (?, ?, ?) ON CONFLICT DO NOTHING', [p.account_id, gcMatch.match_id, p.player_slot]);
-    //     await redisCount('pmh_parsed');
-    //   })
-    // );
+    await Promise.all(gcMatch.players
+      .filter(p => !Boolean(pgroup[p.player_slot]?.account_id))
+      .map(async (p) => {
+        // await db.raw('INSERT INTO player_match_history(account_id, match_id, player_slot) VALUES (?, ?, ?) ON CONFLICT DO NOTHING', [p.account_id, gcMatch.match_id, p.player_slot]);
+        await redisCount('pmh_parsed');
+      })
+    );
 
     // Log successful parse and timing
     log('success');

svc/reconcile.ts

@@ -5,9 +5,49 @@
  * It checks our own database and updates player_caches so these matches get associated with the player.
  */
 import db from "../store/db";
-import { reconcileMatch } from "../util/reconcileMatch";
+import { getMatchDataFromBlobWithMetadata } from "../util/buildMatch";
+import { getPGroup } from "../util/pgroup";
+import { upsertPlayerCaches } from "../util/playerCaches";
 import type { HistoryType } from "../util/types";
 
+export async function reconcileMatch(rows: HistoryType[]) {
+  // validate that all rows have the same match ID
+  const set = new Set(rows.map(r => r.match_id));
+  if (set.size > 1) {
+    throw new Error('multiple match IDs found in input to reconcileMatch');
+  }
+  // optional: Verify each player/match combination doesn't exist in player_caches (or we have parsed data to update)
+  const [match] = await getMatchDataFromBlobWithMetadata(rows[0].match_id);
+  if (!match) {
+    // Note: unless we backfill, we have limited API data for old matches
+    // For more recent matches we're more likely to have data
+    // Maybe we can mark the more recent matches with a flag
+    // Or queue up recent matches from fullhistory and process them in order so fh requests show updates quicker
+    return;
+  }
+  const pgroup = getPGroup(match);
+  // If reconciling after fullhistory, the pgroup won't contain account_id info. Add it.
+  rows.forEach(r => {
+    if (!pgroup[r.player_slot]?.account_id) {
+      pgroup[r.player_slot].account_id = r.account_id;
+    }
+  });
+  const targetSlots = new Set(rows.map(r => r.player_slot));
+  // Filter to only players that we want to fill in
+  match.players = match.players.filter(p => targetSlots.has(p.player_slot));
+  if (!match.players.length) {
+    return;
+  }
+  // Call upsertPlayerCaches: pgroup will be used to populate account_id and heroes fields (for peers search)
+  const result = await upsertPlayerCaches(match, undefined, pgroup, 'reconcile');
+  if (result.every(Boolean)) {
+    // Delete the rows since we successfully updated
+    await Promise.all(rows.map(async (row) => {
+      return db.raw('DELETE FROM player_match_history WHERE account_id = ? AND match_id = ?', [row.account_id, row.match_id]);
+    }));
+  }
+}
+
 async function doReconcile() {
   while (true) {
     // Fetch rows for a single match (could be multiple players to fill)
@@ -17,12 +57,13 @@ async function doReconcile() {
       // We still might have data, so process it with some probability
       // If not processed, retry with short interval
       if (Math.random() < 0.9) {
-        await new Promise(resolve => setTimeout(resolve, 100));
+        console.log('skip', rows[0].match_id);
+        await new Promise(resolve => setTimeout(resolve, 50));
         continue;
       }
     }
     await reconcileMatch(rows);
-    await new Promise(resolve => setTimeout(resolve, 500));
+    await new Promise(resolve => setTimeout(resolve, 250));
   }
 }
 doReconcile();