feat(storage): add deleteSourceObjects option to combine/compose method

handwritten/storage/package.json

@@ -71,7 +71,7 @@
     "samples-test": "npm link && cd samples/ && npm link ../ && npm test && cd ../",
     "system-test:esm": "mocha build/esm/system-test --timeout 600000 --exit",
     "system-test": "mocha build/cjs/system-test --timeout 600000 --exit",
-    "test": "cross-env NODE_OPTIONS='--no-deprecation' c8 mocha build/cjs/test"
+    "test": "cross-env NODE_OPTIONS=\"--require ./scripts/preload-yargs.cjs --no-deprecation\" c8 mocha build/cjs/test"
   },
   "dependencies": {
     "@google-cloud/paginator": "^5.0.0",
@@ -106,7 +106,7 @@
     "@types/request": "^2.48.4",
     "@types/sinon": "^17.0.0",
     "@types/tmp": "0.2.6",
-    "@types/yargs": "^17.0.10",
+    "@types/yargs": "^17.0.35",
     "c8": "^9.0.0",
     "form-data": "^4.0.4",
     "gapic-tools": "^0.4.0",
@@ -125,7 +125,7 @@
     "path-to-regexp": "6.3.0",
     "tmp": "^0.2.0",
     "typescript": "^5.1.6",
-    "yargs": "^17.3.1",
+    "yargs": "^17.7.2",
     "cross-env": "^7.0.3"
   },
   "homepage": "https://github.com/googleapis/google-cloud-node/tree/main/handwritten/storage"

handwritten/storage/scripts/preload-yargs.cjs

