Enhance memory management and error handling in attachment processing

lib/aws-s3.js

@@ -255,6 +255,7 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
       })
 
       const stored = super.put(attachments, metadata, null, isDraftEnabled)
+
       await Promise.all([stored, multipartUpload.done()])
 
       const duration = Date.now() - startTime
@@ -335,7 +336,7 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
 
       const Key = response.url
 
-      logConfig.debug('Downloading file from S3', {
+      logConfig.debug('Streaming file from S3', {
         bucket: this.bucket,
         key: Key
       })
@@ -348,7 +349,7 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
       )
 
       const duration = Date.now() - startTime
-      logConfig.debug('File download from S3 completed successfully', {
+      logConfig.debug('File streamed from S3 successfully', {
         fileId: keys.ID,
         bucket: this.bucket,
         key: Key,
@@ -454,7 +455,7 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
           client: this.client,
           params: input,
         })
-        await Promise.all([multipartUpload.done()])
+        await multipartUpload.done()
 
         const keys = { ID: targetID }
 

lib/basic.js

@@ -139,7 +139,7 @@ module.exports = class AttachmentsService extends cds.Service {
    * Handles non-draft attachment updates by uploading content to the database
    * @param {Express.Request} req - The request object
    * @param {cds.Entity} attachment - Attachments entity definition
-   * @returns 
+   * @returns {Promise} - Result of the upsert operation
    */
   async nonDraftHandler(req, attachment) {
     if (req?.content?.url?.endsWith("/content")) {
@@ -195,7 +195,7 @@ module.exports = class AttachmentsService extends cds.Service {
    * @param {cds.Entity} Attachments - Attachments entity definition
    * @param {string} key - The key of the attachment to update
    * @param {*} data - The data to update the attachment with
-   * @returns 
+   * @returns {Promise} - Result of the update operation
    */
   async update(Attachments, key, data) {
     logConfig.debug("Updating attachment for", {

lib/malwareScanner.js

@@ -80,14 +80,10 @@ async function scanRequest(Attachments, key, req) {
   logConfig.debug('Fetching file content for scanning', { fileId: key.ID })
   const contentStream = await AttachmentsSrv.get(currEntity, key)
 
-  let fileContent
-  try {
-    logConfig.verbose('Converting file stream to string for scanning', { fileId: key.ID })
-    fileContent = await streamToString(contentStream)
-  } catch (err) {
+  if (!contentStream) {
     logConfig.withSuggestion('error',
-      'Cannot read file content for malware scanning', err,
-      'Check file integrity and storage accessibility',
+      'Cannot fetch file content for malware scanning', null,
+      'Check file exists and storage accessibility',
       { fileId: key.ID })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
     return
@@ -97,19 +93,20 @@ async function scanRequest(Attachments, key, req) {
   const scanStartTime = Date.now()
 
   try {
-    logConfig.debug('Sending file to Malware Scanning Service', {
+    logConfig.debug('Streaming file to Malware Scanning Service', {
       fileId: key.ID,
-      scannerUri: credentials.uri,
-      contentLength: fileContent.length
+      scannerUri: credentials.uri
     })
 
+    // Stream the file directly to the scanner without loading into memory
     response = await fetch(`https://${credentials.uri}/scan`, {
       method: "POST",
       headers: {
         Authorization:
           "Basic " + Buffer.from(`${credentials.username}:${credentials.password}`, "binary").toString("base64"),
       },
-      body: fileContent,
+      body: contentStream,
+      duplex: 'half' // Required for streaming request bodies
     })
 
   } catch (error) {
@@ -119,7 +116,13 @@ async function scanRequest(Attachments, key, req) {
       'Check malware scanner service binding and network connectivity',
       { fileId: key.ID, scanDuration, scannerUri: credentials?.uri })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
+
+    // Cleanup: destroy stream on error
+    contentStream?.destroy()
     return
+  } finally {
+    // Cleanup: destroy stream after request completion
+    contentStream?.destroy()
   }
 
   try {
@@ -151,6 +154,8 @@ async function scanRequest(Attachments, key, req) {
       'Check malware scanner service health and response format',
       { fileId: key.ID, scanDuration, httpStatus: response?.status })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
+  } finally {
+    contentStream?.destroy()
   }
 }
 
@@ -206,15 +211,6 @@ function getCredentials() {
   }
 }
 
-function streamToString(stream) {
-  const chunks = []
-  return new Promise((resolve, reject) => {
-    stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)))
-    stream.on('error', (err) => reject(err))
-    stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')))
-  })
-}
-
 module.exports = {
   scanRequest
 }

lib/plugin.js

@@ -118,13 +118,13 @@ cds.once("served", async function registerPluginHandlers() {
    * Reads the attachment content if requested
    * @param {[cds.Entity]} param0
    * @param {import('@sap/cds').Request} req - The request object
-   * @returns 
+   * @returns
    */
   async function readAttachment([attachment], req) {
     if (req._.readAfterWrite || !req?.req?.url?.endsWith("/content") || !attachment || attachment?.content) return
     let keys = { ID: req.req.url.match(attachmentIDRegex)[1] }
     let { target } = req
-    attachment.content = await AttachmentsSrv.get(target, keys) //Dependency -> sending req object for usage in SDM plugin
+    attachment.content = await AttachmentsSrv.get(target, keys)
   }
 })