add log endpoint to replace telemetry arrays

store/buildStatus.mjs

@@ -150,22 +150,6 @@ export default function buildStatus(db, redis, cb) {
           }
         );
       },
-      last_added(cb) {
-        redis.lrange('matches_last_added', 0, -1, (err, result) => {
-          cb(
-            err,
-            result.map((r) => JSON.parse(r))
-          );
-        });
-      },
-      last_parsed(cb) {
-        redis.lrange('matches_last_parsed', 0, -1, (err, result) => {
-          cb(
-            err,
-            result.map((r) => JSON.parse(r))
-          );
-        });
-      },
       load_times(cb) {
         redis.lrange('load_times', 0, -1, (err, arr) => {
           cb(err, generatePercentiles(arr));

store/queries.mjs

@@ -1,4 +1,5 @@
 import async from 'async';
+import moment from 'moment';
 import constants from 'dotaconstants';
 import util from 'util';
 import utility from '../util/utility.mjs';
@@ -1271,22 +1272,14 @@ function insertMatch(match, options, cb) {
     return insertPlayerCache(copy, cb);
   }
   function telemetry(cb) {
-    const types = {
-      api: 'matches_last_added',
-      parsed: 'matches_last_parsed',
-    };
-    if (types[options.type]) {
-      redis.lpush(
-        types[options.type],
-        JSON.stringify({
-          match_id: match.match_id,
-          // start_time + duration = end_time
-          start_time: match.start_time,
-          duration: match.duration,
-        })
-      );
-      redis.ltrim(types[options.type], 0, 9);
-    }
+    // Publish to log stream
+    // Name of process
+    // type of data (parsed gcdata api)
+    // Match ID
+    // When it finished (start_time + duration)
+    const name = process.env.name || process.argv[1];
+    const message = `[${name}] inserted [${options.type}] for match ${match.match_id} finished ${moment.unix(match.start_time + match.duration).fromNow()}`;
+    redis.publish(options.type, message);
     if (options.type === 'parsed') {
       redisCount(redis, 'parser');
     }

svc/web.mjs

@@ -12,13 +12,15 @@ import passportSteam from 'passport-steam';
 import cors from 'cors';
 import bodyParser from 'body-parser';
 import stripeLib from 'stripe';
+import Redis from 'ioredis';
 import keys from '../routes/keyManagement.mjs';
 import api from '../routes/api.mjs';
 import queries from '../store/queries.mjs';
 import db from '../store/db.mjs';
 import redis from '../store/redis.mjs';
 import utility from '../util/utility.mjs';
 import config from '../config.js';
+
 const SteamStrategy = passportSteam.Strategy;
 const stripe = stripeLib(config.STRIPE_SECRET);
 const { redisCount } = utility;
@@ -335,6 +337,29 @@ app.route('/manageSub').post(async (req, res) => {
   return res.json(session);
 });
 app.use('/api', api);
+let openClients = 0;
+app.get('/logs', (req, res, cb) => {
+  // limit the number of max clients
+  if (openClients >= 100) {
+    return cb('too many log clients');
+  }
+  const logSub = new Redis(config.REDIS_URL);
+  logSub.subscribe(['api', 'parsed', 'gcdata']);
+  openClients += 1;
+  // Create a subscriber client
+  logSub.on('message', (channel, message) => {
+    const matched = Array.isArray(req.query.channel) && req.query.channel.includes(channel);
+    if (!req.query.channel || matched) {
+      res.write(message + '\n');
+    }
+  });
+  // Teardown the subscriber on request close
+  req.on('close', () => {
+    console.log('closing log client');
+    logSub.disconnect();
+    openClients -= 1;
+  });
+});
 // CORS Preflight for API keys
 // NB: make sure UI_HOST is set e.g. http://localhost:3000 otherwise CSRF check above will stop preflight from working
 app.options('/keys', cors());