wrapped subscription events

HTTPArchive · Dec 11, 2024 · 56814c4 · 56814c4
1 parent d7f1d0f
commit 56814c4
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 11 deletions.
diff --git a/infra/dataform-export/index.js b/infra/dataform-export/index.js
@@ -36,34 +36,37 @@ async function callRunJob (payload = {}) {
  */
 functions.http('dataform-export', async (req, res) => {
   try {
-    console.log(JSON.stringify(req.body))
-    const message = req.body
+    const message = req.body.message
     if (!message) {
       console.log(`no message received: ${JSON.stringify(req.body)}`)
       res.status(400).send('Bad Request: no message received')
-      return
     }
 
-    const query = message.protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.query.query
+    const messageData = (message.data && JSON.parse(Buffer.from(message.data, 'base64').toString('utf-8'))) || message
+    if (!messageData) {
+      console.info(JSON.stringify(message))
+      res.status(400).send('Bad Request: invalid message format')
+    }
+
+    const query = messageData.protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.query.query
     if (!query) {
-      console.log(`no query found: ${JSON.stringify(message)}`)
+      console.log(`no query found: ${JSON.stringify(messageData)}`)
       res.status(400).send('Bad Request: no query found')
-      return
     }
 
     const regex = /\/\* ({"dataform_trigger":.+) \*\//
     const reportConfig = regex.exec(query)
     if (!reportConfig) {
       console.log(`no trigger config found: ${query}`)
       res.status(400).send('Bad Request: no trigger config found')
-      return
     }
 
     const eventData = JSON.parse(reportConfig[1])
     await callRunJob(eventData)
 
     res.status(200).send('OK')
   } catch (error) {
+    console.log(JSON.stringify(req.body))
     console.error(error)
     res.status(500).send('Internal Server Error')
   }

diff --git a/infra/tf/function_dataform_export.tf b/infra/tf/function_dataform_export.tf
@@ -82,9 +82,6 @@ resource "google_pubsub_subscription" "dataform_export" {
       audience              = google_cloudfunctions2_function.dataform_export.service_config[0].uri
       service_account_email = local.function_identity
     }
-    no_wrapper {
-      write_metadata = false
-    }
   }
   retry_policy {
     maximum_backoff = "60s"

diff --git a/infra/tf/functions.tf b/infra/tf/functions.tf
@@ -3,7 +3,7 @@ locals {
 }
 
 resource "google_project_iam_member" "project" {
-  for_each = toset(["roles/bigquery.jobUser", "roles/dataform.serviceAgent", "roles/run.invoker"])
+  for_each = toset(["roles/bigquery.jobUser", "roles/dataform.serviceAgent", "roles/run.invoker", "roles/run.jobsExecutorWithOverrides"])
 
   project = local.project
   role    = each.value