feat(functions): Reroute Cloud Tasks to emulator when it is running

src/functions/functions-api-client-internal.ts

@@ -29,6 +29,7 @@ import { ComputeEngineCredential } from '../app/credential-internal';
 const CLOUD_TASKS_API_RESOURCE_PATH = 'projects/{projectId}/locations/{locationId}/queues/{resourceId}/tasks';
 const CLOUD_TASKS_API_URL_FORMAT = 'https://cloudtasks.googleapis.com/v2/' + CLOUD_TASKS_API_RESOURCE_PATH;
 const FIREBASE_FUNCTION_URL_FORMAT = 'https://{locationId}-{projectId}.cloudfunctions.net/{resourceId}';
+export const EMULATED_SERVICE_ACCOUNT_DEFAULT = 'emulated-service-acct@email.com';
 
 const FIREBASE_FUNCTIONS_CONFIG_HEADERS = {
   'X-Firebase-Client': `fire-admin-node/${utils.getSdkVersion()}`
@@ -69,8 +70,8 @@ export class FunctionsApiClient {
     }
     if (!validator.isTaskId(id)) {
       throw new FirebaseFunctionsError(
-        'invalid-argument', 'id can contain only letters ([A-Za-z]), numbers ([0-9]), ' 
-        + 'hyphens (-), or underscores (_). The maximum length is 500 characters.');
+        'invalid-argument', 'id can contain only letters ([A-Za-z]), numbers ([0-9]), '
+      + 'hyphens (-), or underscores (_). The maximum length is 500 characters.');
     }
 
     let resources: utils.ParsedResource;
