feat(Statements): Add flexibility for bypassing queue (LLC-1560) (#849)

docker-compose.yml

@@ -2,22 +2,23 @@ version: '3.6'
 services:
   mongo:
     container_name: ll_mongo
-    image: mongo:4.0
+    image: mongo:4.4
+    command: --bind_ip_all --replSet rs0
     ports:
-    - 27017:27017
+      - 27017:27017
     volumes:
-    - mongo-config:/data/configdb:rw
-    - mongo-data:/data/db:rw
+      - mongo-config:/data/configdb:rw
+      - mongo-data:/data/db:rw
   redis:
     # activate persistency
     command: redis-server --appendonly yes
     container_name: ll_redis
     image: redis:4.0
     ports:
-    - 6379:6379
+      - 6379:6379
     volumes:
-    - redis-data:/data:rw
+      - redis-data:/data:rw
 volumes:
   mongo-config: {}
   mongo-data: {}
-  redis-data: {}
+  redis-data: {}

src/apps/statements/expressPresenter/postStatements/alternateRequest.ts

@@ -15,6 +15,7 @@ import getUrlPath from '../utils/getUrlPath';
 import storeStatement from '../utils/storeStatement';
 import validateVersionHeader from '../utils/validateHeaderVersion';
 import { validateStatementProcessingPriority } from '../utils/validateStatementProcessingPriority';
+import { validateStatementBypassQueues } from '../utils/validateStatementBypassQueues';
 import storeStatements from './storeStatements';
 
 export interface Options {
@@ -52,9 +53,13 @@ export default async ({ config, method, req, res }: Options) => {
   checkUnknownParams(req.query, ['method']);
 
   validateStatementProcessingPriority(req.query.priority as string | undefined);
+  validateStatementBypassQueues(req.query.bypassQueues as string | undefined);
   const priority =
     (req.query.priority as StatementProcessingPriority) || StatementProcessingPriority.MEDIUM;
-
+  const bypassQueues =
+    req.query.bypassQueues && (req.query.bypassQueues as string).trim() !== ''
+      ? (req.query.bypassQueues as string).split(',')
+      : [];
   if (method === 'POST' || (method === undefined && config.allowUndefinedMethod)) {
     const bodyParams = await getBodyParams(req);
 
@@ -68,7 +73,7 @@ export default async ({ config, method, req, res }: Options) => {
 
     const body = getBodyContent(bodyParams);
 
-    return storeStatements({ config, client, priority, body, attachments: [], res });
+    return storeStatements({ config, client, priority, bypassQueues, body, attachments: [], res });
   }
 
   if (method === 'GET') {
@@ -100,7 +105,16 @@ export default async ({ config, method, req, res }: Options) => {
     const body = getBodyContent(bodyParams);
     const statementId = bodyParams.statementId as string | undefined;
 
-    return storeStatement({ config, client, body, priority, attachments: [], statementId, res });
+    return storeStatement({
+      config,
+      client,
+      body,
+      priority,
+      bypassQueues,
+      attachments: [],
+      statementId,
+      res,
+    });
   }
 
   throw new InvalidMethod(method);

src/apps/statements/expressPresenter/postStatements/index.ts

@@ -15,6 +15,7 @@ import {
 import getClient from '../utils/getClient';
 import validateVersionHeader from '../utils/validateHeaderVersion';
 import { validateStatementProcessingPriority } from '../utils/validateStatementProcessingPriority';
+import { validateStatementBypassQueues } from '../utils/validateStatementBypassQueues';
 import alternateRequest from './alternateRequest';
 import storeStatements from './storeStatements';
 import storeWithAttachments from './storeWithAttachments';
@@ -38,9 +39,14 @@ export default (config: Config) => {
       const method = req.query.method as string | undefined;
 
       validateStatementProcessingPriority(req.query.priority as string | undefined);
+      validateStatementBypassQueues(req.query.bypassQueues as string | undefined);
 
       const priority =
         (req.query.priority as StatementProcessingPriority) || StatementProcessingPriority.MEDIUM;
+      const bypassQueues =
+        req.query.bypassQueues && (req.query.bypassQueues as string).trim() !== ''
+          ? (req.query.bypassQueues as string).split(',')
+          : [];
       const contentType = defaultTo(req.header('Content-Type'), '');
 
       if (method === undefined && multipartContentTypePattern.test(contentType)) {
@@ -53,7 +59,7 @@ export default (config: Config) => {
 
         const body = await parseJsonBody(config, req);
         const attachments: any[] = [];
-        return storeStatements({ config, client, priority, body, attachments, res });
+        return storeStatements({ config, client, priority, bypassQueues, body, attachments, res });
       }
 
       if (method !== undefined || alternateContentTypePattern.test(contentType)) {

src/apps/statements/expressPresenter/postStatements/storeStatements.ts

@@ -10,14 +10,29 @@ export interface Options {
   readonly config: Config;
   readonly client: ClientModel;
   readonly priority: StatementProcessingPriority;
+  readonly bypassQueues: string[];
   readonly body: any;
   readonly attachments: any[];
   readonly res: Response;
 }
 
-export default async ({ config, client, priority, body, attachments, res }: Options) => {
+export default async ({
+  config,
+  client,
+  priority,
+  bypassQueues,
+  body,
+  attachments,
+  res,
+}: Options) => {
   const models = isArray(body) ? body : [body];
-  const ids = await config.service.storeStatements({ priority, models, attachments, client });
+  const ids = await config.service.storeStatements({
+    priority,
+    bypassQueues,
+    models,
+    attachments,
+    client,
+  });
   res.setHeader('X-Experience-API-Version', xapiHeaderVersion);
   res.status(StatusCodes.OK);
   res.json(ids);

src/apps/statements/expressPresenter/postStatements/storeWithAttachments.ts

@@ -5,6 +5,7 @@ import Config from '../Config';
 import getClient from '../utils/getClient';
 import getMultipartStatements from '../utils/getMultipartStatements';
 import { validateStatementProcessingPriority } from '../utils/validateStatementProcessingPriority';
+import { validateStatementBypassQueues } from '../utils/validateStatementBypassQueues';
 import storeStatements from './storeStatements';
 
 export interface Options {
@@ -17,10 +18,15 @@ export default async ({ config, req, res }: Options) => {
   const client = await getClient(config, defaultTo(req.header('Authorization'), ''));
 
   validateStatementProcessingPriority(req.query.priority as string | undefined);
+  validateStatementBypassQueues(req.query.bypassQueues as string | undefined);
 
   const priority =
     (req.query.priority as StatementProcessingPriority) || StatementProcessingPriority.MEDIUM;
+  const bypassQueues =
+    req.query.bypassQueues && (req.query.bypassQueues as string).trim() !== ''
+      ? (req.query.bypassQueues as string).split(',')
+      : [];
   const { body, attachments } = await getMultipartStatements(req);
 
-  return storeStatements({ config, client, priority, body, attachments, res });
+  return storeStatements({ config, client, priority, bypassQueues, body, attachments, res });
 };

src/apps/statements/expressPresenter/putStatement.ts

@@ -13,29 +13,53 @@ import getMultipartStatements from './utils/getMultipartStatements';
 import storeStatement from './utils/storeStatement';
 import validateVersionHeader from './utils/validateHeaderVersion';
 import { validateStatementProcessingPriority } from './utils/validateStatementProcessingPriority';
+import { validateStatementBypassQueues } from './utils/validateStatementBypassQueues';
 
 export default (config: Config) => {
   return catchErrors(
     config,
     async (req: Request, res: Response): Promise<void> => {
       validateStatementProcessingPriority(req.query.priority as string | undefined);
+      validateStatementBypassQueues(req.query.bypassQueues as string | undefined);
       validateVersionHeader(req.header('X-Experience-API-Version'));
 
       const contentType = defaultTo(req.header('Content-Type'), '');
       const client = await getClient(config, defaultTo(req.header('Authorization'), ''));
       const priority =
         (req.query.priority as StatementProcessingPriority) || StatementProcessingPriority.MEDIUM;
+      const bypassQueues =
+        req.query.bypassQueues && (req.query.bypassQueues as string).trim() !== ''
+          ? (req.query.bypassQueues as string).split(',')
+          : [];
       const statementId = req.query.statementId as string;
 
       if (multipartContentTypePattern.test(contentType)) {
         const { body, attachments } = await getMultipartStatements(req);
-        return storeStatement({ config, priority, body, attachments, client, statementId, res });
+        return storeStatement({
+          config,
+          priority,
+          bypassQueues,
+          body,
+          attachments,
+          client,
+          statementId,
+          res,
+        });
       }
 
       if (jsonContentTypePattern.test(contentType)) {
         const body = parseJson(await streamToString(req), ['body']);
         const attachments: AttachmentModel[] = [];
-        return storeStatement({ config, priority, body, attachments, client, statementId, res });
+        return storeStatement({
+          config,
+          priority,
+          bypassQueues,
+          body,
+          attachments,
+          client,
+          statementId,
+          res,
+        });
       }
 
       throw new InvalidContentType(contentType);

src/apps/statements/expressPresenter/tests/postStatements/bypassQueues.test.ts

@@ -0,0 +1,79 @@
+import * as assert from 'assert';
+import { StatusCodes } from 'http-status-codes';
+import { xapiHeaderVersion } from '../../../../activities/utils/constants';
+import createClientModel from '../../../tests/utils/createClientModel';
+import createStatement from '../../../tests/utils/createStatement';
+import { jsonContentType, statementsRoute } from '../../../utils/constants';
+import setup from '../../tests/utils/setup';
+import repo from '../../../repo';
+
+describe('postStatements', () => {
+  const { supertest } = setup();
+
+  it('should throw error for incorrect bypassQueues query param', async () => {
+    const TEST_ID = '1c86d8e9-f325-404f-b3d9-24c451035586';
+
+    await supertest
+      .post(statementsRoute)
+      .set('Content-Type', jsonContentType)
+      .set('X-Experience-API-Version', xapiHeaderVersion)
+      .query({
+        bypassQueues: 'abc',
+        statementId: TEST_ID,
+      })
+      .send([createStatement()])
+      .expect((response) => {
+        assert.equal(response.status, StatusCodes.BAD_REQUEST);
+        assert.deepEqual(response.body.warnings, [
+          `Problem in 'query.bypassQueues'. Received '"abc"'`,
+        ]);
+      });
+  });
+
+  it('should insert statement with proper bypassQueues query param', async () => {
+    const TEST_ID = '1c86d8e9-f325-404f-b3d9-24c451035587';
+    const TEST_CLIENT = createClientModel();
+    const expectedCompletedQueues = ['STATEMENT_QUEUE_1', 'STATEMENT_QUEUE_2'];
+
+    await supertest
+      .post(statementsRoute)
+      .set('Content-Type', jsonContentType)
+      .set('X-Experience-API-Version', xapiHeaderVersion)
+      .query({
+        bypassQueues: expectedCompletedQueues.join(','),
+        statementId: TEST_ID,
+      })
+      .send([createStatement()])
+      .expect(async (response) => {
+        assert.equal(response.status, StatusCodes.NO_CONTENT);
+
+        const fullStatement = await repo.getStatement({ id: TEST_ID, client: TEST_CLIENT });
+
+        assert.deepEqual(fullStatement.completedQueues, expectedCompletedQueues);
+      });
+  });
+
+  it('should insert statement with proper bypassQueues query param (using alternate request)', async () => {
+    const TEST_ID = '1c86d8e9-f325-404f-b3d9-24c451035588';
+    const TEST_CLIENT = createClientModel();
+    const expectedCompletedQueues = ['STATEMENT_QUEUE_1', 'STATEMENT_QUEUE_2'];
+
+    await supertest
+      .post(statementsRoute)
+      .set('Content-Type', jsonContentType)
+      .set('X-Experience-API-Version', xapiHeaderVersion)
+      .query({
+        bypassQueues: expectedCompletedQueues.join(','),
+        statementId: TEST_ID,
+        method: 'PUT',
+      })
+      .send([createStatement()])
+      .expect(async (response) => {
+        assert.equal(response.status, StatusCodes.NO_CONTENT);
+
+        const fullStatement = await repo.getStatement({ id: TEST_ID, client: TEST_CLIENT });
+
+        assert.deepEqual(fullStatement.completedQueues, expectedCompletedQueues);
+      });
+  });
+});

src/apps/statements/expressPresenter/tests/putStatement/bypassQueues.test.ts

@@ -0,0 +1,55 @@
+import * as assert from 'assert';
+import { StatusCodes } from 'http-status-codes';
+import { xapiHeaderVersion } from '../../../../activities/utils/constants';
+import createClientModel from '../../../tests/utils/createClientModel';
+import createStatement from '../../../tests/utils/createStatement';
+import { jsonContentType, statementsRoute } from '../../../utils/constants';
+import setup from '../../tests/utils/setup';
+import repo from '../../../repo';
+
+describe('putStatement', () => {
+  const { supertest } = setup();
+
+  it('should throw error for incorrect bypassQueues query param', async () => {
+    const TEST_ID = '1c86d8e9-f325-404f-b3d9-24c451035586';
+
+    await supertest
+      .put(statementsRoute)
+      .set('Content-Type', jsonContentType)
+      .set('X-Experience-API-Version', xapiHeaderVersion)
+      .query({
+        bypassQueues: 'abc',
+        statementId: TEST_ID,
+      })
+      .send(createStatement())
+      .expect((response) => {
+        assert.equal(response.status, StatusCodes.BAD_REQUEST);
+        assert.deepEqual(response.body.warnings, [
+          `Problem in 'query.bypassQueues'. Received '"abc"'`,
+        ]);
+      });
+  });
+
+  it('should insert statement with proper bypassQueues query param', async () => {
+    const TEST_ID = '1c86d8e9-f325-404f-b3d9-24c451035587';
+    const TEST_CLIENT = createClientModel();
+    const expectedCompletedQueues = ['STATEMENT_QUEUE_1', 'STATEMENT_QUEUE_2'];
+
+    await supertest
+      .put(statementsRoute)
+      .set('Content-Type', jsonContentType)
+      .set('X-Experience-API-Version', xapiHeaderVersion)
+      .query({
+        bypassQueues: expectedCompletedQueues.join(','),
+        statementId: TEST_ID,
+      })
+      .send(createStatement())
+      .expect(async (response) => {
+        assert.equal(response.status, StatusCodes.NO_CONTENT);
+
+        const fullStatement = await repo.getStatement({ id: TEST_ID, client: TEST_CLIENT });
+
+        assert.deepEqual(fullStatement.completedQueues, expectedCompletedQueues);
+      });
+  });
+});

src/apps/statements/expressPresenter/tests/utils/setup.ts

@@ -0,0 +1,17 @@
+import setupService from 'jscommons/dist/tests/utils/setupService';
+import { SuperTest, Test } from 'supertest';
+import Service from '../../../serviceFactory/Service';
+import service from '../../../utils/testService';
+import supertest from './supertest';
+
+const setup = setupService(service);
+
+export interface Result {
+  readonly service: Service;
+  readonly supertest: SuperTest<Test>;
+}
+
+export default (): Result => {
+  setup();
+  return { service, supertest };
+};