switch to firestore based events

Author: Ehesp

integration_test/cli.ts

@@ -75,7 +75,6 @@ async function writeFirebaseJson(codebase: string): Promise<void> {
     functions: [
       {
         source: "functions",
-        codebase: codebase,
         disallowLegacyRuntimeConfig: true,
         ignore: [
           "node_modules",

integration_test/functions/src/firestore.v2.test.ts

@@ -1,70 +1,53 @@
-import EventEmitter from "node:events";
 import { describe, it, beforeAll, expect } from "vitest";
-import { PubSub } from "@google-cloud/pubsub";
 import { firestore } from "./utils";
 import { GeoPoint } from "firebase-admin/firestore";
 const RUN_ID = String(process.env.RUN_ID);
 
-const pubsub = new PubSub({ projectId: "cf3-integration-tests-v2-qa" });
-const emitter = new EventEmitter();
-
 function waitForEvent<T = unknown>(
   event: string,
   trigger: () => Promise<void>,
   timeoutMs: number = 60_000
 ): Promise<T> {
   return new Promise<T>((resolve, reject) => {
-    emitter.on(event, (data: T) => {
-      emitter.off(event, resolve);
-      resolve(data);
+    let timer: NodeJS.Timeout | null = null;
+
+    const unsubscribe = firestore.collection(`${RUN_ID}_snapshots`).doc(event).onSnapshot(snapshot => {
+      if (snapshot.exists) {
+        console.log("snapshot", snapshot.data());
+        if (timer) clearTimeout(timer);
+        unsubscribe();
+        resolve(snapshot.data() as T);
+      }
     });
 
-    setTimeout(() => {
-      emitter.off(event, resolve);
-      reject(new Error("Timeout waiting for event: " + event));
+    timer = setTimeout(() => {
+      unsubscribe();
+      reject(new Error(`Timeout waiting for event "${event}" after ${timeoutMs}ms`));
     }, timeoutMs);
 
-    trigger().catch(reject);
+    trigger().then().catch(reject);
   });
 }
 
-beforeAll(async () => {
-  const topic = pubsub.topic('vitest');
-  const subscription = topic.subscription('vitest-sub');
-
-  subscription.on("message", (message) => {
-    console.log("message", message.data.toString());
-    const data = message.data.length ? JSON.parse(message.data.toString()) : null;
-    message.ack();
-
-    if (!("event" in data)) {
-      throw new Error("Invalid event data: " + JSON.stringify(data));
-    }
-
-    emitter.emit(data.event, data.data);
-  });
-
-  subscription.on("error", (error) => {
-    console.error("Pubsub error", error);
-    process.exit(1);
-  });
-});
-
 describe("firestore.v2", () => {
   describe("onDocumentCreated", () => {
     let data: any;
 
     beforeAll(async () => {
       data = await waitForEvent("onDocumentCreated", async () => {
+        console.log("triggering event", RUN_ID);
         await firestore
           .collection(RUN_ID)
           .doc("onDocumentCreated")
           .set({
             foo: "bar",
             timestamp: new Date(),
             geopoint: new GeoPoint(10, 20),
+          }).then(() => {
+            console.log("event triggered", RUN_ID);
           });
       });
+      console.log("data", data);
     });
 
     it("should be a CloudEvent", () => {

integration_test/functions/src/firestore.v2.ts

@@ -1,29 +1,25 @@
 import { onDocumentCreated } from "firebase-functions/v2/firestore";
 import { RUN_ID, serializeData } from "./utils";
-import { PubSub } from "@google-cloud/pubsub";
 import { logger } from "firebase-functions";
+import { firestore } from "./utils";
 
-const pubsub = new PubSub();
-
-async function sendEvent(event: string, data: unknown): Promise<void> {
-  const topic = pubsub.topic('vitest');
-
-  await topic.publishMessage({
-    data: event
-      ? Buffer.from(
-          JSON.stringify({
-            event,
-            data,
-          })
-        )
-      : Buffer.from(""),
-  });
+async function sendEvent(event: string, data: any): Promise<void> {
+  await firestore.collection(`${RUN_ID}_snapshots`).doc(event).set(serializeData(data));
 }
 
 export const firestoreOnDocumentCreatedTrigger = onDocumentCreated(
   `${RUN_ID}/{documentId}`,
   async (event) => {
     logger.debug("onDocumentCreated", event);
-    await sendEvent("onDocumentCreated", serializeData(event));
+    await sendEvent("onDocumentCreated", event);
   }
 );
+
+
+export const foo = onDocumentCreated(
+  `test/{documentId}`,
+  async (event) => {
+    logger.debug("onDocumentCreated", event);
+    await sendEvent("onDocumentCreated", event);
+  }
+);

integration_test/functions/src/utils.ts

@@ -6,7 +6,7 @@ export const firestore = adminApp.firestore();
 
 export const RUN_ID = String(process.env.RUN_ID);
 
-export function serializeData(data: unknown): unknown {
+export function serializeData(data: any): any {
   if (data === null || data === undefined) {
     return null;
   }