1 import { createReadStream, createWriteStream, ensureDir, ReadStream } from 'fs-extra'
2 import { dirname } from 'path'
3 import { Readable } from 'stream'
5 CompleteMultipartUploadCommandOutput,
10 } from '@aws-sdk/client-s3'
11 import { Upload } from '@aws-sdk/lib-storage'
12 import { pipelinePromise } from '@server/helpers/core-utils'
13 import { isArray } from '@server/helpers/custom-validators/misc'
14 import { logger } from '@server/helpers/logger'
15 import { CONFIG } from '@server/initializers/config'
16 import { getPrivateUrl } from '../urls'
17 import { getClient } from './client'
18 import { lTags } from './logger'
25 async function listKeysOfPrefix (prefix: string, bucketInfo: BucketInfo) {
26 const s3Client = getClient()
28 const commandPrefix = bucketInfo.PREFIX + prefix
29 const listCommand = new ListObjectsV2Command({
30 Bucket: bucketInfo.BUCKET_NAME,
34 const listedObjects = await s3Client.send(listCommand)
36 if (isArray(listedObjects.Contents) !== true) return []
38 return listedObjects.Contents.map(c => c.Key)
41 // ---------------------------------------------------------------------------
43 async function storeObject (options: {
45 objectStorageKey: string
46 bucketInfo: BucketInfo
48 const { inputPath, objectStorageKey, bucketInfo } = options
50 logger.debug('Uploading file %s to %s%s in bucket %s', inputPath, bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, lTags())
52 const fileStream = createReadStream(inputPath)
54 return uploadToStorage({ objectStorageKey, content: fileStream, bucketInfo })
57 // ---------------------------------------------------------------------------
59 async function removeObject (filename: string, bucketInfo: BucketInfo) {
60 const command = new DeleteObjectCommand({
61 Bucket: bucketInfo.BUCKET_NAME,
62 Key: buildKey(filename, bucketInfo)
65 return getClient().send(command)
68 async function removePrefix (prefix: string, bucketInfo: BucketInfo) {
69 const s3Client = getClient()
71 const commandPrefix = bucketInfo.PREFIX + prefix
72 const listCommand = new ListObjectsV2Command({
73 Bucket: bucketInfo.BUCKET_NAME,
77 const listedObjects = await s3Client.send(listCommand)
79 // FIXME: use bulk delete when s3ninja will support this operation
80 // const deleteParams = {
81 // Bucket: bucketInfo.BUCKET_NAME,
82 // Delete: { Objects: [] }
85 if (isArray(listedObjects.Contents) !== true) {
86 const message = `Cannot remove ${commandPrefix} prefix in bucket ${bucketInfo.BUCKET_NAME}: no files listed.`
88 logger.error(message, { response: listedObjects, ...lTags() })
89 throw new Error(message)
92 for (const object of listedObjects.Contents) {
93 const command = new DeleteObjectCommand({
94 Bucket: bucketInfo.BUCKET_NAME,
98 await s3Client.send(command)
100 // FIXME: use bulk delete when s3ninja will support this operation
101 // deleteParams.Delete.Objects.push({ Key: object.Key })
104 // FIXME: use bulk delete when s3ninja will support this operation
105 // const deleteCommand = new DeleteObjectsCommand(deleteParams)
106 // await s3Client.send(deleteCommand)
108 // Repeat if not all objects could be listed at once (limit of 1000?)
109 if (listedObjects.IsTruncated) await removePrefix(prefix, bucketInfo)
112 // ---------------------------------------------------------------------------
114 async function makeAvailable (options: {
117 bucketInfo: BucketInfo
119 const { key, destination, bucketInfo } = options
121 await ensureDir(dirname(options.destination))
123 const command = new GetObjectCommand({
124 Bucket: bucketInfo.BUCKET_NAME,
125 Key: buildKey(key, bucketInfo)
127 const response = await getClient().send(command)
129 const file = createWriteStream(destination)
130 await pipelinePromise(response.Body as Readable, file)
135 function buildKey (key: string, bucketInfo: BucketInfo) {
136 return bucketInfo.PREFIX + key
139 // ---------------------------------------------------------------------------
151 // ---------------------------------------------------------------------------
153 async function uploadToStorage (options: {
155 objectStorageKey: string
156 bucketInfo: BucketInfo
158 const { content, objectStorageKey, bucketInfo } = options
160 const input: PutObjectCommandInput = {
162 Bucket: bucketInfo.BUCKET_NAME,
163 Key: buildKey(objectStorageKey, bucketInfo)
166 if (CONFIG.OBJECT_STORAGE.UPLOAD_ACL) {
167 input.ACL = CONFIG.OBJECT_STORAGE.UPLOAD_ACL
170 const parallelUploads3 = new Upload({
173 partSize: CONFIG.OBJECT_STORAGE.MAX_UPLOAD_PART,
175 // `leavePartsOnError` must be set to `true` to avoid silently dropping failed parts
176 // More detailed explanation:
177 // https://github.com/aws/aws-sdk-js-v3/blob/v3.164.0/lib/lib-storage/src/Upload.ts#L274
178 // https://github.com/aws/aws-sdk-js-v3/issues/2311#issuecomment-939413928
179 leavePartsOnError: true,
183 const response = (await parallelUploads3.done()) as CompleteMultipartUploadCommandOutput
184 // Check is needed even if the HTTP status code is 200 OK
185 // For more information, see https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html
186 if (!response.Bucket) {
187 const message = `Error uploading ${objectStorageKey} to bucket ${bucketInfo.BUCKET_NAME}`
188 logger.error(message, { response, ...lTags() })
189 throw new Error(message)
193 'Completed %s%s in bucket %s',
194 bucketInfo.PREFIX, objectStorageKey, bucketInfo.BUCKET_NAME, lTags()
197 return getPrivateUrl(bucketInfo, objectStorageKey)