const { S3Client, HeadBucketCommand, HeadObjectCommand, GetObjectCommand, PutObjectCommand, DeleteObjectCommand, DeleteObjectsCommand, ListObjectsV2Command, CopyObjectCommand, CreateMultipartUploadCommand, UploadPartCommand, CompleteMultipartUploadCommand, AbortMultipartUploadCommand } = require('@aws-sdk/client-s3'); const { Upload } = require('@aws-sdk/lib-storage'); const { getSignedUrl } = require('@aws-sdk/s3-request-presigner'); const fs = require('fs'); const fsPromises = require('fs').promises; const path = require('path'); const stream = require('stream'); const crypto = require('crypto'); const logger = require('../../utils/logger'); /** * S3 Storage Adapter for handling file uploads to S3 and S3-compatible services * * Features: * - Support for S3 and S3-compatible services (MinIO, DigitalOcean Spaces, etc.) * - Multipart upload for large files (>100MB) * - Progress tracking with event emitter * - Retry logic with exponential backoff * - Stream support for memory efficiency * - Path-style URL support for MinIO * - Connection testing * - Comprehensive error handling * * @class S3StorageAdapter */ class S3StorageAdapter extends stream.EventEmitter { /** * Creates an instance of S3StorageAdapter * * @param {Object} config - Configuration object * @param {string} config.bucket - S3 bucket name * @param {string} [config.region='us-east-1'] - AWS region * @param {string} [config.endpoint] - Custom endpoint URL for S3-compatible services * @param {string} [config.accessKeyId] - AWS access key ID * @param {string} [config.secretAccessKey] - AWS secret access key * @param {boolean} [config.forcePathStyle=false] - Force path-style URLs (required for MinIO) * @param {boolean} [config.sslEnabled=true] - Enable SSL for connections * @param {number} [config.multipartThreshold=104857600] - Threshold for multipart upload (default 100MB) * @param {number} [config.partSize=10485760] - Part size for multipart upload (default 10MB) * @param {number} [config.maxRetries=3] - Maximum number of retry attempts * @param {number} [config.retryDelay=1000] - Initial retry delay in milliseconds */ constructor(config) { super(); // Validate required config if (!config.bucket) { throw new Error('S3 bucket name is required'); } // Set defaults this.config = { region: 'us-east-1', forcePathStyle: false, sslEnabled: true, multipartThreshold: 100 * 1024 * 1024, // 100MB partSize: 10 * 1024 * 1024, // 10MB maxRetries: 3, retryDelay: 1000, ...config }; // Initialize S3 client const s3Config = { region: this.config.region, forcePathStyle: this.config.forcePathStyle }; // Add credentials if provided if (this.config.accessKeyId && this.config.secretAccessKey) { s3Config.credentials = { accessKeyId: this.config.accessKeyId, secretAccessKey: this.config.secretAccessKey }; } // Add custom endpoint if provided (for S3-compatible services) if (this.config.endpoint) { s3Config.endpoint = this.config.endpoint; // For MinIO and other S3-compatible services if (!this.config.endpoint.startsWith('https://') && this.config.sslEnabled) { s3Config.endpoint = `https://${this.config.endpoint}`; } else if (!this.config.endpoint.startsWith('http://') && !this.config.sslEnabled) { s3Config.endpoint = `http://${this.config.endpoint}`; } } this.s3Client = new S3Client(s3Config); this.bucket = this.config.bucket; // Bind methods to preserve context this.upload = this.upload.bind(this); this.uploadStream = this.uploadStream.bind(this); this.download = this.download.bind(this); this.downloadStream = this.downloadStream.bind(this); this.delete = this.delete.bind(this); this.exists = this.exists.bind(this); this.list = this.list.bind(this); this.copy = this.copy.bind(this); this.move = this.move.bind(this); this.getSignedUrl = this.getSignedUrl.bind(this); this.testConnection = this.testConnection.bind(this); } /** * Test connection to S3 bucket * * @returns {Promise} - True if connection successful * @throws {Error} - If connection fails */ async testConnection() { try { await this.s3Client.send(new HeadBucketCommand({ Bucket: this.bucket })); logger.info(`Successfully connected to S3 bucket: ${this.bucket}`); return true; } catch (error) { logger.error(`Failed to connect to S3 bucket ${this.bucket}:`, error); throw new Error(`S3 connection test failed: ${error.message}`); } } /** * Upload a file to S3 with automatic multipart for large files * * @param {string} localPath - Local file path to upload * @param {string} s3Key - S3 object key (path in bucket) * @param {Object} [options={}] - Additional options * @param {Object} [options.metadata] - Object metadata * @param {string} [options.contentType] - Content type * @param {string} [options.cacheControl] - Cache control header * @param {Function} [options.onProgress] - Progress callback function(loaded, total) * @returns {Promise} - Upload result with Location, ETag, etc. */ async upload(localPath, s3Key, options = {}) { try { const stats = await fsPromises.stat(localPath); const fileSize = stats.size; // Emit upload start event this.emit('uploadStart', { key: s3Key, size: fileSize }); // Create file stream const fileStream = fs.createReadStream(localPath); // Prepare upload parameters const uploadParams = { Bucket: this.bucket, Key: s3Key, Body: fileStream, ContentType: options.contentType || 'application/octet-stream', Metadata: options.metadata || {}, CacheControl: options.cacheControl }; // Remove undefined values Object.keys(uploadParams).forEach(key => uploadParams[key] === undefined && delete uploadParams[key]); // Use AWS SDK v3 Upload for automatic multipart handling const parallelUploads3 = new Upload({ client: this.s3Client, params: uploadParams, queueSize: 4, // Optional: concurrent uploads partSize: this.config.partSize, leavePartsOnError: false }); // Track upload progress parallelUploads3.on('httpUploadProgress', (progress) => { if (options.onProgress) { options.onProgress(progress.loaded, progress.total); } this.emit('uploadProgress', { key: s3Key, loaded: progress.loaded, total: progress.total }); }); // Perform upload with retry const result = await this._retryOperation(() => parallelUploads3.done()); this.emit('uploadComplete', { key: s3Key, location: result.Location }); return result; } catch (error) { logger.error(`Failed to upload file ${localPath} to S3:`, error); this.emit('uploadError', { key: s3Key, error }); throw error; } } /** * Upload a stream to S3 * * @param {stream.Readable} readStream - Readable stream to upload * @param {string} s3Key - S3 object key * @param {Object} [options={}] - Additional options * @returns {Promise} - Upload result */ async uploadStream(readStream, s3Key, options = {}) { const uploadParams = { Bucket: this.bucket, Key: s3Key, Body: readStream, ContentType: options.contentType || 'application/octet-stream', Metadata: options.metadata || {}, CacheControl: options.cacheControl }; // Remove undefined values Object.keys(uploadParams).forEach(key => uploadParams[key] === undefined && delete uploadParams[key]); const parallelUploads3 = new Upload({ client: this.s3Client, params: uploadParams, queueSize: 4, partSize: this.config.partSize, leavePartsOnError: false }); // Track progress if callback provided if (options.onProgress) { parallelUploads3.on('httpUploadProgress', (progress) => { options.onProgress(progress.loaded, progress.total); this.emit('uploadProgress', { key: s3Key, ...progress }); }); } return await this._retryOperation(() => parallelUploads3.done()); } /** * Download a file from S3 * * @param {string} s3Key - S3 object key * @param {string} localPath - Local file path to save to * @param {Object} [options={}] - Additional options * @param {Function} [options.onProgress] - Progress callback * @returns {Promise} */ async download(s3Key, localPath, options = {}) { try { // Ensure directory exists await fsPromises.mkdir(path.dirname(localPath), { recursive: true }); // Get object metadata first for progress tracking const headResult = await this._retryOperation(() => this.s3Client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: s3Key })) ); const fileSize = headResult.ContentLength; this.emit('downloadStart', { key: s3Key, size: fileSize }); // Get object const getObjectResult = await this._retryOperation(() => this.s3Client.send(new GetObjectCommand({ Bucket: this.bucket, Key: s3Key })) ); // Create write stream const writeStream = fs.createWriteStream(localPath); // Download with progress tracking await new Promise((resolve, reject) => { let downloaded = 0; const bodyStream = getObjectResult.Body; bodyStream.on('data', (chunk) => { downloaded += chunk.length; if (options.onProgress) { options.onProgress(downloaded, fileSize); } this.emit('downloadProgress', { key: s3Key, loaded: downloaded, total: fileSize }); }); bodyStream.on('error', reject); writeStream.on('error', reject); writeStream.on('finish', resolve); bodyStream.pipe(writeStream); }); this.emit('downloadComplete', { key: s3Key }); } catch (error) { logger.error(`Failed to download file ${s3Key} from S3:`, error); this.emit('downloadError', { key: s3Key, error }); throw error; } } /** * Get a download stream from S3 * * @param {string} s3Key - S3 object key * @param {Object} [options={}] - Additional options * @param {string} [options.range] - Byte range to download (e.g., 'bytes=0-1023') * @returns {Promise} - Readable stream */ async downloadStream(s3Key, options = {}) { const params = { Bucket: this.bucket, Key: s3Key, Range: options.range }; // Remove undefined values Object.keys(params).forEach(key => params[key] === undefined && delete params[key]); const result = await this.s3Client.send(new GetObjectCommand(params)); return result.Body; } /** * Delete a file from S3 * * @param {string} s3Key - S3 object key * @returns {Promise} */ async delete(s3Key) { return await this._retryOperation(() => this.s3Client.send(new DeleteObjectCommand({ Bucket: this.bucket, Key: s3Key })) ); } /** * Delete multiple files from S3 * * @param {string[]} s3Keys - Array of S3 object keys * @returns {Promise} - Delete result */ async deleteMany(s3Keys) { if (!s3Keys || s3Keys.length === 0) { return { Deleted: [], Errors: [] }; } // S3 deleteObjects has a limit of 1000 keys per request const chunks = []; for (let i = 0; i < s3Keys.length; i += 1000) { chunks.push(s3Keys.slice(i, i + 1000)); } const results = await Promise.all( chunks.map(chunk => this._retryOperation(() => this.s3Client.send(new DeleteObjectsCommand({ Bucket: this.bucket, Delete: { Objects: chunk.map(key => ({ Key: key })), Quiet: false } })) ) ) ); // Combine results return { Deleted: results.flatMap(r => r.Deleted || []), Errors: results.flatMap(r => r.Errors || []) }; } /** * Check if a file exists in S3 * * @param {string} s3Key - S3 object key * @returns {Promise} - True if exists */ async exists(s3Key) { try { await this.s3Client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: s3Key })); return true; } catch (error) { if (error.name === 'NotFound' || error.$metadata?.httpStatusCode === 404) { return false; } throw error; } } /** * List files in S3 * * @param {string} prefix - S3 prefix to list * @param {Object} [options={}] - Additional options * @param {number} [options.maxKeys=1000] - Maximum number of keys to return * @param {string} [options.continuationToken] - Continuation token for pagination * @returns {Promise} - List result with Contents array and NextContinuationToken */ async list(prefix, options = {}) { const params = { Bucket: this.bucket, Prefix: prefix, MaxKeys: options.maxKeys || 1000, ContinuationToken: options.continuationToken }; return await this._retryOperation(() => this.s3Client.send(new ListObjectsV2Command(params)) ); } /** * Copy a file within S3 * * @param {string} sourceKey - Source S3 object key * @param {string} targetKey - Target S3 object key * @param {Object} [options={}] - Additional options * @returns {Promise} - Copy result */ async copy(sourceKey, targetKey, options = {}) { const copySource = `${this.bucket}/${sourceKey}`; const params = { Bucket: this.bucket, CopySource: copySource, Key: targetKey, MetadataDirective: options.metadata ? 'REPLACE' : 'COPY', Metadata: options.metadata, ContentType: options.contentType, CacheControl: options.cacheControl }; // Remove undefined values Object.keys(params).forEach(key => params[key] === undefined && delete params[key]); return await this._retryOperation(() => this.s3Client.send(new CopyObjectCommand(params)) ); } /** * Move a file within S3 (copy then delete) * * @param {string} sourceKey - Source S3 object key * @param {string} targetKey - Target S3 object key * @param {Object} [options={}] - Additional options * @returns {Promise} - Move result */ async move(sourceKey, targetKey, options = {}) { // First copy the object const copyResult = await this.copy(sourceKey, targetKey, options); // Then delete the original await this.delete(sourceKey); return copyResult; } /** * Get a pre-signed URL for downloading or uploading * * @param {string} operation - Operation type ('getObject' or 'putObject') * @param {string} s3Key - S3 object key * @param {Object} [options={}] - Additional options * @param {number} [options.expiresIn=3600] - URL expiration in seconds * @param {Object} [options.params] - Additional parameters for the operation * @returns {Promise} - Pre-signed URL */ async getSignedUrl(operation, s3Key, options = {}) { const params = { Bucket: this.bucket, Key: s3Key, ...options.params }; let command; switch (operation.toLowerCase()) { case 'getobject': command = new GetObjectCommand(params); break; case 'putobject': command = new PutObjectCommand(params); break; default: throw new Error(`Unsupported operation: ${operation}`); } return await getSignedUrl(this.s3Client, command, { expiresIn: options.expiresIn || 3600 }); } /** * Get object metadata * * @param {string} s3Key - S3 object key * @returns {Promise} - Object metadata */ async getMetadata(s3Key) { return await this._retryOperation(() => this.s3Client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: s3Key })) ); } /** * Update object metadata * * @param {string} s3Key - S3 object key * @param {Object} metadata - New metadata * @returns {Promise} - Update result */ async updateMetadata(s3Key, metadata) { // S3 requires copying the object to itself to update metadata return await this.copy(s3Key, s3Key, { metadata }); } /** * Manual multipart upload for advanced use cases * * @param {string} localPath - Local file path * @param {string} s3Key - S3 object key * @param {number} fileSize - File size in bytes * @param {Object} options - Upload options * @returns {Promise} - Upload result * @private */ async _manualMultipartUpload(localPath, s3Key, fileSize, options) { logger.info(`Starting manual multipart upload for ${s3Key} (${fileSize} bytes)`); // Initiate multipart upload const multipartParams = { Bucket: this.bucket, Key: s3Key, ContentType: options.contentType || 'application/octet-stream', Metadata: options.metadata || {}, CacheControl: options.cacheControl }; // Remove undefined values Object.keys(multipartParams).forEach(key => multipartParams[key] === undefined && delete multipartParams[key]); const multipart = await this._retryOperation(() => this.s3Client.send(new CreateMultipartUploadCommand(multipartParams)) ); const uploadId = multipart.UploadId; const partSize = this.config.partSize; const numParts = Math.ceil(fileSize / partSize); let uploaded = 0; const parts = []; try { // Upload parts for (let partNum = 1; partNum <= numParts; partNum++) { const start = (partNum - 1) * partSize; const end = Math.min(start + partSize, fileSize); const partStream = fs.createReadStream(localPath, { start, end: end - 1 }); const partParams = { Bucket: this.bucket, Key: s3Key, PartNumber: partNum, UploadId: uploadId, Body: partStream }; // Upload part with retry const partResult = await this._retryOperation(() => this.s3Client.send(new UploadPartCommand(partParams)) ); parts.push({ ETag: partResult.ETag, PartNumber: partNum }); uploaded += (end - start); if (options.onProgress) { options.onProgress(uploaded, fileSize); } this.emit('uploadProgress', { key: s3Key, loaded: uploaded, total: fileSize }); logger.info(`Uploaded part ${partNum}/${numParts} for ${s3Key}`); } // Complete multipart upload const completeParams = { Bucket: this.bucket, Key: s3Key, UploadId: uploadId, MultipartUpload: { Parts: parts } }; const result = await this._retryOperation(() => this.s3Client.send(new CompleteMultipartUploadCommand(completeParams)) ); this.emit('uploadComplete', { key: s3Key, location: result.Location }); logger.info(`Completed multipart upload for ${s3Key}`); return result; } catch (error) { // Abort multipart upload on error logger.error(`Multipart upload failed for ${s3Key}, aborting:`, error); try { await this.s3Client.send(new AbortMultipartUploadCommand({ Bucket: this.bucket, Key: s3Key, UploadId: uploadId })); } catch (abortError) { logger.error(`Failed to abort multipart upload:`, abortError); } throw error; } } /** * Retry operation with exponential backoff * @private */ async _retryOperation(operation, retryCount = 0) { try { return await operation(); } catch (error) { if (retryCount >= this.config.maxRetries) { throw error; } // Check if error is retryable const retryableErrors = ['ECONNRESET', 'ETIMEDOUT', 'ENOTFOUND', 'ESOCKETTIMEDOUT', 'RequestTimeout', 'SlowDown', 'ServiceUnavailable', 'InternalError']; const isRetryable = retryableErrors.some(code => error.code === code || error.name === code || error.$metadata?.httpStatusCode === 503 || error.$metadata?.httpStatusCode === 500 ); if (!isRetryable) { throw error; } // Calculate delay with exponential backoff and jitter const delay = Math.min( this.config.retryDelay * Math.pow(2, retryCount) + Math.random() * 1000, 30000 // Max 30 seconds ); logger.warn(`Retrying operation after ${delay}ms (attempt ${retryCount + 1}/${this.config.maxRetries}):`, error.message); await new Promise(resolve => setTimeout(resolve, delay)); return this._retryOperation(operation, retryCount + 1); } } /** * Generate a unique S3 key for a file * * @param {string} originalName - Original filename * @param {string} [prefix=''] - Optional prefix for the key * @returns {string} - Generated S3 key */ generateKey(originalName, prefix = '') { const timestamp = Date.now(); const randomStr = crypto.randomBytes(8).toString('hex'); const ext = path.extname(originalName); const basename = path.basename(originalName, ext); // Sanitize basename const safeName = basename.replace(/[^a-zA-Z0-9-_]/g, '_'); const key = `${timestamp}_${randomStr}_${safeName}${ext}`; return prefix ? path.posix.join(prefix, key) : key; } /** * Get storage statistics * * @param {string} [prefix=''] - Optional prefix to filter * @returns {Promise} - Storage statistics */ async getStats(prefix = '') { let totalSize = 0; let totalCount = 0; let continuationToken; do { const result = await this.list(prefix, { continuationToken }); if (result.Contents) { totalCount += result.Contents.length; totalSize += result.Contents.reduce((sum, obj) => sum + (obj.Size || 0), 0); } continuationToken = result.NextContinuationToken; } while (continuationToken); return { totalSize, totalCount, totalSizeFormatted: this._formatBytes(totalSize) }; } /** * Format bytes to human readable format * @private */ _formatBytes(bytes, decimals = 2) { if (bytes === 0) return '0 Bytes'; const k = 1024; const dm = decimals < 0 ? 0 : decimals; const sizes = ['Bytes', 'KB', 'MB', 'GB', 'TB']; const i = Math.floor(Math.log(bytes) / Math.log(k)); return parseFloat((bytes / Math.pow(k, i)).toFixed(dm)) + ' ' + sizes[i]; } } module.exports = S3StorageAdapter;