fix: properly rehydrate Eventarc CloudEvents from Raw Pub/Sub fallback

src/pubsub_middleware.ts

@@ -150,6 +150,59 @@
   },
 });
 
+/**
+ * Attempt to restore a dropped CloudEvent that Eventarc sent as a Raw PubSub Message.
+ *
+ * This occurs when Eventarc's Push Subscription fails to unwrap CloudEvent HTTP 
+ * headers. For example, if a document ID contains a trailing space, the `ce-subject` 
+ * HTTP header is considered invalid by Node's strict HTTP parser and is stripped. 
+ * Eventarc gracefully falls back to delivering a raw Pub/Sub message body where 
+ * the CloudEvent metadata is buried inside the `attributes` dictionary.
+ * 
+ * This method rescues the event by mutating the Express request headers (promoting 
+ * the hidden attributes into standard `ce-` HTTP headers). This guarantees that
+ * downstream middlewares (like `isBinaryCloudEvent()`) will successfully validate 
+ * the request as a Binary Cloud Event.
+ *
+ * @param req - Express request object to hydrate with CloudEvent headers
+ * @param body - An unmarshalled http request body from a Pub/Sub push subscription
+ * @param attributes - The Pub/Sub message attributes
+ */
+const rehydrateCloudEvent = (
+  req: Request,
+  body: RawPubSubBody,
+  attributes: {[key: string]: string}
+): void => {
+  // 1. Promote attributes to HTTP headers. The official CloudEvents Pub/Sub Protocol 
+  // Binding dictates that attributes map without the `ce-` prefix. However, some 
+  // Google Cloud systems attach them with the `ce-` prefix. We handle both.
+  for (const [key, value] of Object.entries(attributes)) {
+    if (key.startsWith('ce-')) {
+      req.headers[key.toLowerCase()] = value;
+    } else if (['type', 'source', 'subject', 'id', 'time', 'specversion', 'datacontenttype'].includes(key.toLowerCase())) {
+      req.headers['ce-' + key.toLowerCase()] = value;
+    }
+  }
+
+  // 2. Extract the binary payload. 
+  const dataBuf = Buffer.from(body.message.data || '', 'base64');
+  const contentType = req.headers['ce-datacontenttype'] || req.headers['content-type'] || '';
+
+  // 3. Attempt JSON parse. If it fails (e.g. malformed JSON or a mismatched content-type 
+  // like application/protobuf), fallback to the raw Buffer. This mirrors `bodyParser.raw()` 
+  // and safely passes the binary payload to the user's function without crashing the HTTP stream.
+  if (typeof contentType === 'string' && contentType.includes('application/json')) {
+    try {
+      req.body = JSON.parse(dataBuf.toString('utf8'));
+      return;
+    } catch {
+      req.body = dataBuf;
+      return;
+    }
+  }
+  req.body = dataBuf;
+};
+
 /**
  * Express middleware used to marshal the HTTP request body received directly from a
  * Pub/Sub subscription into the format that is expected downstream by wrapEventFunction
@@ -164,7 +217,12 @@
 ) => {
   const {body, path} = req;
   if (isRawPubSubRequestBody(body) && !isBinaryCloudEvent(req)) {
-    req.body = marshalPubSubRequestBody(body, path);
+    const pubsubAttributes = body.message.attributes || {};
+    if ((pubsubAttributes['ce-type'] || pubsubAttributes['type']) && (pubsubAttributes['ce-source'] || pubsubAttributes['source'])) {
+      rehydrateCloudEvent(req, body, pubsubAttributes);
+    } else {
+      req.body = marshalPubSubRequestBody(body, path);
+    }
   }
   next();
 };

test/pubsub_middleware.ts

@@ -65,6 +65,7 @@ describe('legacyPubSubEventMiddleware', () => {
     path: string;
     body: object;
     expectedBody: () => object;
+    expectedHeaders?: () => object;
   }
 
   const testData: TestData[] = [
@@ -86,6 +87,52 @@ describe('legacyPubSubEventMiddleware', () => {
       body: {foo: 'bar'},
       expectedBody: () => ({foo: 'bar'}),
     },
+    {
+      name: 'Eventarc fallback CloudEvent (JSON payload)',
+      path: `/${PUB_SUB_TOPIC}`,
+      body: {
+        subscription: 'projects/FOO/subscriptions/BAR_SUB',
+        message: {
+          data: Buffer.from('{"hello":"world"}', 'utf8').toString('base64'),
+          messageId: '1',
+          attributes: {
+            'ce-type': 'google.cloud.firestore.document.v1.written',
+            'ce-source': '//firestore.googleapis.com/...',
+            'ce-subject': 'documents/... ',
+            'ce-datacontenttype': 'application/json',
+          },
+        },
+      },
+      expectedBody: () => ({hello: 'world'}),
+      expectedHeaders: () => ({
+        'ce-type': 'google.cloud.firestore.document.v1.written',
+        'ce-source': '//firestore.googleapis.com/...',
+        'ce-subject': 'documents/... ',
+        'ce-datacontenttype': 'application/json',
+      }),
+    },
+    {
+      name: 'Eventarc fallback CloudEvent (Unprefixed attributes & Binary payload)',
+      path: `/${PUB_SUB_TOPIC}`,
+      body: {
+        subscription: 'projects/FOO/subscriptions/BAR_SUB',
+        message: {
+          data: Buffer.from('bad-json', 'utf8').toString('base64'),
+          messageId: '1',
+          attributes: {
+            'type': 'google.cloud.storage.object.v1.finalized',
+            'source': '//storage.googleapis.com/...',
+            'datacontenttype': 'application/protobuf',
+          },
+        },
+      },
+      expectedBody: () => Buffer.from('bad-json', 'utf8'),
+      expectedHeaders: () => ({
+        'ce-type': 'google.cloud.storage.object.v1.finalized',
+        'ce-source': '//storage.googleapis.com/...',
+        'ce-datacontenttype': 'application/protobuf',
+      }),
+    },
   ];
 
   testData.forEach((test: TestData) => {
@@ -95,11 +142,15 @@ describe('legacyPubSubEventMiddleware', () => {
       const request = {
         path: test.path,
         body: test.body,
+        headers: {} as {[key: string]: string},
         // eslint-disable-next-line @typescript-eslint/no-unused-vars
         header: (_: string) => '',
       };
       legacyPubSubEventMiddleware(request as Request, {} as Response, next);
       assert.deepStrictEqual(request.body, test.expectedBody());
+      if (test.expectedHeaders) {
+        assert.deepStrictEqual(request.headers, test.expectedHeaders());
+      }
       assert.strictEqual(next.called, true);
       clock.restore();
     });