Compaction API #350
+257
−32
Compact records based on timestamp and keep the latest one
- Loading branch information
| @@ -354,6 +354,7 @@ RequestUtil.prototype.s3ObjectsToRecords = function (s3Objects) { | ||
| const radix64 = require('../lib/radix64') | ||
| const output = [] | ||
| const partBuffer = {} | ||
| let objectMap = {} | ||
| for (let s3Object of s3Objects) { | ||
| const parsedKey = s3Helper.parseS3Key(s3Object.Key) | ||
| const fullCrc = parsedKey.recordCrc | ||
| @@ -362,14 +363,22 @@ RequestUtil.prototype.s3ObjectsToRecords = function (s3Objects) { | ||
| partBuffer[fullCrc] = partBuffer[fullCrc].concat(data) | ||
| data = partBuffer[fullCrc] | ||
| } | ||
| if (objectMap[fullCrc]) { | ||
| objectMap[fullCrc].push(s3Object.Key) | ||
| } else { | ||
| objectMap[fullCrc] = [s3Object.Key] | ||
| } | ||
| const dataBytes = s3Helper.s3StringToByteArray(data) | ||
| const dataCrc = radix64.fromNumber(crc.crc32.unsigned(dataBytes.buffer)) | ||
| if (dataCrc === fullCrc) { | ||
| let decrypted = {} | ||
| try { | ||
| decrypted = this.decrypt(dataBytes) | ||
| decrypted.syncTimestamp = parsedKey.timestamp | ||
| output.push(decrypted) | ||
| output.push( | ||
| { record: decrypted, | ||
| objects: objectMap[fullCrc] | ||
| }) | ||
| } catch (e) { | ||
| console.log(`Record with CRC ${crc} can't be decrypted: ${e}`) | ||
| } | ||
| @@ -420,6 +429,42 @@ RequestUtil.prototype.put = function (category, record) { | ||
| }) | ||
| } | ||
|
|
||
| RequestUtil.prototype.compactRecords = function (category) { | ||
| const thisCategory = category | ||
| if (!recordUtil.CATEGORY_IDS.includes(thisCategory)) { | ||
| throw new Error(`Unsupported sync category: ${category}`) | ||
| } | ||
|
|
||
| const prefix = `${this.apiVersion}/${this.userId}/${category}` | ||
| let options = { | ||
| MaxKeys: 1000, | ||
|
This conversation was marked as resolved
by SergeyZhukovsky
|
||
| Bucket: this.bucket, | ||
| Prefix: prefix | ||
| } | ||
| this.withRetry(() => { | ||
| return s3Helper.listObjects(this.s3, options, false) | ||
| }).then((s3Objects) => { | ||
| let latestRecords = {} | ||
|
||
| let s3ObjectsToDelete = [] | ||
| const recordObjects = this.s3ObjectsToRecords(s3Objects.contents) | ||
| console.error(recordObjects) | ||
| recordObjects.forEach((recordObject) => { | ||
| const record = recordObject.record | ||
|
This conversation was marked as resolved
by AlexeyBarabash
|
||
| if (latestRecords[record.objectId]) { | ||
| if (record.syncTimestamp > latestRecords[record.objectId].record.syncTimestamp) { | ||
| s3ObjectsToDelete = s3ObjectsToDelete.concat(latestRecords[record.objectId].objects) | ||
| latestRecords[record.objectId] = recordObject | ||
| } else { | ||
| s3ObjectsToDelete = s3ObjectsToDelete.concat(recordObject.objects) | ||
| } | ||
| } else { | ||
| latestRecords[record.objectId] = recordObject | ||
| } | ||
| }) | ||
| s3Helper.deleteObjects(this.s3, this.bucket, s3ObjectsToDelete) | ||
| }) | ||
| } | ||
|
|
||
| RequestUtil.prototype.s3PostFormData = function (objectKey) { | ||
| let formData = new FormData() // eslint-disable-line | ||
| formData.append('key', objectKey) | ||
ProTip!
Use n and p to navigate between commits in a pull request.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Do we list 1000 objects at a time? For example we have 10K records and a client sends that command. We will check only 1000 records. There could be a situation when we have 2K unique records and 8K need to be compacted, but we pull 1000 unique always and this will never be compacted. The list command returns
nextContinuationTokenwhich should be used on a next execution to avoid getting same records.