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),
153 await getClient().send(command)
155 return getPrivateUrl(bucketInfo, objectStorageKey)
158 async function multiPartUpload (options: {
160 objectStorageKey: string
161 bucketInfo: BucketInfo
163 const { objectStorageKey, inputPath, bucketInfo } = options
165 const key = buildKey(objectStorageKey, bucketInfo)
166 const s3Client = getClient()
168 const statResult = await stat(inputPath)
170 const createMultipartCommand = new CreateMultipartUploadCommand({
171 Bucket: bucketInfo.BUCKET_NAME,
175 const createResponse = await s3Client.send(createMultipartCommand)
177 const fd = await open(inputPath, 'r')
179 const parts: CompletedPart[] = []
180 const partSize = CONFIG.OBJECT_STORAGE.MAX_UPLOAD_PART
182 for (let start = 0; start < statResult.size; start += partSize) {
184 'Uploading part %d of file to %s%s in bucket %s',
185 partNumber, bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, lTags()
188 // FIXME: Remove when https://github.com/aws/aws-sdk-js-v3/pull/2637 is released
189 // The s3 sdk needs to know the length of the http body beforehand, but doesn't support
190 // streams with start and end set, so it just tries to stat the file in stream.path.
191 // This fails for us because we only want to send part of the file. The stream type
192 // is modified so we can set the byteLength here, which s3 detects because array buffers
193 // have this field set
194 const stream: ReadStream & { byteLength: number } =
197 { fd, autoClose: false, start, end: (start + partSize) - 1 }
198 ) as ReadStream & { byteLength: number }
200 // Calculate if the part size is more than what's left over, and in that case use left over bytes for byteLength
201 stream.byteLength = min([ statResult.size - start, partSize ])
203 const uploadPartCommand = new UploadPartCommand({
204 Bucket: bucketInfo.BUCKET_NAME,
206 UploadId: createResponse.UploadId,
207 PartNumber: partNumber,
210 const uploadResponse = await s3Client.send(uploadPartCommand)
212 parts.push({ ETag: uploadResponse.ETag, PartNumber: partNumber })
217 const completeUploadCommand = new CompleteMultipartUploadCommand({
218 Bucket: bucketInfo.BUCKET_NAME,
220 UploadId: createResponse.UploadId,
221 MultipartUpload: { Parts: parts }
223 await s3Client.send(completeUploadCommand)
226 'Completed %s%s in bucket %s in %d parts',
227 bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, partNumber - 1, lTags()
230 return getPrivateUrl(bucketInfo, objectStorageKey)