@@ -91,7 +92,8 @@ export class FunctionsApiClient {
     }
 
     try {
-      const serviceUrl = await this.getUrl(resources, CLOUD_TASKS_API_URL_FORMAT.concat('/', id));
+      const serviceUrl = tasksEmulatorUrl(resources, functionName)?.concat('/', id)
+        ?? await this.getUrl(resources, CLOUD_TASKS_API_URL_FORMAT.concat('/', id));
       const request: HttpRequestConfig = {
         method: 'DELETE',
         url: serviceUrl,
@@ -144,7 +146,10 @@ export class FunctionsApiClient {
 
     const task = this.validateTaskOptions(data, resources, opts);
     try {
-      const serviceUrl = await this.getUrl(resources, CLOUD_TASKS_API_URL_FORMAT);
+      const serviceUrl =
+        tasksEmulatorUrl(resources, functionName) ??
+        await this.getUrl(resources, CLOUD_TASKS_API_URL_FORMAT);
+
       const taskPayload = await this.updateTaskPayload(task, resources, extensionId);
       const request: HttpRequestConfig = {
         method: 'POST',
@@ -237,7 +242,7 @@ export class FunctionsApiClient {
           serviceAccountEmail: '',
         },
         body: Buffer.from(JSON.stringify({ data })).toString('base64'),
-        headers: { 
+        headers: {
           'Content-Type': 'application/json',
           ...opts?.headers,
         }
@@ -252,7 +257,7 @@ export class FunctionsApiClient {
       if ('scheduleTime' in opts && 'scheduleDelaySeconds' in opts) {
         throw new FirebaseFunctionsError(
           'invalid-argument', 'Both scheduleTime and scheduleDelaySeconds are provided. '
-          + 'Only one value should be set.');
+        + 'Only one value should be set.');
       }
       if ('scheduleTime' in opts && typeof opts.scheduleTime !== 'undefined') {
         if (!(opts.scheduleTime instanceof Date)) {
@@ -275,15 +280,15 @@ export class FunctionsApiClient {
           || opts.dispatchDeadlineSeconds > 1800) {
           throw new FirebaseFunctionsError(
             'invalid-argument', 'dispatchDeadlineSeconds must be a non-negative duration in seconds '
-            + 'and must be in the range of 15s to 30 mins.');
+          + 'and must be in the range of 15s to 30 mins.');
         }
         task.dispatchDeadline = `${opts.dispatchDeadlineSeconds}s`;
       }
       if ('id' in opts && typeof opts.id !== 'undefined') {
         if (!validator.isTaskId(opts.id)) {
           throw new FirebaseFunctionsError(
             'invalid-argument', 'id can contain only letters ([A-Za-z]), numbers ([0-9]), '
-            + 'hyphens (-), or underscores (_). The maximum length is 500 characters.');
+          + 'hyphens (-), or underscores (_). The maximum length is 500 characters.');
         }
         const resourcePath = utils.formatString(CLOUD_TASKS_API_RESOURCE_PATH, {
           projectId: resources.projectId,
@@ -304,9 +309,14 @@ export class FunctionsApiClient {
   }
 
   private async updateTaskPayload(task: Task, resources: utils.ParsedResource, extensionId?: string): Promise<Task> {
-    const functionUrl = validator.isNonEmptyString(task.httpRequest.url)
-      ? task.httpRequest.url 
+    const defaultUrl = process.env.CLOUD_TASKS_EMULATOR_HOST ?
+      ''
       : await this.getUrl(resources, FIREBASE_FUNCTION_URL_FORMAT);
+
+    const functionUrl = validator.isNonEmptyString(task.httpRequest.url)
+      ? task.httpRequest.url
+      : defaultUrl;
+
     task.httpRequest.url = functionUrl;
     // When run from a deployed extension, we should be using ComputeEngineCredentials
     if (validator.isNonEmptyString(extensionId) && this.app.options.credential instanceof ComputeEngineCredential) {
@@ -315,8 +325,16 @@ export class FunctionsApiClient {
       // Don't send httpRequest.oidcToken if we set Authorization header, or Cloud Tasks will overwrite it.
       delete task.httpRequest.oidcToken;
     } else {
-      const account =  await this.getServiceAccount();
-      task.httpRequest.oidcToken = { serviceAccountEmail: account };
+      try {
+        const account = await this.getServiceAccount();
+        task.httpRequest.oidcToken = { serviceAccountEmail: account };
+      } catch (e) {
+        if (process.env.CLOUD_TASKS_EMULATOR_HOST) {
+          task.httpRequest.oidcToken = { serviceAccountEmail: EMULATED_SERVICE_ACCOUNT_DEFAULT };
+        } else {
+          throw e;
+        }
+      }
     }
     return task;
   }
@@ -417,3 +435,10 @@ export class FirebaseFunctionsError extends PrefixedFirebaseError {
     (this as any).__proto__ = FirebaseFunctionsError.prototype;
   }
 }
+
+function tasksEmulatorUrl(resources: utils.ParsedResource, functionName: string): string | undefined {
+  if (process.env.CLOUD_TASKS_EMULATOR_HOST) {
+    return `http://${process.env.CLOUD_TASKS_EMULATOR_HOST}/projects/${resources.projectId}/locations/${resources.locationId}/queues/${functionName}/tasks`;
+  }
+  return undefined;
+}

test/unit/functions/functions-api-client-internal.spec.ts

@@ -29,6 +29,7 @@ import { FirebaseFunctionsError, FunctionsApiClient, Task } from '../../../src/f
 import { HttpClient } from '../../../src/utils/api-request';
 import { FirebaseAppError } from '../../../src/utils/error';
 import { deepCopy } from '../../../src/utils/deep-copy';
+import { EMULATED_SERVICE_ACCOUNT_DEFAULT } from '../../../src/functions/functions-api-client-internal';
 
 const expect = chai.expect;
 
@@ -90,6 +91,10 @@ describe('FunctionsApiClient', () => {
   const CLOUD_TASKS_URL_FULL_RESOURCE = `https://cloudtasks.googleapis.com/v2/projects/${CUSTOM_PROJECT_ID}/locations/${CUSTOM_REGION}/queues/${FUNCTION_NAME}/tasks`;
 
   const CLOUD_TASKS_URL_PARTIAL_RESOURCE = `https://cloudtasks.googleapis.com/v2/projects/${mockOptions.projectId}/locations/${CUSTOM_REGION}/queues/${FUNCTION_NAME}/tasks`;
+
+  const CLOUD_TASKS_EMULATOR_HOST = '127.0.0.1:9499';
+
+  const CLOUD_TASKS_URL_EMULATOR = `http://${CLOUD_TASKS_EMULATOR_HOST}/projects/${mockOptions.projectId}/locations/${DEFAULT_REGION}/queues/${FUNCTION_NAME}/tasks`;
 
   const clientWithoutProjectId = new FunctionsApiClient(mocks.mockCredentialApp());
 
@@ -106,6 +111,9 @@ describe('FunctionsApiClient', () => {
   afterEach(() => {
     _.forEach(stubs, (stub) => stub.restore());
     stubs = [];
+    if (process.env.CLOUD_TASKS_EMULATOR_HOST) {
+      delete process.env.CLOUD_TASKS_EMULATOR_HOST;
+    }
     return app.delete();
   });
 
@@ -477,8 +485,79 @@ describe('FunctionsApiClient', () => {
           });
         });
     });
+
+    it('should redirect to the emulator when CLOUD_TASKS_EMULATOR_HOST is set', () => {
+      const expectedPayload = deepCopy(TEST_TASK_PAYLOAD);
+      const stub = sinon
+        .stub(HttpClient.prototype, 'send')
+        .resolves(utils.responseFrom({}, 200));
+      stubs.push(stub);
+      process.env.CLOUD_TASKS_EMULATOR_HOST = CLOUD_TASKS_EMULATOR_HOST;
+      return apiClient.enqueue({}, FUNCTION_NAME, '', { uri: TEST_TASK_PAYLOAD.httpRequest.url })
+        .then(() => {
+          expect(stub).to.have.been.calledOnce.and.calledWith({
+            method: 'POST',
+            url: CLOUD_TASKS_URL_EMULATOR,
+            headers: EXPECTED_HEADERS,
+            data: {
+              task: expectedPayload
+            }
+          });
+        });
+    });
+
+    it('should leave empty urls alone when CLOUD_TASKS_EMULATOR_HOST is set', () => {
+      const expectedPayload = deepCopy(TEST_TASK_PAYLOAD);
+      expectedPayload.httpRequest.url = '';
+      const stub = sinon
+        .stub(HttpClient.prototype, 'send')
+        .resolves(utils.responseFrom({}, 200));
+      stubs.push(stub);
+      process.env.CLOUD_TASKS_EMULATOR_HOST = CLOUD_TASKS_EMULATOR_HOST;
+      return apiClient.enqueue({}, FUNCTION_NAME)
+        .then(() => {
+          expect(stub).to.have.been.calledOnce.and.calledWith({
+            method: 'POST',
+            url: CLOUD_TASKS_URL_EMULATOR,
+            headers: EXPECTED_HEADERS,
+            data: {
+              task: expectedPayload
+            }
+          });
+        });
+    });
+
+    it('should use a fake service account if the emulator is running and no service account is defined', () => {
+      app = mocks.appWithOptions({
+        credential: new mocks.MockCredential(),
+        projectId: 'test-project',
+        serviceAccountId: ''
+      });
+      apiClient = new FunctionsApiClient(app);
+
+      const expectedPayload = deepCopy(TEST_TASK_PAYLOAD);
+      expectedPayload.httpRequest.oidcToken = { serviceAccountEmail: EMULATED_SERVICE_ACCOUNT_DEFAULT };
+      const stub = sinon
+        .stub(HttpClient.prototype, 'send')
+        .resolves(utils.responseFrom({}, 200));
+      stubs.push(stub);
+      process.env.CLOUD_TASKS_EMULATOR_HOST = CLOUD_TASKS_EMULATOR_HOST;
+      return apiClient.enqueue({}, FUNCTION_NAME, '', { uri: TEST_TASK_PAYLOAD.httpRequest.url })
+        .then(() => {
+          expect(stub).to.have.been.calledOnce.and.calledWith({
+            method: 'POST',
+            url: CLOUD_TASKS_URL_EMULATOR,
+            headers: EXPECTED_HEADERS,
+            data: {
+              task: expectedPayload
+            }
+          });
+        });
+    })
+
   });
 
+
   describe('delete', () => {
     for (const invalidTaskId of [1234, 'task!', 'id:0', '[1234]', '(1234)']) {
       it(`should throw given an invalid task ID: ${invalidTaskId}`, () => {
@@ -514,6 +593,20 @@ describe('FunctionsApiClient', () => {
       expect(apiClient.delete('nonexistent-task', FUNCTION_NAME)).to.eventually.not.throw(utils.errorFrom({}, 404));
     });
 
+    it('should redirect to the emulator when CLOUD_TASKS_EMULATOR_HOST is set', async () => {
+      process.env.CLOUD_TASKS_EMULATOR_HOST = CLOUD_TASKS_EMULATOR_HOST;
+      const stub = sinon
+        .stub(HttpClient.prototype, 'send')
+        .resolves(utils.responseFrom({}, 200));
+      stubs.push(stub);
+      await apiClient.delete('mock-task', FUNCTION_NAME);
+      expect(stub).to.have.been.calledWith({
+        method: 'DELETE',
+        url: CLOUD_TASKS_URL_EMULATOR.concat('/', 'mock-task'),
+        headers: EXPECTED_HEADERS,
+      });
+    });
+
     it('should throw on non-404 HTTP errors', () => {
       const stub = sinon
         .stub(HttpClient.prototype, 'send')