Sync Buckets recursively

Creates missing child 'buckets' in COMS db
Syncs provided bucket and all child buckets
Uses object sync queue

Author: TimCsaky

app/src/controllers/sync.js

@@ -1,7 +1,8 @@
 const { NIL: SYSTEM_USER } = require('uuid');
+const { Permissions } = require('../components/constants');
 
 const errorToProblem = require('../components/errorToProblem');
-const { addDashesToUuid, getCurrentIdentity } = require('../components/utils');
+const { addDashesToUuid, getCurrentIdentity, isAtPath } = require('../components/utils');
 const utils = require('../db/models/utils');
 const { bucketService, objectService, storageService, objectQueueService, userService } = require('../services');
 
@@ -11,6 +12,126 @@ const SERVICE = 'ObjectQueueService';
  * The Sync Controller
  */
 const controller = {
+
+  async syncBucketRecursive(req, res, next) {
+    try {
+      const bucketId = addDashesToUuid(req.params.bucketId);
+      const parentBucket = await bucketService.read(bucketId);
+      const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
+
+      const [
+        dbResponse, // buckets below the parent in COMS db
+        s3Response // 'directories' that exist below the parent's in S3
+      ] = await Promise.all([
+        bucketService.searchChildBuckets(parentBucket),
+        storageService.listChildDirectories({ bucketId: bucketId })
+      ]);
+      const dbChildBuckets = dbResponse;
+      const s3ChildKeys = s3Response.CommonPrefixes.map(element => {
+        return element.Prefix.replace(/\/$/, '');
+      });
+
+      /**
+       * create new 'buckets' in COMS db for each new directory found in S3
+       * assign all permissions to user that created parent bucket
+       */
+      const newDbBuckets = [];
+      const newKeys = s3ChildKeys.filter(x => !dbChildBuckets.map(b => b.key).includes(x));
+      for (const s3ChildKey of newKeys) {
+        const data = {
+          bucketName: s3ChildKey.substring(s3ChildKey.lastIndexOf('/') + 1),
+          accessKeyId: parentBucket.accessKeyId,
+          bucket: parentBucket.bucket,
+          endpoint: parentBucket.endpoint,
+          key: s3ChildKey,
+          secretAccessKey: parentBucket.secretAccessKey,
+          region: parentBucket.region ?? undefined,
+          active: parentBucket.active,
+          userId: parentBucket.createdBy ?? SYSTEM_USER,
+          permCodes: Object.values(Permissions)
+        };
+        const dbResponse = await bucketService.create(data);
+        newDbBuckets.push(dbResponse);
+      }
+      // Array of all buckets now in db
+      const allBuckets = [parentBucket].concat(dbChildBuckets).concat(newDbBuckets);
+
+      /**
+       * get arrays of objects from COMS db and S3 for comparison
+       */
+      const [dbObjects, s3Objects] = await Promise.all([
+        // get all objects in existing buckets in db
+        objectService.searchObjects({
+          bucketId: dbChildBuckets.map(b => b.bucketId)
+        }),
+        // list all objects (recursive) in S3
+        storageService.listAllObjectVersions({
+          bucketId: bucketId,
+          filePath: parentBucket.key,
+          precisePath: false,
+          filterLatest: true
+        })
+      ]);
+
+      /**
+       * 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: allBuckets
+              // .filter(b => b.key === object.Key.substring(0, object.Key.lastIndexOf('/')))
+              .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: allBuckets
+                .filter(b => isAtPath(b.key, object.Key))
+                .map(b => b.bucketId)[0]
+            };
+          })
+      ])];
+      // remove duplicates
+      const jobs = [...new Map(objects.map(o => [o.path, o])).values()];
+
+      const response = await utils.trxWrapper(async (trx) => {
+        // update 'lastSyncRequestedDate' value in db for each bucket
+        for (const bucket of allBuckets) {
+          await bucketService.update({
+            bucketId: bucket.bucketId,
+            userId: userId,
+            lastSyncRequestedDate: new Date().toISOString()
+          }, trx);
+        }
+        // queue jobs
+        return await objectQueueService.enqueue({ jobs: jobs }, trx);
+      });
+
+      res.status(202).json(response);
+    } catch (e) {
+      next(errorToProblem(SERVICE, e));
+    }
+  },
+
   /**
    * @function syncBucket
    * Synchronizes a bucket

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);
       },

app/src/routes/v1/bucket.js

@@ -55,7 +55,7 @@ router.put('/:bucketId/child', express.json(), bucketValidator.createBucketChild
 
 /** Synchronizes a bucket */
 router.get('/:bucketId/sync', bucketValidator.syncBucket, hasPermission(Permissions.READ), (req, res, next) => {
-  syncController.syncBucket(req, res, next);
+  syncController.syncBucketRecursive(req, res, next);
 });
 
 module.exports = router;

app/src/services/bucket.js

@@ -164,6 +164,13 @@ const service = {
       }));
   },
 
+  searchChildBuckets: async (parentBucket) => {
+    return Bucket.query()
+      .modify('filterKeyIsChild', parentBucket.key)
+      .modify('filterEndpoint', parentBucket.endpoint)
+      .where('bucket', parentBucket.bucket);
+    // .where('accessKeyId', parentBucket.accessKeyId)
+  },
   /**
    * @function read
    * Get a bucket db record based on bucketId

app/src/services/storage.js

@@ -397,6 +397,17 @@ const objectStorageService = {
     return this._getS3Client(data).send(new ListObjectVersionsCommand(params));
   },
 
+  async listChildDirectories({ bucketId }) {
+    const data = await utils.getBucket(bucketId);
+    const prefix = data.key !== DELIMITER ? data.key : '';
+    const params = {
+      Bucket: data.bucket,
+      Delimiter: DELIMITER,
+      Prefix: prefix + '/'
+    };
+    return this._getS3Client(data).send(new ListObjectVersionsCommand(params));
+  },
+
   /**
    * @function presignUrl
    * Generates a presigned url for the `command` with a limited expiration window