@@ -0,0 +1,39 @@
+const Module = require('module');
+const fs = require('fs');
+const path = require('path');
+
+const originalResolveFilename = Module._resolveFilename;
+
+Module._resolveFilename = function(request, parent, isMain, options) {
+  if (request === 'yargs/yargs') {
+    const resolved = originalResolveFilename.apply(this, arguments);
+    if (resolved.endsWith('.mjs')) {
+      return resolved;
+    }
+    // Create a unique shim file in the local scripts directory to avoid shared temp directory vulnerabilities
+    const safeHash = Buffer.from(resolved).toString('base64').replace(/[^a-zA-Z0-9]/g, '');
+    const shimPath = path.join(__dirname, `yargs-shim-${safeHash}.cjs`);
+
+    if (!fs.existsSync(shimPath)) {
+      const content = fs.readFileSync(resolved, 'utf8');
+      // Replace `./build/index.cjs` with the absolute path
+      const buildIndexPath = path.join(path.dirname(resolved), 'build', 'index.cjs');
+      // We must replace ALL relative requires. Luckily yargs/yargs only requires `./build/index.cjs`
+      const newContent = content.replace(/require\(['"]\.\/build\/index\.cjs['"]\)/g, `require(${JSON.stringify(buildIndexPath)})`);
+      fs.writeFileSync(shimPath, newContent);
+    }
+
+    global.__yargsShimCleanups = global.__yargsShimCleanups || new Set();
+    if (!global.__yargsShimCleanups.has(shimPath)) {
+      global.__yargsShimCleanups.add(shimPath);
+      process.on('exit', () => {
+        try {
+          fs.unlinkSync(shimPath);
+        } catch (e) {}
+      });
+    }
+
+    return shimPath;
+  }
+  return originalResolveFilename.apply(this, arguments);
+};

handwritten/storage/src/bucket.ts

@@ -67,6 +67,7 @@ import {CRC32CValidatorGenerator} from './crc32c.js';
 import {URL} from 'url';
 import {
   BaseMetadata,
+  DeleteOptions,
   SetMetadataOptions,
 } from './nodejs-common/service-object.js';
 
@@ -190,6 +191,7 @@ export interface CombineOptions extends PreconditionOptions {
       [key: string]: ContextValue;
     } | null;
   };
+  deleteSourceObjects?: boolean;
 }
 
 export interface CombineCallback {
@@ -198,6 +200,24 @@ export interface CombineCallback {
 
 export type CombineResponse = [File, unknown];
 
+export class ComposeCleanupError extends Error {
+  errors: Error[];
+  newFile: File;
+  apiResponse: unknown;
+  constructor(
+    message: string,
+    errors: Error[],
+    newFile: File,
+    apiResponse: unknown
+  ) {
+    super(message);
+    this.name = 'ComposeCleanupError';
+    this.errors = errors;
+    this.newFile = newFile;
+    this.apiResponse = apiResponse;
+  }
+}
+
 export interface CreateChannelConfig extends WatchAllOptions {
   address: string;
 }
@@ -1579,7 +1599,9 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
    * metadata's `kms_key_name` value, if any.
    * @property {string} [userProject] The ID of the project which will be
    *     billed for the request.
-   */
+    * @property {boolean} [deleteSourceObjects] If true, the source objects
+    *     will be permanently deleted after a successful compose operation.
+    */
   /**
    * @callback CombineCallback
    * @param {?Error} err Request error, if any.
@@ -1612,7 +1634,8 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
    * metadata's `kms_key_name` value, if any.
    * @param {string} [options.userProject] The ID of the project which will be
    *     billed for the request.
-
+   * @param {boolean} [options.deleteSourceObjects] If true, the source objects
+   *     will be permanently deleted after a successful compose operation.
    * @param {CombineCallback} [callback] Callback function.
    * @returns {Promise<CombineResponse>}
    *
@@ -1709,8 +1732,17 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
       maxRetries = 0;
     }
 
-    if (options.ifGenerationMatch === undefined) {
-      Object.assign(options, destinationFile.instancePreconditionOpts, options);
+    const deleteSourceObjects = options.deleteSourceObjects;
+
+    const requestQueryObject = Object.assign({}, options);
+    delete requestQueryObject.deleteSourceObjects;
+
+    if (requestQueryObject.ifGenerationMatch === undefined) {
+      Object.assign(
+        requestQueryObject,
+        destinationFile.instancePreconditionOpts,
+        requestQueryObject
+      );
     }
 
     // Make the request from the destination File object.
@@ -1723,23 +1755,23 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
           destination: {
             contentType: destinationFile.metadata.contentType,
             contentEncoding: destinationFile.metadata.contentEncoding,
-            contexts: options.contexts || destinationFile.metadata.contexts,
+            contexts:
+              requestQueryObject.contexts || destinationFile.metadata.contexts,
           },
           sourceObjects: (sources as File[]).map(source => {
             const sourceObject = {
               name: source.name,
             } as SourceObject;
 
-            if (source.metadata && source.metadata.generation) {
-              sourceObject.generation = parseInt(
-                source.metadata.generation.toString(),
-              );
+            const generation = source.generation ?? source.metadata?.generation;
+            if (generation !== undefined) {
+              sourceObject.generation = parseInt(generation.toString());
             }
 
             return sourceObject;
           }),
         },
-        qs: options,
+        qs: requestQueryObject,
       },
       (err, resp) => {
         this.storage.retryOptions.autoRetry = this.instanceRetryValue;
@@ -1748,8 +1780,45 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
           return;
         }
 
-        callback!(null, destinationFile, resp);
-      },
+        if (deleteSourceObjects) {
+          const deletePromises = (sources as File[]).map(source => {
+            const deleteOptions: DeleteOptions = {
+              ignoreNotFound: true,
+              userProject: options.userProject,
+            };
+
+            const generation = source.generation ?? source.metadata?.generation;
+            if (generation !== undefined) {
+              deleteOptions.ifGenerationMatch = generation;
+            }
+
+            return source
+              .delete(deleteOptions)
+              .catch(deleteErr => deleteErr as Error);
+          });
+
+          Promise.all(deletePromises).then(results => {
+            const errors = results.filter(
+              (res): res is Error => res instanceof Error
+            );
+
+            if (errors.length > 0) {
+              const cleanupErr = new ComposeCleanupError(
+                `Compose operation succeeded, but cleaning up source objects failed. Failed to delete ${errors.length} source object(s).`,
+                errors,
+                destinationFile,
+                resp
+              );
+              callback!(cleanupErr, destinationFile, resp);
+              return;
+            }
+
+            callback!(null, destinationFile, resp);
+          });
+        } else {
+          callback!(null, destinationFile, resp);
+        }
+      }
     );
   }
 

handwritten/storage/src/file.ts

@@ -2187,6 +2187,29 @@ class File extends ServiceObject<File, FileMetadata> {
       // remove temporary noop listener as we now create a pipeline that handles the errors
       emitStream.removeListener('error', noop);
 
+      if (fileWriteStream.destroyed) {
+        let callbackCalled = false;
+        const onError = (err: Error) => {
+          if (!callbackCalled) {
+            callbackCalled = true;
+            pipelineCallback(err);
+          }
+        };
+        fileWriteStream.once('error', onError);
+        emitStream.destroy();
+
+        process.nextTick(() => {
+          fileWriteStream.removeListener('error', onError);
+          if (!callbackCalled) {
+            callbackCalled = true;
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            const err = (fileWriteStream as any).errored || new Error('Write stream destroyed');
+            pipelineCallback(err);
+          }
+        });
+        return;
+      }
+
       pipeline(
         emitStream,
         ...(transformStreams as [Transform]),

handwritten/storage/src/index.ts

@@ -111,6 +111,7 @@ export {
   CombineCallback,
   CombineOptions,
   CombineResponse,
+  ComposeCleanupError,
   CreateChannelCallback,
   CreateChannelConfig,
   CreateChannelOptions,

handwritten/storage/src/nodejs-common/service-object.ts

@@ -111,6 +111,7 @@ export interface CreateCallback<T> {
 
 export type DeleteOptions = {
   ignoreNotFound?: boolean;
+  userProject?: string;
   ifGenerationMatch?: number | string;
   ifGenerationNotMatch?: number | string;
   ifMetagenerationMatch?: number | string;