Implement V1 compat API for PubSub

spec/v2/providers/pubsub.spec.ts

@@ -170,6 +170,40 @@ describe("onMessagePublished", () => {
     expect(json).to.deep.equal({ hello: "world" });
   });
 
+  it("should construct a CloudEvent with the correct context", async () => {
+    const publishTime = new Date().toISOString();
+    const message = {
+      messageId: "uuid",
+      data: Buffer.from(JSON.stringify({ hello: "world" })).toString("base64"),
+      publishTime,
+    };
+    const data: pubsub.MessagePublishedData = {
+      message: message as any,
+      subscription: "projects/aProject/subscriptions/aSubscription",
+    };
+    const event: CloudEvent<pubsub.MessagePublishedData> = {
+      specversion: "1.0",
+      id: "uuid",
+      time: publishTime,
+      type: "google.cloud.pubsub.topic.v1.messagePublished",
+      source: "//pubsub.googleapis.com/projects/aProject/topics/topic",
+      data,
+    };
+
+    let receivedEvent: CloudEvent<pubsub.MessagePublishedData<any>>;
+    const func = pubsub.onMessagePublished("topic", (e) => {
+      receivedEvent = e;
+    });
+
+    await func(event);
+
+    expect(receivedEvent.id).to.equal("uuid");
+    expect(receivedEvent.time).to.equal(publishTime);
+    expect(receivedEvent.type).to.equal("google.cloud.pubsub.topic.v1.messagePublished");
+    expect(receivedEvent.source).to.equal("//pubsub.googleapis.com/projects/aProject/topics/topic");
+    expect(receivedEvent.data.message.json).to.deep.equal({ hello: "world" });
+  });
+
   // These tests pass if the transpiler works
   it("allows desirable syntax", () => {
     pubsub.onMessagePublished<string>(
@@ -193,4 +227,103 @@ describe("onMessagePublished", () => {
       (event: CloudEvent<pubsub.MessagePublishedData>) => undefined
     );
   });
+
+  it("should not modify a CloudEvent that already has a context", async () => {
+    const publishTime = new Date().toISOString();
+    const message = {
+      messageId: "uuid",
+      data: Buffer.from(JSON.stringify({ hello: "world" })).toString("base64"),
+      publishTime,
+    };
+    const data: pubsub.MessagePublishedData = {
+      message: message as any,
+      subscription: "projects/aProject/subscriptions/aSubscription",
+    };
+    const existingContext = {
+      eventId: "custom-id",
+      timestamp: publishTime,
+      eventType: "custom.type",
+      resource: "custom/resource",
+      params: {},
+    };
+    const event: CloudEvent<pubsub.MessagePublishedData> = {
+      specversion: "1.0",
+      id: "uuid",
+      time: publishTime,
+      type: "google.cloud.pubsub.topic.v1.messagePublished",
+      source: "//pubsub.googleapis.com/projects/aProject/topics/topic",
+      data,
+      context: existingContext as any,
+    };
+
+    let receivedEvent: CloudEvent<pubsub.MessagePublishedData<any>>;
+    const func = pubsub.onMessagePublished("topic", (e) => {
+      receivedEvent = e;
+    });
+
+    await func(event);
+
+    expect(receivedEvent.context).to.deep.equal(existingContext);
+  });
+
+  it("should use GCLOUD_PROJECT as fallback for resource name", async () => {
+    const publishTime = new Date().toISOString();
+    const message = {
+      messageId: "uuid",
+      data: Buffer.from(JSON.stringify({ hello: "world" })).toString("base64"),
+      publishTime,
+    };
+    const data: pubsub.MessagePublishedData = {
+      message: message as any,
+      subscription: "projects/aProject/subscriptions/aSubscription",
+    };
+    const event: CloudEvent<pubsub.MessagePublishedData> = {
+      specversion: "1.0",
+      id: "uuid",
+      time: publishTime,
+      type: "google.cloud.pubsub.topic.v1.messagePublished",
+      source: "//pubsub.googleapis.com/topics/topic", // Malformed source
+      data,
+    };
+
+    let receivedEvent: CloudEvent<pubsub.MessagePublishedData<any>>;
+    const func = pubsub.onMessagePublished("topic", (e) => {
+      receivedEvent = e;
+    });
+
+    await func(event);
+
+    expect(receivedEvent.context.resource.name).to.equal("projects/aProject/topics/topic");
+  });
+
+  it("should use 'unknown-project' as fallback for resource name", async () => {
+    delete process.env.GCLOUD_PROJECT;
+    const publishTime = new Date().toISOString();
+    const message = {
+      messageId: "uuid",
+      data: Buffer.from(JSON.stringify({ hello: "world" })).toString("base64"),
+      publishTime,
+    };
+    const data: pubsub.MessagePublishedData = {
+      message: message as any,
+      subscription: "projects/aProject/subscriptions/aSubscription",
+    };
+    const event: CloudEvent<pubsub.MessagePublishedData> = {
+      specversion: "1.0",
+      id: "uuid",
+      time: publishTime,
+      type: "google.cloud.pubsub.topic.v1.messagePublished",
+      source: "//pubsub.googleapis.com/topics/topic", // Malformed source
+      data,
+    };
+
+    let receivedEvent: CloudEvent<pubsub.MessagePublishedData<any>>;
+    const func = pubsub.onMessagePublished("topic", (e) => {
+      receivedEvent = e;
+    });
+
+    await func(event);
+
+    expect(receivedEvent.context.resource.name).to.equal("project/unknown-project/topics/topic");
+  });
 });

