Add blocking trigger type

CHANGELOG.md

@@ -3,3 +3,4 @@
 - Fixes Firestore emulator deadlock bug. [#2452](https://github.com/firebase/firebase-tools/issues/2452)
 - Fixes console error on large uploads to Storage Emulator (#4407).
 - Fixes cross-platform incompatibility with Storage Emulator exports (#4411).
+- Adds a blocking trigger type (#4395).

src/deploy/functions/backend.ts

@@ -3,6 +3,7 @@ import * as gcf from "../../gcp/cloudfunctions";
 import * as gcfV2 from "../../gcp/cloudfunctionsv2";
 import * as utils from "../../utils";
 import * as runtimes from "./runtimes";
+import * as events from "../../functions/events";
 import { FirebaseError } from "../../error";
 import { Context } from "./args";
 import { previews } from "../../previews";
@@ -131,6 +132,15 @@ export interface TaskQueueTriggered {
   taskQueueTrigger: TaskQueueTrigger;
 }
 
+export interface BlockingTrigger {
+  eventType: string;
+  options?: Record<string, any>;
+}
+
+export interface BlockingTriggered {
+  blockingTrigger: BlockingTrigger;
+}
+
 /** A user-friendly string for the kind of trigger of an endpoint. */
 export function endpointTriggerType(endpoint: Endpoint): string {
   if (isScheduleTriggered(endpoint)) {
@@ -143,6 +153,8 @@ export function endpointTriggerType(endpoint: Endpoint): string {
     return endpoint.eventTrigger.eventType;
   } else if (isTaskQueueTriggered(endpoint)) {
     return "taskQueue";
+  } else if (isBlockingTriggered(endpoint)) {
+    return endpoint.blockingTrigger.eventType;
   } else {
     throw new Error("Unexpected trigger type for endpoint " + JSON.stringify(endpoint));
   }
@@ -227,7 +239,8 @@ export type Triggered =
   | CallableTriggered
   | EventTriggered
   | ScheduleTriggered
-  | TaskQueueTriggered;
+  | TaskQueueTriggered
+  | BlockingTriggered;
 
 /** Whether something has an HttpsTrigger */
 export function isHttpsTriggered(triggered: Triggered): triggered is HttpsTriggered {
@@ -254,6 +267,11 @@ export function isTaskQueueTriggered(triggered: Triggered): triggered is TaskQue
   return {}.hasOwnProperty.call(triggered, "taskQueueTrigger");
 }
 
+/** Whether something has a BlockingTrigger */
+export function isBlockingTriggered(triggered: Triggered): triggered is BlockingTriggered {
+  return {}.hasOwnProperty.call(triggered, "blockingTrigger");
+}
+
 /**
  * An endpoint that serves traffic to a stack of services.
  * For now, this is always a Cloud Function. Future iterations may use complex

src/deploy/functions/prepare.ts

@@ -21,6 +21,7 @@ import { ensureServiceAgentRoles } from "./checkIam";
 import { FirebaseError } from "../../error";
 import { normalizeAndValidate } from "../../functions/projectConfig";
 import { previews } from "../../previews";
+import { AUTH_BLOCKING_EVENTS } from "../../functions/events/v1";
 
 function hasUserConfig(config: Record<string, unknown>): boolean {
   // "firebase" key is always going to exist in runtime config.
@@ -204,6 +205,7 @@ export async function prepare(
 
   await ensureServiceAgentRoles(projectNumber, wantBackend, haveBackend);
   inferDetailsFromExisting(wantBackend, haveBackend, usedDotenv);
+  inferBlockingDetails(wantBackend);
   await ensureTriggerRegions(wantBackend);
 
   // Display a warning and prompt if any functions in the release have failurePolicies.
@@ -271,3 +273,35 @@ function maybeCopyTriggerRegion(wantE: backend.Endpoint, haveE: backend.Endpoint
   }
   wantE.eventTrigger.region = haveE.eventTrigger.region;
 }
+
+/** Figures out the blocking endpoint options by taking the OR of every trigger option and reassigning that value back to the endpoint. */
+export function inferBlockingDetails(want: backend.Backend): void {
+  const authBlockingEndpoints = backend
+    .allEndpoints(want)
+    .filter(
+      (ep) =>
+        backend.isBlockingTriggered(ep) &&
+        AUTH_BLOCKING_EVENTS.includes(ep.blockingTrigger.eventType as any)
+    ) as (backend.Endpoint & backend.BlockingTriggered)[];
+
+  if (authBlockingEndpoints.length === 0) {
+    return;
+  }
+
+  let accessToken = false;
+  let idToken = false;
+  let refreshToken = false;
+  for (const blockingEp of authBlockingEndpoints) {
+    accessToken ||= !!blockingEp.blockingTrigger.options?.accessToken;
+    idToken ||= !!blockingEp.blockingTrigger.options?.idToken;
+    refreshToken ||= !!blockingEp.blockingTrigger.options?.refreshToken;
+  }
+  for (const blockingEp of authBlockingEndpoints) {
+    if (!blockingEp.blockingTrigger.options) {
+      blockingEp.blockingTrigger.options = {};
+    }
+    blockingEp.blockingTrigger.options.accessToken = accessToken;
+    blockingEp.blockingTrigger.options.idToken = idToken;
+    blockingEp.blockingTrigger.options.refreshToken = refreshToken;
+  }
+}

src/deploy/functions/release/fabricator.ts

@@ -21,6 +21,8 @@ import * as reporter from "./reporter";
 import * as run from "../../../gcp/run";
 import * as scheduler from "../../../gcp/cloudscheduler";
 import * as utils from "../../../utils";
+import * as services from "../services";
+import { AUTH_BLOCKING_EVENTS } from "../../../functions/events/v1";
 
 // TODO: Tune this for better performance.
 const gcfV1PollerOptions: Omit<poller.OperationPollerOptions, "operationResourceName"> = {
@@ -54,6 +56,7 @@ export interface FabricatorArgs {
 const rethrowAs =
   <T>(endpoint: backend.Endpoint, op: reporter.OperationType) =>
   (err: unknown): T => {
+    logger.error((err as Error).message);
     throw new reporter.DeploymentError(endpoint, op, err);
   };
 
@@ -64,13 +67,15 @@ export class Fabricator {
   sourceUrl: string | undefined;
   storage: Record<string, gcfV2.StorageSource> | undefined;
   appEngineLocation: string;
+  triggerQueue: Promise<void>;
 
   constructor(args: FabricatorArgs) {
     this.executor = args.executor;
     this.functionExecutor = args.functionExecutor;
     this.sourceUrl = args.sourceUrl;
     this.storage = args.storage;
     this.appEngineLocation = args.appEngineLocation;
+    this.triggerQueue = Promise.resolve();
   }
 
   async applyPlan(plan: planner.DeploymentPlan): Promise<reporter.Summary> {
@@ -164,7 +169,7 @@ export class Fabricator {
       assertExhaustive(endpoint.platform);
     }
 
-    await this.setTrigger(endpoint);
+    await this.setTrigger(endpoint, false);
   }
 
   async updateEndpoint(update: planner.EndpointUpdate, scraper: SourceTokenScraper): Promise<void> {
@@ -183,7 +188,7 @@ export class Fabricator {
       assertExhaustive(update.endpoint.platform);
     }
 
-    await this.setTrigger(update.endpoint);
+    await this.setTrigger(update.endpoint, true);
   }
 
   async deleteEndpoint(endpoint: backend.Endpoint): Promise<void> {
@@ -248,6 +253,16 @@ export class Fabricator {
           })
           .catch(rethrowAs(endpoint, "set invoker"));
       }
+    } else if (
+      backend.isBlockingTriggered(endpoint) &&
+      AUTH_BLOCKING_EVENTS.includes(endpoint.blockingTrigger.eventType as any)
+    ) {
+      // Auth Blocking functions should always be public
+      await this.executor
+        .run(async () => {
+          await gcf.setInvokerCreate(endpoint.project, backend.functionName(endpoint), ["public"]);
+        })
+        .catch(rethrowAs(endpoint, "set invoker"));
     }
   }
 
@@ -317,6 +332,14 @@ export class Fabricator {
           })
           .catch(rethrowAs(endpoint, "set invoker"));
       }
+    } else if (
+      backend.isBlockingTriggered(endpoint) &&
+      AUTH_BLOCKING_EVENTS.includes(endpoint.blockingTrigger.eventType as any)
+    ) {
+      // Auth Blocking functions should always be public
+      await this.executor
+        .run(() => run.setInvokerCreate(endpoint.project, serviceName, ["public"]))
+        .catch(rethrowAs(endpoint, "set invoker"));
     }
 
     const mem = endpoint.availableMemoryMb || backend.DEFAULT_MEMORY;
@@ -354,6 +377,11 @@ export class Fabricator {
       invoker = endpoint.httpsTrigger.invoker;
     } else if (backend.isTaskQueueTriggered(endpoint)) {
       invoker = endpoint.taskQueueTrigger.invoker;
+    } else if (
+      backend.isBlockingTriggered(endpoint) &&
+      AUTH_BLOCKING_EVENTS.includes(endpoint.blockingTrigger.eventType as any)
+    ) {
+      invoker = ["public"];
     }
     if (invoker) {
       await this.executor
@@ -395,6 +423,11 @@ export class Fabricator {
       invoker = endpoint.httpsTrigger.invoker;
     } else if (backend.isTaskQueueTriggered(endpoint)) {
       invoker = endpoint.taskQueueTrigger.invoker;
+    } else if (
+      backend.isBlockingTriggered(endpoint) &&
+      AUTH_BLOCKING_EVENTS.includes(endpoint.blockingTrigger.eventType as any)
+    ) {
+      invoker = ["public"];
     }
     if (invoker) {
       await this.executor
@@ -460,7 +493,7 @@ export class Fabricator {
 
   // Set/Delete trigger is responsible for wiring up a function with any trigger not owned
   // by the GCF API. This includes schedules, task queues, and blocking function triggers.
-  async setTrigger(endpoint: backend.Endpoint): Promise<void> {
+  async setTrigger(endpoint: backend.Endpoint, update: boolean): Promise<void> {
     if (backend.isScheduleTriggered(endpoint)) {
       if (endpoint.platform === "gcfv1") {
         await this.upsertScheduleV1(endpoint);
@@ -472,6 +505,8 @@ export class Fabricator {
       assertExhaustive(endpoint.platform);
     } else if (backend.isTaskQueueTriggered(endpoint)) {
       await this.upsertTaskQueue(endpoint);
+    } else if (backend.isBlockingTriggered(endpoint)) {
+      await this.registerBlockingTrigger(endpoint, update);
     }
   }
 
@@ -487,6 +522,8 @@ export class Fabricator {
       assertExhaustive(endpoint.platform);
     } else if (backend.isTaskQueueTriggered(endpoint)) {
       await this.disableTaskQueue(endpoint);
+    } else if (backend.isBlockingTriggered(endpoint)) {
+      await this.unregisterBlockingTrigger(endpoint);
     }
   }
 
@@ -519,6 +556,18 @@ export class Fabricator {
     }
   }
 
+  async registerBlockingTrigger(
+    endpoint: backend.Endpoint & backend.BlockingTriggered,
+    update: boolean
+  ): Promise<void> {
+    this.triggerQueue = this.triggerQueue.then(async () => {
+      await this.executor
+        .run(() => services.serviceForEndpoint(endpoint).registerTrigger(endpoint, update))
+        .catch(rethrowAs(endpoint, "register blocking trigger"));
+    });
+    return this.triggerQueue;
+  }
+
   async deleteScheduleV1(endpoint: backend.Endpoint & backend.ScheduleTriggered): Promise<void> {
     const job = scheduler.jobFromEndpoint(endpoint, this.appEngineLocation);
     await this.executor
@@ -546,6 +595,17 @@ export class Fabricator {
       .catch(rethrowAs(endpoint, "disable task queue"));
   }
 
+  async unregisterBlockingTrigger(
+    endpoint: backend.Endpoint & backend.BlockingTriggered
+  ): Promise<void> {
+    this.triggerQueue = this.triggerQueue.then(async () => {
+      await this.executor
+        .run(() => services.serviceForEndpoint(endpoint).unregisterTrigger(endpoint))
+        .catch(rethrowAs(endpoint, "unregister blocking trigger"));
+    });
+    return this.triggerQueue;
+  }
+
   logOpStart(op: string, endpoint: backend.Endpoint): void {
     const runtime = getHumanFriendlyRuntimeName(endpoint.runtime);
     const label = helper.getFunctionLabel(endpoint);

src/deploy/functions/release/planner.ts

@@ -235,6 +235,8 @@ export function checkForIllegalUpdate(want: backend.Endpoint, have: backend.Endp
       return "a scheduled";
     } else if (backend.isTaskQueueTriggered(e)) {
       return "a task queue";
+    } else if (backend.isBlockingTriggered(e)) {
+      return e.blockingTrigger.eventType;
     }
     // Unfortunately TypeScript isn't like Scala and I can't prove to it
     // that all cases have been handled

src/deploy/functions/release/reporter.ts

@@ -28,7 +28,9 @@ export type OperationType =
   | "create topic"
   | "delete topic"
   | "set invoker"
-  | "set concurrency";
+  | "set concurrency"
+  | "register blocking trigger"
+  | "unregister blocking trigger";
 
 /** An error with a deployment phase. */
 export class DeploymentError extends Error {
@@ -245,5 +247,9 @@ export function triggerTag(endpoint: backend.Endpoint): string {
     return `${prefix}.https`;
   }
 
+  if (backend.isBlockingTriggered(endpoint)) {
+    return `${prefix}.blocking`;
+  }
+
   return endpoint.eventTrigger.eventType;
 }

src/deploy/functions/runtimes/discovery/v1alpha1.ts

@@ -19,6 +19,7 @@ export type ManifestEndpoint = backend.ServiceConfiguration &
   Partial<backend.CallableTriggered> &
   Partial<backend.EventTriggered> &
   Partial<backend.TaskQueueTriggered> &
+  Partial<backend.BlockingTriggered> &
   Partial<backend.ScheduleTriggered> & {
     region?: string[];
     entryPoint: string;
@@ -102,6 +103,7 @@ function parseEndpoints(
     eventTrigger: "object",
     scheduleTrigger: "object",
     taskQueueTrigger: "object",
+    blockingTrigger: "object",
   });
   let triggerCount = 0;
   if (ep.httpsTrigger) {
@@ -119,6 +121,9 @@ function parseEndpoints(
   if (ep.taskQueueTrigger) {
     triggerCount++;
   }
+  if (ep.blockingTrigger) {
+    triggerCount++;
+  }
   if (!triggerCount) {
     throw new FirebaseError("Expected trigger in endpoint " + id);
   }
@@ -192,6 +197,13 @@ function parseEndpoints(
         });
       }
       triggered = { taskQueueTrigger: ep.taskQueueTrigger };
+    } else if (backend.isBlockingTriggered(ep)) {
+      requireKeys(prefix + ".blockingTrigger", ep.blockingTrigger, "eventType");
+      assertKeyTypes(prefix + ".blockingTrigger", ep.blockingTrigger, {
+        eventType: "string",
+        options: "object",
+      });
+      triggered = { blockingTrigger: ep.blockingTrigger };
     } else {
       throw new FirebaseError(
         `Do not recognize trigger type for endpoint ${id}. Try upgrading ` +