GCP Pub/Sub script with preprocessor template

/**

 * General Trigger Preprocessor

 * 

 * ⚠️ This function runs BEFORE the main function.

 * 

 * It processes raw trigger data (e.g., MQTT, HTTP, SQS) before passing it to `main()`.

 * Common tasks:

 * - Convert binary payloads to string/JSON

 * - Extract metadata

 * - Filter messages

 * - Add timestamps/context

 * 

 * The returned object determines `main()` parameters:

 * - `{a: 1, b: 2}` → `main(a, b)`

 * - `{payload}` → `main(payload)`

 * 

 * @param event - Trigger data and details (e.g., MQTT, HTTP)

 * @returns Processed data for `main()`

 */

export async function preprocessor(

  event: {

    payload: string,

    kind: 'gcp',

    message_id: string,

    subscription: string,

    ordering_key?: string,

    attributes?: Record<string, string>,

    delivery_type: "push" | "pull",

    headers?: Record<string, string>,

    publish_time?: string,

    ack_id?: string // acknowledgment ID for manual acknowledgment (only for pull delivery)

  },

) {

  if (event.kind === 'gcp') {

    const decodedString = atob(event.payload); // Direct decode, assume valid base64

    const attributes = event.attributes || {};

    const contentType = attributes['content-type'] || attributes['Content-Type'];

    const isJson = contentType === 'application/json';

    let parsedMessage: any = decodedString;

    if (isJson) {

      try {

        parsedMessage = JSON.parse(decodedString);

      } catch (err) {

        throw new Error(`Invalid JSON payload: ${err}`);

      }

    }

    return {

      messageAsDecodedString: decodedString,

      contentType,

      parsedMessage,

      attributes,

      ack_id: event.ack_id

    };

  }

  // We assume the script is triggered by a GCP message, which is why an error is thrown for other trigger kinds.

  // If the script is intended to support other triggers, update this logic to handle the respective trigger kind.

  throw new Error(`Expected gcp trigger kind got: ${event.kind}`)

}

/**

 * Main Function - Handles processed trigger events

 * 

 * ⚠️ Called AFTER `preprocessor()`, with its return values.

 * 

 * @param payload - Raw base64 encoded payload

 * @param payloadAsString - Decoded string payload

 * @param contentType - Content type from message attributes (e.g., application/json)

 * @param parsedMessage - Parsed object if payload is JSON, otherwise same as payloadAsString

 * @param attributes - Key-value metadata sent with the message

 * @param ack_id - Acknowledgment ID for manual acknowledgment (only available for pull delivery, undefined for push)

 */

export async function main(

  messageAsDecodedString: string,

  contentType?: string,

  parsedMessage?: any,

  attributes?: Record<string, string>,

  ack_id?: string

) {

}