archive using pg ids

Author: howardchung

dev/streamArchive.ts

@@ -2,40 +2,46 @@ import cassandra from '../store/cassandra';
 import db from '../store/db';
 import { doArchiveFromLegacy } from '../store/queries';
 
+const stream = db.raw('SELECT match_id from parsed_matches WHERE is_archived IS NULL').stream();
+stream.on('data', async (row) => {
+  await doArchiveFromLegacy(row.match_id.toString());
+  // console.log(row);
+});
+
 let i = 0;
-cassandra
-  .stream('select match_id, version from matches', [], {
-    prepare: true,
-    autoPage: true,
-  })
-  .on('readable', async function () {
-    // readable is emitted as soon a row is received and parsed
-    let row;
-    //@ts-ignore
-    while ((row = this.read())) {
-      // process row
-      i += 1;
-      console.log(i, row.match_id);
-      if (row.version) {
-        await db.raw(
-          'INSERT INTO parsed_matches(match_id) VALUES(?) ON CONFLICT DO NOTHING',
-          [row.match_id]
-        );
-        await doArchiveFromLegacy(row.match_id.toString());
-      } else {
-        await deleteFromStore(row.match_id.toString());
-      }
-    }
-  })
-  .on('end', function () {
-    // emitted when all rows have been retrieved and read
-    console.log('finished');
-    process.exit(0);
-  })
-  .on('error', function (e) {
-    console.error(e);
-    process.exit(1);
-  });
+// const stream = cassandra
+//   .stream('select match_id, version from matches', [], {
+//     prepare: true,
+//     autoPage: true,
+//   })
+//   .on('readable', async function () {
+//     // readable is emitted as soon a row is received and parsed
+//     let row;
+//     //@ts-ignore
+//     while ((row = this.read())) {
+//       // process row
+//       i += 1;
+//       console.log(i, row.match_id);
+//       if (row.version) {
+//         await db.raw(
+//           'INSERT INTO parsed_matches(match_id) VALUES(?) ON CONFLICT DO NOTHING',
+//           [row.match_id]
+//         );
+//         await doArchiveFromLegacy(row.match_id.toString());
+//       } else {
+//         await deleteFromStore(row.match_id.toString());
+//       }
+//     }
+//   });
+stream.on('end', function () {
+  // emitted when all rows have been retrieved and read
+  console.log('finished');
+  // process.exit(0);
+})
+stream.on('error', function (e) {
+  console.error(e);
+  // process.exit(1);
+});
 
 async function deleteFromStore(id: string) {
   await Promise.all([

dev/teamElo.ts

@@ -1,7 +1,6 @@
 /**
  * Computes team Elo ratings by game
  * */
-import JSONStream from 'JSONStream';
 import db from '../store/db';
 
 // Keep each team's rating in memory and update
@@ -23,7 +22,6 @@ ORDER BY match_id ASC
 `
   )
   .stream();
-stream.pipe(JSONStream.parse());
 stream.on('data', (match) => {
   // console.log(JSON.stringify(match));
   if (!teams[match.team_id1]) {

dev/wordcount.ts

@@ -1,4 +1,3 @@
-import JSONStream from 'JSONStream';
 import db from '../store/db';
 import {countWords} from '../util/compute';
 import { mergeObjects } from '../util/utility';
@@ -17,7 +16,6 @@ stream.on('end', () => {
   console.log(JSON.stringify(counts));
   process.exit(0);
 });
-stream.pipe(JSONStream.parse());
 stream.on('data', (match) => {
   mergeObjects(counts, countWords(match, null));
 });

global.d.ts

@@ -6,7 +6,6 @@ declare module 'steam';
 declare module 'dota2';
 declare module 'passport-steam';
 declare module 'uuid';
-declare module 'JSONStream';
 declare module 'ws';
 
 declare namespace Express {

package-lock.json

@@ -45,14 +45,14 @@
     "http-proxy": "^1.18.1",
     "ioredis": "^5.3.2",
     "json-bigint": "^0.3.0",
-    "JSONStream": "^1.3.5",
     "knex": "^3.0.1",
     "long": "^4.0.0",
     "moment": "^2.24.0",
     "nock": "^13.0.5",
     "passport": "^0.4.1",
     "passport-steam": "^1.0.18",
     "pg": "^8.11.3",
+    "pg-query-stream": "^4.5.3",
     "protobufjs": "^7.2.5",
     "request": "^2.88.2",
     "simple-vdf": "^1.1.1",