Handling `application/protobuf` Payloads in Eventarc Events

[yamankatby]

Eventarc events, such as those from Cloud Firestore, utilize application/protobuf for their payloads. Could you provide guidance or examples on decoding and processing this format effectively? A sample or demonstration would be greatly appreciated.

[ivanmartinvalle]

Finally figured this out after hours of Google searching.

background

• nodejs-proto-files
  • purpose: protos, distributed as an npm package
  • issue regarding missing protos
• protobufjs
  • purpose: library for parsing binary content using protos
  • issue regarding relative imports
• google-cloudevents-nodejs
  • purpose: typescript types (not protos)
  • issue regarding CloudEvent improvements
  • issue regarding not supporting protobuf
• google-cloudevents
  • purpose: CloudEvent protos
• cloudevents/spec
  • relevant purpose: describe CloudEvent HTTP binding spec
• googleapis/googleapis
  • purpose: protos

node, typescript, and express implementation

• because nodejs-proto-files is missing protos, you must download proto files yourself and ship them with your application
  • googleapi/googleapis: cloud protos
  • google-cloudevents: event protos
• the content type is application/protobuf, so you will need app.use(express.raw({type:'application/protobuf'})) for req.body to be a Buffer
• you need to decode new Uint8Array(buffer), not the Buffer itself
• the POST body is only data. The other properties of the CloudEvent envelope come in via headers.
• directory structure
  • src/controllers/thiscontroller.ts
  • protos/google/...
  • keep same directory structure as the source proto repos...probably

Below is a proof-of-concept implementation. DO NOT USE THIS IN PRODUCTION AS-IS. There's probably stuff you need to optimize with loading all protos the application needs, possibly bundling them. Probably also missing stuff with hydrating the CloudEvent object. Also not very pretty reusable.

import path from 'path'
import protobuf from 'protobufjs'
import { toDocumentEventData } from '@google/events/cloud/firestore/v1/DocumentEventData'

export const firestore = async (
  req: Request<never, never, Buffer>,
  res: Response,
): Promise<void> => {
  const cloudEvent = await decodeCloudEventFromProto(
    req,
    '../../proto',
    ['google/events/cloud/firestore/v1/data.proto', 'google/type/latlng.proto'],
    'google.events.cloud.firestore.v1.DocumentEventData',
    toDocumentEventData,
  )

  await processFirestoreEvent(cloudEvent)

  res.json({})
}

const decodeCloudEventFromProto = async <T>(
  req: Request,
  protoBaseDirectory: string,
  protos: string[],
  type: string,
  toFunction: (json: object) => T,
): Promise<CloudEvent<T>> => {
  const ceEntries = Object.entries(req.headers)
    .filter(([key]) => key.startsWith('ce-'))
    .map(([key, value]) => [key.replace('ce-', ''), value])

  const cloudEvent = new CloudEvent<T>(
    {
      ...Object.fromEntries(ceEntries)
    },
    false,
  )

  const decoded = await decodeProtoFromBuffer(
    protoBaseDirectory,
    protos,
    type,
    req.body,
    toFunction,
  )

  return cloudEvent.cloneWith({
    data: decoded,
  })
}

const decodeProtoFromBuffer = async <T>(
  protoBaseDirectory: string,
  protos: string[],
  type: string,
  buffer: Buffer,
  toFunction: (json: object) => T,
): Promise<T> => {
  const root = new protobuf.Root({})
  root.resolvePath = (origin, target): string | null => {
    if (target.startsWith('google/type')) {
      return path.resolve(__dirname, protoBaseDirectory, target)
    }

    return target
  }

  const files = protos.map((x) =>
    path.resolve(__dirname, protoBaseDirectory, x),
  )
  await root.load(files)

  const decoded = root.lookupType(type).decode(new Uint8Array(buffer))

  return toFunction(decoded)
}