Recursive Bucket Sync option

TimCsaky · TimCsaky · commit a2e7ba84a14e · 2025-03-20T16:47:51.000-07:00
diff --git a/app/src/controllers/sync.js b/app/src/controllers/sync.js
@@ -1,28 +1,197 @@
 const { NIL: SYSTEM_USER } = require('uuid');
 
 const errorToProblem = require('../components/errorToProblem');
-const { addDashesToUuid, getCurrentIdentity } = require('../components/utils');
+const { addDashesToUuid, getCurrentIdentity, isAtPath, isTruthy } = require('../components/utils');
 const utils = require('../db/models/utils');
-const { bucketService, objectService, storageService, objectQueueService, userService } = require('../services');
+const {
+  bucketPermissionService,
+  bucketService,
+  objectService,
+  storageService,
+  objectQueueService,
+  userService
+} = require('../services');
 
 const SERVICE = 'ObjectQueueService';
 
 /**
  * The Sync Controller
  */
 const controller = {
+
   /**
    * @function syncBucket
-   * Synchronizes a bucket
+   * Synchronizes COMS db with bucket and objects in S3
+   * calls subsequent controller functions depending on whether doing a 'resursive' sync
    * @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 {
-      // TODO: Consider adding an "all" mode for checking through all known objects and buckets for job enumeration
-      // const allMode = isTruthy(req.query.all);
+      if (!isTruthy(req.query.recursive)) {
+        this.syncBucketRecursive(req, res, next);
+      }
+      else {
+        this.syncBucketSingle(req, res, next);
+      }
+    } catch (e) {
+      next(errorToProblem(SERVICE, e));
+    }
+  },
+
+  /**
+   * @function syncBucketRecursive
+   * Synchronizes all objects and subfolders found at the Key and below for the given parent folder (bucket)
+   * NOTE: OIDC users reuire MANAGE permission to do a recursive sync on a folder
+   * All their permissions will be copied to any new sub-folders created
+   * @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 syncBucketRecursive(req, res, next) {
+    try {
+      const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
+      const bucketId = addDashesToUuid(req.params.bucketId);
+      const parentBucket = await bucketService.read(bucketId);
+      // current user's permissions on parent folder
+      const currentUserParentBucketPerms = userId !== SYSTEM_USER ? (await bucketPermissionService.searchPermissions({
+        bucketId: parentBucket.bucketId,
+        userId: userId
+      })).map(p => p.permCode) : [];
+
+      /**
+       * get the two following lists for comparison:
+       */
+      // get parent + child bucket records already in COMS db
+      const dbChildBuckets = await bucketService.searchChildBuckets(parentBucket);
+      let dbBuckets = [parentBucket].concat(dbChildBuckets);
+
+      // get 'folders' that exist below (and including) the parent 'folder'
+      const s3Response = await storageService.listAllObjectVersions({ bucketId: bucketId, precisePath: false });
+      const formatS3KeyForCompare = (k => {
+        let key = k.substr(0, k.lastIndexOf('/')); // remove trailing slash and file name
+        return key ? key : '/'; // if parent is root set as '/' to match convention in COMS db
+      });
+      const s3Keys = [...new Set([
+        ...s3Response.DeleteMarkers.map(object => formatS3KeyForCompare(object.Key)),
+        ...s3Response.Versions.map(object => formatS3KeyForCompare(object.Key)),
+      ])];
+      // console.log('s3Keys', s3Keys);
+
+
+      /**
+       * compare each list and sync (ie create or delete) bucket records in COMS db to match 'folders' that exist in S3
+       * note: we assign all permissions to all users that created parent bucket
+       */
+      // delete buckets not found in S3 from COMS db
+      const oldDbBuckets = dbBuckets.filter(b => !s3Keys.includes(b.key));
+      for (const dbBucket of oldDbBuckets) {
+        await bucketService.delete(dbBucket.bucketId);
+        dbBuckets = dbBuckets.filter(b => b.bucketId != dbBucket.bucketId);
+      }
+      // Create buckets only found in S3 in COMS db
+      const newS3Keys = s3Keys.filter(k => !dbBuckets.map(b => b.key).includes(k));
+      for (const s3Key of newS3Keys) {
+        const data = {
+          bucketName: s3Key.substring(s3Key.lastIndexOf('/') + 1),
+          accessKeyId: parentBucket.accessKeyId,
+          bucket: parentBucket.bucket,
+          endpoint: parentBucket.endpoint,
+          key: s3Key,
+          secretAccessKey: parentBucket.secretAccessKey,
+          region: parentBucket.region ?? undefined,
+          active: parentBucket.active,
+          userId: parentBucket.createdBy ?? SYSTEM_USER,
+          // if current user has MANAGE perm on parent folder, copy all their perms to new subfolders
+          // TODO: consider copy all user's permissions to new sub-folders
+          permCodes: currentUserParentBucketPerms.includes('MANAGE') ? currentUserParentBucketPerms : []
+        };
+        const dbResponse = await bucketService.create(data);
+        dbBuckets.push(dbResponse);
+      }
+
+      /**
+       * Sync all the objects found in all the parent and child 'folders'.
+       * by comparing objects in COMS db with the keys of the object found in S3
+       */
+      // get all objects in existing buckets in all 'buckets' in COMS db
+      const dbObjects = await objectService.searchObjects({
+        bucketId: dbBuckets.map(b => b.bucketId)
+      });
+      // get all objects below parent 'key' in S3
+      const s3Objects = s3Response;
+
+      /**
+       * merge arrays of objects from COMS db and S3 to form an array of jobs with format:
+       *  [ { path: '/images/img3.jpg', bucketId: '123' }, { path: '/images/album1/img1.jpg', bucketId: '456' } ]
+       */
+      const objects = [...new Set([
+        // objects already in database
+        ...dbObjects.data.map(object => {
+          return {
+            path: object.path,
+            bucketId: object.bucketId
+          };
+        }),
+        // DeleteMarkers found in S3
+        ...s3Objects.DeleteMarkers.map(object => {
+          return {
+            path: object.Key,
+            bucketId: dbBuckets
+              .filter(b => isAtPath(b.key, object.Key))
+              .map(b => b.bucketId)[0]
+          };
+        }),
+        // Versions found in S3
+        ...s3Objects.Versions
+          .filter(v => v.Size > 0) // is an file (not a 'directory')
+          .map(object => {
+            return {
+              path: object.Key,
+              bucketId: dbBuckets
+                .filter(b => isAtPath(b.key, object.Key))
+                .map(b => b.bucketId)[0],
+              // adding current userId will give ALL perms on new objects
+              // and set createdBy on all downstream resources (versions, tags, meta)
+              // userId: userId
+            };
+          })
+      ])];
+      // merge and remove duplicates
+      const jobs = [...new Map(objects.map(o => [o.path, o])).values()];
+
+      // create jobs in COMS db object_queue for each object
+      const response = await utils.trxWrapper(async (trx) => {
+        // update 'lastSyncRequestedDate' value in COMS db for each bucket
+        for (const bucket of dbBuckets) {
+          await bucketService.update({
+            bucketId: bucket.bucketId,
+            userId: userId,
+            lastSyncRequestedDate: new Date().toISOString()
+          }, trx);
+        }
+        return await objectQueueService.enqueue({ jobs: jobs }, trx);
+      });
+      // return number of jobs inserted
+      res.status(202).json(response);
+    } catch (e) {
+      next(errorToProblem(SERVICE, e));
+    }
+  },
+
+  /**
+   * @function syncBucketSingle
+   * Synchronizes objects found at the Key of the given bucket, ignoring subfolders and files after the next delimiter
+   * @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 syncBucketSingle(req, res, next) {
+    try {
       const bucketId = addDashesToUuid(req.params.bucketId);
       const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
 
@@ -36,7 +205,13 @@ const controller = {
         ...dbResponse.data.map(object => object.path),
         ...s3Response.DeleteMarkers.map(object => object.Key),
         ...s3Response.Versions.map(object => object.Key)
-      ])].map(path => ({ path: path, bucketId: bucketId }));
+      ])].map(path => ({
+        path: path,
+        bucketId: bucketId
+        // adding current userId will give ALL perms on new objects
+        // and set createdBy on all downstream resources (versions, tags, meta)
+        // userId: userId
+      }));
 
       const response = await utils.trxWrapper(async (trx) => {
         await bucketService.update({
diff --git a/app/src/db/models/tables/bucket.js b/app/src/db/models/tables/bucket.js
@@ -48,9 +48,19 @@ class Bucket extends mixin(Model, [
       filterBucketName(query, value) {
         filterILike(query, value, 'bucket.bucketName');
       },
+      filterEndpoint(query, value) {
+        filterILike(query, value, 'bucket.endpoint');
+      },
       filterKey(query, value) {
         filterILike(query, value, 'bucket.key');
       },
+      filterKeyIsChild(query, value) {
+        if (value && value !== '/') {
+          query.where('bucket.key', 'like', `${value}%`);
+        }
+        query
+          .where('bucket.key', '!=', value);
+      },
       filterActive(query, value) {
         if (value !== undefined) query.where('bucket.active', value);
       },
diff --git a/app/src/services/bucket.js b/app/src/services/bucket.js
@@ -164,6 +164,20 @@ const service = {
       }));
   },
 
+  /**
+   * @function searchChildBuckets
+   * Get db records for each bucket that acts as a sub-folder of the provided bucket
+   * @param {object} parentBucket a bucket model (record) from the COMS db
+   * @returns {Promise<object[]>} An array of bucket records
+   * @throws If there are no records found
+   */
+  searchChildBuckets: async (parentBucket) => {
+    return Bucket.query()
+      .modify('filterKeyIsChild', parentBucket.key)
+      .modify('filterEndpoint', parentBucket.endpoint)
+      .where('bucket', parentBucket.bucket);
+  },
+
   /**
    * @function read
    * Get a bucket db record based on bucketId
diff --git a/app/src/services/objectQueue.js b/app/src/services/objectQueue.js
@@ -37,7 +37,7 @@ const service = {
    * Adds a set of jobs to the object queue. If the same job already exists, the existing
    * job will take precedence and will not be reinserted.
    * @param {object[]} options.jobs An array of job objects typically containing `path` and `bucketId` attributes
-   * @param {boolean} [options.full=false] Optional boolean indicating whether to execute full recursive run
+   * @param {boolean} [options.full=false] Optional boolean whether to sync related versions, tags, metadata
    * @param {integer} [options.retries=0] Optional integer indicating how many previous retries this job has had
    * @param {string} [options.createdBy=SYSTEM_USER] Optional uuid attributing which user added the job
    * @param {object} [etrx=undefined] Optional Objection Transaction object
diff --git a/app/src/validators/bucket.js b/app/src/validators/bucket.js
@@ -59,6 +59,9 @@ const schema = {
   syncBucket: {
     params: Joi.object({
       bucketId: type.uuidv4.required()
+    }),
+    query: Joi.object({
+      recursive: type.truthy,
     })
   },
 
@@ -84,7 +87,7 @@ const validator = {
   deleteBucket: validate(schema.deleteBucket),
   headBucket: validate(schema.headBucket),
   readBucket: validate(schema.readBucket),
-  syncBucket: validate(schema.readBucket),
+  syncBucket: validate(schema.syncBucket),
   searchBuckets: validate(schema.searchBuckets),
   updateBucket: validate(schema.updateBucket)
 };
diff --git a/app/tests/unit/controllers/sync.spec.js b/app/tests/unit/controllers/sync.spec.js
@@ -30,7 +30,7 @@ afterEach(() => {
   jest.resetAllMocks();
 });
 
-describe('syncBucket', () => {
+describe('syncBucketSingle', () => {
   const enqueueSpy = jest.spyOn(objectQueueService, 'enqueue');
   const getCurrentIdentitySpy = jest.spyOn(utils, 'getCurrentIdentity');
   const getCurrentUserIdSpy = jest.spyOn(userService, 'getCurrentUserId');
@@ -40,7 +40,7 @@ describe('syncBucket', () => {
   const updateSpy = jest.spyOn(bucketService, 'update');
   const next = jest.fn();
 
-  it('should enqueue all objects in a bucket', async () => {
+  it('should enqueue all objects in in current \'directory\' in bucket', async () => {
     const USR_IDENTITY = 'xxxy';
     const USR_ID = 'abc-123';
     const req = {
@@ -58,7 +58,7 @@ describe('syncBucket', () => {
     trxWrapperSpy.mockImplementation(callback => callback({}));
     updateSpy.mockResolvedValue({});
 
-    await controller.syncBucket(req, res, next);
+    await controller.syncBucketSingle(req, res, next);
 
     expect(enqueueSpy).toHaveBeenCalledTimes(1);
     expect(listAllObjectVersionsSpy).toHaveBeenCalledTimes(1);
@@ -76,7 +76,7 @@ describe('syncBucket', () => {
     listAllObjectVersionsSpy.mockImplementation(() => { throw new Error('error'); });
     searchObjectsSpy.mockResolvedValue([{ path: path }]);
 
-    await controller.syncBucket(req, res, next);
+    await controller.syncBucketSingle(req, res, next);
 
     expect(enqueueSpy).toHaveBeenCalledTimes(0);
     expect(listAllObjectVersionsSpy).toHaveBeenCalledTimes(1);
