Sync status

app/src/controllers/sync.js

@@ -1,7 +1,12 @@
 const { NIL: SYSTEM_USER } = require('uuid');
 
 const errorToProblem = require('../components/errorToProblem');
-const { addDashesToUuid, getCurrentIdentity, formatS3KeyForCompare, isPrefixOfPath } = require('../components/utils');
+const {
+  addDashesToUuid,
+  getCurrentIdentity,
+  formatS3KeyForCompare,
+  isPrefixOfPath,
+  mixedQueryToArray } = require('../components/utils');
 const utils = require('../db/models/utils');
 const log = require('../components/log')(module.filename);
 
@@ -277,9 +282,11 @@ const controller = {
    * @param {function} next The next callback function
    * @returns {function} Express middleware function
    */
-  async syncStatus(_req, res, next) {
+  async syncStatus(req, res, next) {
     try {
-      const response = await objectQueueService.queueSize();
+      const bucketIds = mixedQueryToArray(req.query.bucketId);
+
+      const response = await objectQueueService.queueSize(bucketIds);
       res.status(200).json(response);
     } catch (e) {
       next(errorToProblem(SERVICE, e));

app/src/docs/v1.api-spec.yaml

@@ -1423,13 +1423,16 @@ paths:
   /sync/status:
     get:
       summary: Get sync queue size
-      description: Returns the number of objects currently awaiting processing by COMS.
+      description: >-
+        Returns the number of objects (optionally filtered by bucketId)
+        currently in the queue awaiting synchronization with the state in object-storage.
       operationId: syncStatus
       tags:
         - Sync
       parameters:
         - $ref: "#/components/parameters/Header-S3AccessMode-Bucket"
         - $ref: "#/components/parameters/Header-S3AccessMode-Endpoint"
+        - $ref: "#/components/parameters/Query-BucketId"
       responses:
         "200":
           description: Returns the number of objects currently in the queue

app/src/routes/v1/sync.js

@@ -3,6 +3,8 @@ const router = require('express').Router();
 const { syncController } = require('../../controllers');
 const { checkAppMode } = require('../../middleware/authorization');
 const { requireBasicAuth, requireSomeAuth } = require('../../middleware/featureToggle');
+const { syncValidator } = require('../../validators');
+
 
 router.use(checkAppMode);
 router.use(requireSomeAuth);
@@ -14,8 +16,9 @@ router.get('/', requireBasicAuth, (req, res, next) => {
 });
 
 /** Check sync queue size */
-router.get('/status', (req, res, next) => {
-  syncController.syncStatus(req, res, next);
-});
+router.get('/status',
+  syncValidator.syncStatus, (req, res, next) => {
+    syncController.syncStatus(req, res, next);
+  });
 
 module.exports = router;

app/src/services/objectQueue.js

@@ -72,11 +72,16 @@ const service = {
 
   /**
    * @function queueSize
-   * Returns the number of jobs currently waiting for processing in the object queue.
+   * Returns the number of jobs currently waiting for processing in the object queue,
+   * optionally filtered by the bucketId(s)
+   * @param {string[]} optional array of bucketIds
    * @returns {Promise<number>} An integer representing how many jobs are in the queue.
    */
-  async queueSize() {
-    return ObjectQueue.query().count().first()
+  async queueSize(bucketIds) {
+    return ObjectQueue.query()
+      .modify('filterBucketIds', bucketIds)
+      .count()
+      .first()
       .then(response => parseInt(response.count));
   },
 };

app/src/services/sync.js

@@ -168,20 +168,19 @@ const service = {
           });
         }
 
-        // Case: already synced - record & update public status as needed
+        // Case: already synced - update public status as needed
         if (comsObject) {
 
           response = await objectService.update({
             id: comsObject.id,
             userId: userId,
             path: comsObject.path,
-            // update if public in s3
-            public: s3Public ? s3Public : comsObject.public,
+            public: s3Public,
             lastSyncedDate: new Date().toISOString()
           }, trx);
 
           // if public flag changed mark as modified
-          if (s3Public && !comsObject.public) modified = true;
+          if (s3Public !== comsObject.public) modified = true;
         }
 
         // Case: not in COMS - insert new COMS object

app/src/validators/index.js

@@ -5,6 +5,7 @@ module.exports = {
   metadataValidator: require('./metadata'),
   objectValidator: require('./object'),
   objectPermissionValidator: require('./objectPermission'),
+  syncValidator: require('./sync'),
   tagValidator: require('./tag'),
   userValidator: require('./user'),
   versionValidator: require('./version'),

app/src/validators/sync.js

@@ -0,0 +1,20 @@
+const Joi = require('joi');
+
+const { scheme } = require('./common');
+const { validate } = require('../middleware/validation');
+
+
+const schema = {
+  syncStatus: {
+    query: Joi.object({
+      bucketId: scheme.guid,
+    })
+  }
+};
+
+const validator = {
+  syncStatus: validate(schema.syncStatus)
+};
+
+module.exports = validator;
+module.exports.schema = schema;

app/tests/unit/controllers/sync.spec.js

@@ -70,9 +70,13 @@ describe('syncObject', () => {
 describe('syncStatus', () => {
   const queueSizeSpy = jest.spyOn(objectQueueService, 'queueSize');
   const next = jest.fn();
+  const req = {
+    query: {
+      bucketId: undefined
+    }
+  };
 
   it('should return the current sync queue size', async () => {
-    const req = {};
     queueSizeSpy.mockResolvedValue(0);
 
     await controller.syncStatus(req, res, next);
@@ -84,7 +88,6 @@ describe('syncStatus', () => {
   });
 
   it('should handle unexpected errors', async () => {
-    const req = {};
     queueSizeSpy.mockImplementation(() => { throw new Error('error'); });
 
     await controller.syncStatus(req, res, next);

app/tests/unit/services/sync.spec.js

@@ -454,7 +454,7 @@ describe('syncObject', () => {
     const result = await service.syncObject(path, bucketId);
 
     expect(result).toBeTruthy();
-    expect(result.modified).toBeFalsy();
+    expect(result.modified).toBeTruthy();
     expect(result.object).toEqual(comsObject);
 
     expect(ObjectModel.startTransaction).toHaveBeenCalledTimes(1);
@@ -478,7 +478,7 @@ describe('syncObject', () => {
     expect(objectModelTrx.commit).toHaveBeenCalledTimes(1);
     expect(updateSpy).toHaveBeenCalledTimes(1);
     expect(updateSpy).toHaveBeenCalledWith(expect.objectContaining({
-      id: validUuidv4, path: path, public: true, lastSyncedDate: expect.anything(), userId: SYSTEM_USER
+      id: validUuidv4, path: path, public: false, lastSyncedDate: expect.anything(), userId: SYSTEM_USER
     }), expect.any(Object));
   });
 
@@ -492,7 +492,7 @@ describe('syncObject', () => {
     const result = await service.syncObject(path, bucketId);
 
     expect(result).toBeTruthy();
-    expect(result.modified).toBeFalsy();
+    expect(result.modified).toBeTruthy();
     expect(result.object).toEqual(comsObject);
 
     expect(ObjectModel.startTransaction).toHaveBeenCalledTimes(1);

app/tests/unit/validators/sync.spec.js

@@ -0,0 +1,20 @@
+const jestJoi = require('jest-joi');
+expect.extend(jestJoi.matchers);
+
+const { schema } = require('../../../src/validators/sync');
+const { scheme } = require('../../../src/validators/common');
+
+
+describe('syncStatus', () => {
+  describe('query', () => {
+    const query = schema.syncStatus.query.describe();
+
+    describe('bucketId', () => {
+      const bucketId = query.keys.bucketId;
+
+      it('is the expected schema', () => {
+        expect(bucketId).toEqual(scheme.guid.describe());
+      });
+    });
+  });
+});