Implement synchronization endpoints

app/src/controllers/index.js

@@ -4,6 +4,7 @@ module.exports = {
  metadataController: require('./metadata'),
  objectController: require('./object'),
  objectPermissionController: require('./objectPermission'),
+ syncController: require('./sync'),
  userController: require('./user'),
  tagController: require('./tag'),
  versionController: require('./version')

app/src/controllers/sync.js

@@ -0,0 +1,92 @@
+const { NIL: SYSTEM_USER } = require('uuid');
+
+const errorToProblem = require('../components/errorToProblem');
+const { addDashesToUuid, getCurrentIdentity, isTruthy } = require('../components/utils');
+const { objectService, storageService, objectQueueService, userService } = require('../services');
+
+const SERVICE = 'ObjectQueueService';
+
+/**
+ * The Sync Controller
+ */
+const controller = {
+ /**
+ * @function syncBucket
+ * Synchronizes a bucket
+ * @param {object} req Express request object
+ * @param {object} res Express response object
+ * @param {function} next The next callback function
+ * @returns {function} Express middleware function
+ */
+ async syncBucket(req, res, next) {
+ try {
+ const allMode = isTruthy(req.query.all);
+ const bucketId = addDashesToUuid(req.params.bucketId);
+ const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
+
+ const dbParams = {};
+ if (!allMode) dbParams.bucketId = bucketId;
+
+ const [dbResponse, s3Response] = await Promise.all([
+ objectService.searchObjects(dbParams),
+ storageService.listAllObjectVersions({ bucketId: bucketId, filterLatest: true })
+ ]);
+
+ // Aggregate and dedupe all file paths to consider
+ const jobs = [...new Set([
+ ...dbResponse.map(object => object.path),
+ ...s3Response.DeleteMarkers.map(object => object.Key),
+ ...s3Response.Versions.map(object => object.Key)
+ ])].map(path => ({ path: path, bucketId: bucketId }));
+
+ const response = await objectQueueService.enqueue({ jobs: jobs, full: isTruthy(req.query.full), createdBy: userId });
+ res.status(202).json(response);
+ } catch (e) {
+ next(errorToProblem(SERVICE, e));
+ }
+ },
+
+ /**
+ * @function syncObject
+ * Synchronizes an object
+ * @param {object} req Express request object
+ * @param {object} res Express response object
+ * @param {function} next The next callback function
+ * @returns {function} Express middleware function
+ */
+ async syncObject(req, res, next) {
+ try {
+ const bucketId = req.currentObject?.bucketId;
+ const path = req.currentObject?.path;
+ const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
+
+ const response = await objectQueueService.enqueue({
+ jobs: [{ path: path, bucketId: bucketId }],
+ full: isTruthy(req.query.full),
+ createdBy: userId
+ });
+ res.status(202).json(response);
+ } catch (e) {
+ next(errorToProblem(SERVICE, e));
+ }
+ },
+
+ /**
+ * @function syncStatus
+ * Reports on current sync queue size
+ * @param {object} req Express request object
+ * @param {object} res Express response object
+ * @param {function} next The next callback function
+ * @returns {function} Express middleware function
+ */
+ async syncStatus(_req, res, next) {
+ try {
+ const response = await objectQueueService.queueSize();
+ res.status(200).json(response);
+ } catch (e) {
+ next(errorToProblem(SERVICE, e));
+ }
+ }
+};
+
+module.exports = controller;

app/src/db/models/tables/objectModel.js

@@ -60,7 +60,11 @@ class ObjectModel extends Timestamps(Model) {
  filterOneOrMany(query, value, 'object.id');
  },
  filterBucketIds(query, value) {
- filterOneOrMany(query, value, 'object.bucketId');
+ if (value === null) {
+ query.whereNull('object.bucketId');
+ } else {
+ filterOneOrMany(query, value, 'object.bucketId');
+ }
  },
  filterName(query, value) {
  filterILike(query, value, 'object.name');
@@ -192,7 +196,7 @@ class ObjectModel extends Timestamps(Model) {
  path: { type: 'string', minLength: 1, maxLength: 1024 },
  public: { type: 'boolean' },
  active: { type: 'boolean' },
- bucketId: { type: 'string', maxLength: 255 },
+ bucketId: { type: 'string', maxLength: 255, nullable: true },
  name: { type: 'string', maxLength: 1024 },
  ...stamps
  },

app/src/db/models/tables/objectQueue.js

@@ -39,7 +39,7 @@ class ObjectModel extends Timestamps(Model) {
  required: ['path', 'full', 'retries'],
  properties: {
  id: { type: 'integer' },
- bucketId: { type: 'string', maxLength: 255 },
+ bucketId: { type: 'string', maxLength: 255, nullable: true },
  path: { type: 'string', minLength: 1, maxLength: 1024 },
  full: { type: 'boolean' },
  retries: { type: 'integer' },

app/src/routes/v1/bucket.js

@@ -1,7 +1,7 @@
 const router = require('express').Router();
 
 const { Permissions } = require('../../components/constants');
-const { bucketController } = require('../../controllers');
+const { bucketController, syncController } = require('../../controllers');
 const { bucketValidator } = require('../../validators');
 const { requireSomeAuth } = require('../../middleware/featureToggle');
 const { checkAppMode, hasPermission } = require('../../middleware/authorization');
@@ -44,4 +44,9 @@ router.delete('/:bucketId', bucketValidator.deleteBucket, hasPermission(Permissi
  bucketController.deleteBucket(req, res, next);
 });
 
+/** Synchronizes a bucket */
+router.get('/:bucketId/sync', bucketValidator.syncBucket, hasPermission(Permissions.READ), (req, res, next) => {
+ syncController.syncBucket(req, res, next);
+});
+
 module.exports = router;

app/src/routes/v1/index.js

@@ -12,6 +12,8 @@ router.get('/', (_req, res) => {
  '/metadata',
  '/object',
  '/permission',
+ '/sync',
+ '/tagging',
  '/user',
  '/version'
  ]
@@ -33,6 +35,9 @@ router.use('/object', require('./object'));
 /** Permission Router */
 router.use('/permission', require('./permission'));
 
+/** Sync Router */
+router.use('/sync', require('./sync'));
+
 /** Tagging Router */
 router.use('/tagging', require('./tag'));
 

app/src/routes/v1/object.js

@@ -1,7 +1,7 @@
 const router = require('express').Router();
 
 const { Permissions } = require('../../components/constants');
-const { objectController } = require('../../controllers');
+const { objectController, syncController } = require('../../controllers');
 const { objectValidator } = require('../../validators');
 const { requireSomeAuth } = require('../../middleware/featureToggle');
 const { checkAppMode, currentObject, hasPermission } = require('../../middleware/authorization');
@@ -74,6 +74,11 @@ router.delete('/:objectId/metadata', objectValidator.deleteMetadata, requireSome
  objectController.deleteMetadata(req, res, next);
 });
 
+/** Synchronizes an object */
+router.get('/:objectId/sync', objectValidator.syncObject, requireSomeAuth, currentObject, hasPermission(Permissions.READ), (req, res, next) => {
+ syncController.syncObject(req, res, next);
+});
+
 /** Add tags to an object */
 router.patch('/:objectId/tagging', objectValidator.addTags, requireSomeAuth, currentObject, hasPermission(Permissions.UPDATE), (req, res, next) => {
  objectController.addTags(req, res, next);

app/src/routes/v1/sync.js

@@ -0,0 +1,22 @@
+const router = require('express').Router();
+
+const { syncController } = require('../../controllers');
+const { syncValidator } = require('../../validators');
+const { checkAppMode } = require('../../middleware/authorization');
+const { requireBasicAuth, requireSomeAuth } = require('../../middleware/featureToggle');
+
+router.use(checkAppMode);
+router.use(requireSomeAuth);
+
+/** Synchronizes the default bucket */
+router.get('/', requireBasicAuth, syncValidator.syncDefault, (req, res, next) => {
+ req.params.bucketId = null;
+ syncController.syncBucket(req, res, next);
+});
+
+/** Check sync queue size */
+router.get('/status', (req, res, next) => {
+ syncController.syncStatus(req, res, next);
+});
+
+module.exports = router;

app/src/services/storage.js

@@ -24,6 +24,8 @@ const { MetadataDirective, TaggingDirective } = require('../components/constants
 const log = require('../components/log')(module.filename);
 const utils = require('../components/utils');
 
+const DELIMITER = '/';
+
 // Get app configuration
 const defaultTempExpiresIn = parseInt(config.get('objectStorage.defaultTempExpiresIn'), 10);
 
@@ -244,19 +246,21 @@ const objectStorageService = {
  */
  async listAllObjects({ filePath = undefined, bucketId = undefined, precisePath = true } = {}) {
  const key = filePath ?? (await utils.getBucket(bucketId)).key;
+ const path = key !== DELIMITER ? key : '';
+
  const objects = [];
 
  let incomplete = false;
  let nextToken = undefined;
  do {
  const { Contents, IsTruncated, NextContinuationToken } = await this.listObjectsV2({
- filePath: key,
+ filePath: path,
  continuationToken: nextToken,
  bucketId: bucketId
  });
 
  if (Contents) objects.push(
- ...Contents.filter(object => !precisePath || utils.isAtPath(key, object.Key))
+ ...Contents.filter(object => !precisePath || utils.isAtPath(path, object.Key))
  );
  incomplete = IsTruncated;
  nextToken = NextContinuationToken;
@@ -265,6 +269,49 @@ const objectStorageService = {
  return Promise.resolve(objects);
  },
 
+ /**
+ * @function listAllObjectVersions
+ * Lists all objects in the bucket with the prefix of `filePath`.
+ * Performs pagination behind the scenes if required.
+ * @param {string} [options.filePath=undefined] Optional filePath of the objects
+ * @param {string} [options.bucketId=undefined] Optional bucketId
+ * @param {boolean} [options.precisePath=true] Optional boolean for filtering results based on the precise path
+ * @param {boolean} [options.filterLatest=false] Optional boolean for filtering results to only entries with IsLatest being true
+ * @returns {Promise<object>} An object containg an array of DeleteMarkers and Versions
+ */
+ async listAllObjectVersions({ filePath = undefined, bucketId = undefined, precisePath = true, filterLatest = false } = {}) {
+ const key = filePath ?? (await utils.getBucket(bucketId)).key;
+ const path = key !== DELIMITER ? key : '';
+
+ const deleteMarkers = [];
+ const versions = [];
+
+ let incomplete = false;
+ let nextKeyMarker = undefined;
+ do {
+ const { DeleteMarkers, Versions, IsTruncated, NextKeyMarker } = await this.listObjectVersion({
+ filePath: path,
+ keyMarker: nextKeyMarker,
+ bucketId: bucketId
+ });
+
+ if (DeleteMarkers) deleteMarkers.push(
+ ...DeleteMarkers
+ .filter(object => !precisePath || utils.isAtPath(path, object.Key))
+ .filter(object => !filterLatest || object.IsLatest === true)
+ );
+ if (Versions) versions.push(
+ ...Versions
+ .filter(object => !precisePath || utils.isAtPath(path, object.Key))
+ .filter(object => !filterLatest || object.IsLatest === true)
+ );
+ incomplete = IsTruncated;
+ nextKeyMarker = NextKeyMarker;
+ } while (incomplete);
+
+ return Promise.resolve({ DeleteMarkers: deleteMarkers, Versions: versions });
+ },
+
  /**
  * @deprecated Use `listObjectsV2` instead
  * @function listObjects
@@ -296,11 +343,12 @@ const objectStorageService = {
  */
  async listObjectsV2({ filePath = undefined, continuationToken = undefined, maxKeys = undefined, bucketId = undefined } = {}) {
  const data = await utils.getBucket(bucketId);
+ const prefix = data.key !== DELIMITER ? data.key : '';
  const params = {
  Bucket: data.bucket,
  ContinuationToken: continuationToken,
  MaxKeys: maxKeys,
- Prefix: filePath ?? data.key // Must filter via "prefix" - https://stackoverflow.com/a/56569856
+ Prefix: filePath ?? prefix // Must filter via "prefix" - https://stackoverflow.com/a/56569856
  };
 
  return this._getS3Client(data).send(new ListObjectsV2Command(params));
@@ -309,15 +357,20 @@ const objectStorageService = {
  /**
  * @function ListObjectVersion
  * Lists the versions for the object at `filePath`
- * @param {string} options.filePath The filePath of the object
- * @param {string} [options.bucketId] Optional bucketId
+ * @param {string} [options.filePath=undefined] Optional filePath of the objects
+ * @param {string} [options.keyMarker=undefined] Optional keyMarker for pagination
+ * @param {number} [options.maxKeys=undefined] Optional maximum number of keys to return
+ * @param {string} [options.bucketId=undefined] Optional bucketId
  * @returns {Promise<object>} The response of the list object version operation
  */
- async listObjectVersion({ filePath, bucketId = undefined }) {
+ async listObjectVersion({ filePath = undefined, keyMarker = undefined, maxKeys = undefined, bucketId = undefined } = {}) {
  const data = await utils.getBucket(bucketId);
+ const prefix = data.key !== DELIMITER ? data.key : '';
  const params = {
  Bucket: data.bucket,
- Prefix: filePath // Must filter via "prefix" - https://stackoverflow.com/a/56569856
+ KeyMarker: keyMarker,
+ MaxKeys: maxKeys,
+ Prefix: filePath ?? prefix // Must filter via "prefix" - https://stackoverflow.com/a/56569856
  };
 
  return this._getS3Client(data).send(new ListObjectVersionsCommand(params));

app/src/validators/bucket.js

@@ -43,6 +43,15 @@ const schema = {
  })
  },
 
+ syncBucket: {
+ params: Joi.object({
+ bucketId: type.uuidv4.required()
+ }),
+ query: Joi.object({
+ full: type.truthy
+ })
+ },
+
  updateBucket: {
  body: Joi.object().keys({
  bucketName: Joi.string().max(255),
@@ -64,6 +73,7 @@ const validator = {
  deleteBucket: validate(schema.deleteBucket, { statusCode: 422 }),
  headBucket: validate(schema.headBucket, { statusCode: 422 }),
  readBucket: validate(schema.readBucket, { statusCode: 422 }),
+ syncBucket: validate(schema.readBucket, { statusCode: 422 }),
  searchBuckets: validate(schema.searchBuckets, { statusCode: 422 }),
  updateBucket: validate(schema.updateBucket, { statusCode: 422 })
 };

app/src/validators/index.js

@@ -4,6 +4,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/object.js

@@ -150,6 +150,15 @@ const schema = {
  })
  },
 
+ syncObject: {
+ params: Joi.object({
+ objectId: type.uuidv4.required()
+ }),
+ query: Joi.object({
+ full: type.truthy
+ })
+ },
+
  togglePublic: {
  params: Joi.object({
  objectId: type.uuidv4
@@ -185,6 +194,7 @@ const validator = {
  replaceMetadata: validate(schema.replaceMetadata, { statusCode: 422 }),
  replaceTags: validate(schema.replaceTags, { statusCode: 422 }),
  searchObjects: validate(schema.searchObjects, { statusCode: 422 }),
+ syncObject: validate(schema.searchObjects, { statusCode: 422 }),
  togglePublic: validate(schema.togglePublic, { statusCode: 422 }),
  updateObject: validate(schema.updateObject, { statusCode: 422 })
 };