1 import { close, createReadStream, createWriteStream, ensureDir, open, ReadStream, stat } from 'fs-extra'
2 import { min } from 'lodash'
3 import { dirname } from 'path'
4 import { Readable } from 'stream'
7 CompleteMultipartUploadCommand,
8 CreateMultipartUploadCommand,
14 } from '@aws-sdk/client-s3'
15 import { pipelinePromise } from '@server/helpers/core-utils'
16 import { isArray } from '@server/helpers/custom-validators/misc'
17 import { logger } from '@server/helpers/logger'
18 import { CONFIG } from '@server/initializers/config'
19 import { getPrivateUrl } from '../urls'
20 import { getClient } from './client'
21 import { lTags } from './logger'
28 async function storeObject (options: {
30 objectStorageKey: string
31 bucketInfo: BucketInfo
33 const { inputPath, objectStorageKey, bucketInfo } = options
35 logger.debug('Uploading file %s to %s%s in bucket %s', inputPath, bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, lTags())
37 const stats = await stat(inputPath)
39 // If bigger than max allowed size we do a multipart upload
40 if (stats.size > CONFIG.OBJECT_STORAGE.MAX_UPLOAD_PART) {
41 return multiPartUpload({ inputPath, objectStorageKey, bucketInfo })
44 const fileStream = createReadStream(inputPath)
45 return objectStoragePut({ objectStorageKey, content: fileStream, bucketInfo })
48 async function removeObject (filename: string, bucketInfo: BucketInfo) {
49 const command = new DeleteObjectCommand({
50 Bucket: bucketInfo.BUCKET_NAME,
51 Key: buildKey(filename, bucketInfo)
54 return getClient().send(command)
57 async function removePrefix (prefix: string, bucketInfo: BucketInfo) {
58 const s3Client = getClient()
60 const commandPrefix = bucketInfo.PREFIX + prefix
61 const listCommand = new ListObjectsV2Command({
62 Bucket: bucketInfo.BUCKET_NAME,
66 const listedObjects = await s3Client.send(listCommand)
68 // FIXME: use bulk delete when s3ninja will support this operation
69 // const deleteParams = {
70 // Bucket: bucketInfo.BUCKET_NAME,
71 // Delete: { Objects: [] }
74 if (isArray(listedObjects.Contents) !== true) {
75 const message = `Cannot remove ${commandPrefix} prefix in bucket ${bucketInfo.BUCKET_NAME}: no files listed.`
77 logger.error(message, { response: listedObjects, ...lTags() })
78 throw new Error(message)
81 for (const object of listedObjects.Contents) {
82 const command = new DeleteObjectCommand({
83 Bucket: bucketInfo.BUCKET_NAME,
87 await s3Client.send(command)
89 // FIXME: use bulk delete when s3ninja will support this operation
90 // deleteParams.Delete.Objects.push({ Key: object.Key })
93 // FIXME: use bulk delete when s3ninja will support this operation
94 // const deleteCommand = new DeleteObjectsCommand(deleteParams)
95 // await s3Client.send(deleteCommand)
97 // Repeat if not all objects could be listed at once (limit of 1000?)
98 if (listedObjects.IsTruncated) await removePrefix(prefix, bucketInfo)
101 async function makeAvailable (options: {
104 bucketInfo: BucketInfo
106 const { key, destination, bucketInfo } = options
108 await ensureDir(dirname(options.destination))
110 const command = new GetObjectCommand({
111 Bucket: bucketInfo.BUCKET_NAME,
112 Key: buildKey(key, bucketInfo)
114 const response = await getClient().send(command)
116 const file = createWriteStream(destination)
117 await pipelinePromise(response.Body as Readable, file)
122 function buildKey (key: string, bucketInfo: BucketInfo) {
123 return bucketInfo.PREFIX + key
126 // ---------------------------------------------------------------------------
137 // ---------------------------------------------------------------------------
139 async function objectStoragePut (options: {
140 objectStorageKey: string
142 bucketInfo: BucketInfo
144 const { objectStorageKey, content, bucketInfo } = options
146 const command = new PutObjectCommand({
147 Bucket: bucketInfo.BUCKET_NAME,
148 Key: buildKey(objectStorageKey, bucketInfo),
152 await getClient().send(command)
154 return getPrivateUrl(bucketInfo, objectStorageKey)
157 async function multiPartUpload (options: {
159 objectStorageKey: string
160 bucketInfo: BucketInfo
162 const { objectStorageKey, inputPath, bucketInfo } = options
164 const key = buildKey(objectStorageKey, bucketInfo)
165 const s3Client = getClient()
167 const statResult = await stat(inputPath)
169 const createMultipartCommand = new CreateMultipartUploadCommand({
170 Bucket: bucketInfo.BUCKET_NAME,
173 const createResponse = await s3Client.send(createMultipartCommand)
175 const fd = await open(inputPath, 'r')
177 const parts: CompletedPart[] = []
178 const partSize = CONFIG.OBJECT_STORAGE.MAX_UPLOAD_PART
180 for (let start = 0; start < statResult.size; start += partSize) {
182 'Uploading part %d of file to %s%s in bucket %s',
183 partNumber, bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, lTags()
186 // FIXME: Remove when https://github.com/aws/aws-sdk-js-v3/pull/2637 is released
187 // The s3 sdk needs to know the length of the http body beforehand, but doesn't support
188 // streams with start and end set, so it just tries to stat the file in stream.path.
189 // This fails for us because we only want to send part of the file. The stream type
190 // is modified so we can set the byteLength here, which s3 detects because array buffers
191 // have this field set
192 const stream: ReadStream & { byteLength: number } =
195 { fd, autoClose: false, start, end: (start + partSize) - 1 }
196 ) as ReadStream & { byteLength: number }
198 // Calculate if the part size is more than what's left over, and in that case use left over bytes for byteLength
199 stream.byteLength = min([ statResult.size - start, partSize ])
201 const uploadPartCommand = new UploadPartCommand({
202 Bucket: bucketInfo.BUCKET_NAME,
204 UploadId: createResponse.UploadId,
205 PartNumber: partNumber,
208 const uploadResponse = await s3Client.send(uploadPartCommand)
210 parts.push({ ETag: uploadResponse.ETag, PartNumber: partNumber })
215 const completeUploadCommand = new CompleteMultipartUploadCommand({
216 Bucket: bucketInfo.BUCKET_NAME,
217 Key: objectStorageKey,
218 UploadId: createResponse.UploadId,
219 MultipartUpload: { Parts: parts }
221 await s3Client.send(completeUploadCommand)
224 'Completed %s%s in bucket %s in %d parts',
225 bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, partNumber - 1, lTags()
228 return getPrivateUrl(bucketInfo, objectStorageKey)