Merge branch 'Seneca-CDOT:master' into master

suhhee1011 · Dec 10, 2021 · d53d13b · d53d13b
2 parents 4733381 + 5730360
commit d53d13b
Show file tree

Hide file tree

Showing 7 changed files with 123 additions and 3 deletions.
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
diff --git a/src/api/posts/package.json b/src/api/posts/package.json
@@ -17,6 +17,7 @@
   },
   "dependencies": {
     "@senecacdot/satellite": "^1.x",
+    "bull": "^3.20.1",
     "express-validator": "^6.13.0",
     "ioredis": "^4.28.0",
     "ioredis-mock": "^5.8.0",

diff --git a/src/api/posts/src/queue.js b/src/api/posts/src/queue.js
@@ -0,0 +1,20 @@
+const Bull = require('bull');
+const redis = require('./redis');
+
+const client = redis;
+const subscriber = redis;
+
+const queue = new Bull('feed-queue', {
+  createClient: (type) => {
+    switch (type) {
+      case 'client':
+        return client;
+      case 'subscriber':
+        return subscriber;
+      default:
+        return redis;
+    }
+  },
+});
+
+module.exports = queue;
diff --git a/src/api/posts/src/routes/feeds.js b/src/api/posts/src/routes/feeds.js
@@ -1,7 +1,8 @@
 const { Router, logger, isAuthenticated, isAuthorized } = require('@senecacdot/satellite');
 const Feed = require('../data/feed');
-const { getFeeds, getInvalidFeeds } = require('../storage');
+const { getFeeds, getInvalidFeeds, getDelayedFeeds } = require('../storage');
 const { validateNewFeed, validateFeedsIdParam } = require('../validation');
+const queue = require('../queue');
 
 const feeds = Router();
 
@@ -45,6 +46,34 @@ feeds.get('/invalid', async (req, res, next) => {
   );
 });
 
+feeds.get('/delayed', async (req, res, next) => {
+  let delayedFeeds;
+  try {
+    delayedFeeds = await getDelayedFeeds();
+  } catch (error) {
+    logger.error({ error }, 'Unable to get delayed feeds from Redis');
+    return next(error);
+  }
+  res.set('X-Total-Count', delayedFeeds.length);
+  return res.json(
+    delayedFeeds.map((element) => ({
+      ...element,
+      url: `${feedURL}/${element.id}`,
+    }))
+  );
+});
+
+feeds.get('/info', async (req, res, next) => {
+  try {
+    const [jobCnt, queueInfo] = await Promise.all([queue.count(), queue.getJobCounts()]);
+    queueInfo.jobCnt = jobCnt;
+    res.json({ queueInfo });
+  } catch (error) {
+    logger.error({ error }, 'Unable to get information from feed-queue');
+    next(error);
+  }
+});
+
 feeds.get('/:id', validateFeedsIdParam(), async (req, res, next) => {
   const { id } = req.params;
 

diff --git a/src/api/posts/src/storage.js b/src/api/posts/src/storage.js
@@ -90,6 +90,16 @@ module.exports = {
     );
   },
 
+  getDelayedFeeds: async () => {
+    const delayedKeys = await getFeedKeysUsingScanStream(`${feedNamespace}*${delayedSuffix}`);
+    return delayedKeys.map((key) => {
+      const id = key.replace(feedNamespace, '').replace(delayedSuffix, '');
+      return {
+        id,
+      };
+    });
+  },
+
   getFlaggedFeeds: () => redis.smembers(flaggedFeedsKey),
 
   getFeed: (id) => redis.hgetall(feedNamespace.concat(id)),
@@ -127,7 +137,7 @@ module.exports = {
 
   isInvalid: (id) => redis.exists(createInvalidFeedKey(id)),
 
-  setDelayedFeed: (id, seconds) => redis.set(createDelayedFeedKey(id), seconds, 1),
+  setDelayedFeed: (id, seconds) => redis.set(createDelayedFeedKey(id), seconds, 'EX', seconds),
 
   isDelayed: (id) => redis.exists(createDelayedFeedKey(id)),
 

diff --git a/src/api/posts/test/feeds.test.js b/src/api/posts/test/feeds.test.js
@@ -37,3 +37,61 @@ describe('Test GET /feeds/invalid endpoint', () => {
     expect(res.body[0].id).toBe(feeds[0].id);
   });
 });
+describe('Test GET /feeds/delayed endpoint', () => {
+  const createdItems = 150;
+  // Array of feeds
+  const feeds = [...Array(createdItems).keys()].map((item) => {
+    return new Feed('foo', `http://telescope${item}.cdot.systems`);
+  });
+  beforeAll(() => Promise.all(feeds.map((feed) => feed.save())));
+
+  it('Should return 200 and empty body when there are no delayed feeds', async () => {
+    const res = await request(app).get('/feeds/delayed');
+
+    expect(res.status).toEqual(200);
+    expect(res.get('Content-type')).toContain('application/json');
+    expect(res.get('X-Total-Count')).toBe('0');
+    expect(res.body.length).toBe(0);
+    expect(res.body instanceof Array).toBe(true);
+  });
+
+  it('Should return 200 and delayed feeds object', async () => {
+    // arrange invalid feeds
+    const createDelayedFeeds = 10;
+    for (let i = 0; i < createDelayedFeeds; i += 1) {
+      feeds[i].setDelayed(3600);
+    }
+    const res = await request(app).get('/feeds/delayed');
+
+    expect(res.status).toEqual(200);
+    expect(res.get('Content-type')).toContain('application/json');
+    expect(res.get('X-Total-Count')).toBe(createDelayedFeeds.toString());
+    expect(res.body.length).toBe(createDelayedFeeds);
+    expect(res.body instanceof Array).toBe(true);
+    expect(res.body[0].id).toBe(feeds[0].id);
+  });
+});
+
+describe('GET /feeds/info', () => {
+  test('Should return 200 and valid response object', async () => {
+    function checkKeys(resBody) {
+      let bool = true;
+      const allKeys = ['waiting', 'active', 'completed', 'failed', 'delayed', 'paused', 'jobCnt'];
+      const resKeys = Object.keys(resBody.queueInfo);
+      for (let i = 0; i < resKeys.length; i += 1) {
+        if (allKeys.indexOf(resKeys[i]) < 0 || typeof resBody.queueInfo[resKeys[i]] !== 'number') {
+          bool = false;
+          break;
+        }
+      }
+      return bool;
+    }
+
+    const res = await request(app).get('/feeds/info');
+
+    expect(res.status).toEqual(200);
+    expect(typeof res.body).toEqual('object');
+    expect(typeof res.body.queueInfo).toEqual('object');
+    expect(checkKeys(res.body)).toBe(true);
+  });
+});
diff --git a/src/backend/utils/storage.js b/src/backend/utils/storage.js
@@ -93,7 +93,7 @@ module.exports = {
 
   isInvalid: (id) => redis.exists(createInvalidFeedKey(id)),
 
-  setDelayedFeed: (id, seconds) => redis.set(createDelayedFeedKey(id), seconds, 1),
+  setDelayedFeed: (id, seconds) => redis.set(createDelayedFeedKey(id), seconds, 'EX', seconds),
 
   isDelayed: (id) => redis.exists(createDelayedFeedKey(id)),