src/v2/core.ts

@@ -25,6 +25,7 @@
  * @packageDocumentation
  */
 
+import type { EventContext } from "../v1/cloud-functions";
 import { Change } from "../common/change";
 import { ManifestEndpoint } from "../runtime/manifest";
 
@@ -91,6 +92,13 @@
 
   /** Information about this specific event. */
   data: T;
+
+  /** V1- compatible context of this event.
+   * 
+   * This getter is added at runtime for V1 compatibility.
+   * May be undefined it not set by a provider
+   */
+  readonly context?: EventContext;
 }
 
 /**

src/v2/providers/pubsub.ts

@@ -34,6 +34,7 @@
 import * as options from "../options";
 import { SecretParam } from "../../params/types";
 import { withInit } from "../../common/onInit";
+import type { EventContext, Resource } from "../../v1/cloud-functions";
 
 /**
  * Google Cloud Pub/Sub is a globally distributed message bus that automatically scales as you need it.
@@ -304,7 +305,9 @@
       subscription: string;
     };
     messagePublishedData.message = new Message(messagePublishedData.message);
-    return wrapTraceContext(withInit(handler))(raw as CloudEvent<MessagePublishedData<T>>);
+    const event = raw as CloudEvent<MessagePublishedData<T>>;
+    attachPubSubContext(event, topic);
+    return wrapTraceContext(withInit(handler))(event);
   };
 
   func.run = handler;
@@ -353,3 +356,66 @@
 
   return func;
 }
+
+/**
+ * @internal
+ *
+ * Adds a v1-style context to the event.
+ *
+ * @param event - The event to add the context to.
+ * @param topic - The topic the event is for.
+ */
+function attachPubSubContext<T>(event: CloudEvent<MessagePublishedData<T>>, topic: string) {
+  if ("context" in event && event.context) {
+    return;
+  }
+
+  const resourceName = getResourceName(event, topic);
+  const resource: Resource = {
+
+    service: "pubsub.googleapis.com",
+    name: resourceName ?? "",
+
+  };
+
+  const context: EventContext = {
+    eventId: event.id,
+    timestamp: event.time,
+    resource,
+    eventType: "google.cloud.pubsub.topic.v1.messagePublished",
+    params: {}
+
+  };
+
+  Object.defineProperty(event, "context", {
+
+    get: () => context,
+    enumerable: false,
+    configurable: false,
+
+  });
+}
+
+/**
+ * @internal
+ *
+ * Extracts the resource name from the event source.
+ *
+ * @param event - The event to extract the resource name from.
+ * @param topic - The topic the event is for.
+ * @returns The resource name.
+ */
+function getResourceName(event: CloudEvent<MessagePublishedData<any>>, topic: string) {
+  const match = event.source?.match(/projects\/([^/]+)\/topics\/([^/]+)/);
+  const project =
+    match?.[1] ?? process.env.GCLOUD_PROJECT ?? process.env.GCLOUD_PROJECT_ID ?? process.env.PROJECT_ID;
+
+  const topicName = match?.[2] ?? topic;
+
+  if (!project) {
+    return `project/unknown-project/topics/${topicName}`;
+  }
+
+  return `projects/${project}/topics/${topicName}`;
+
+}