From 7b0ebb9dbf799a8d4bf2abe89b39b9b032efb5ea Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 26 May 2022 10:35:43 +0800
Subject: [PATCH] Add pubsub subscription to trigger webhook

---
 packages/api/src/routers/svc/webhooks.ts | 86 ++++++++++++++++++++++++
 packages/api/src/server.ts               |  2 +
 2 files changed, 88 insertions(+)
 create mode 100644 packages/api/src/routers/svc/webhooks.ts

diff --git a/packages/api/src/routers/svc/webhooks.ts b/packages/api/src/routers/svc/webhooks.ts
new file mode 100644
index 000000000..c9b0697e0
--- /dev/null
+++ b/packages/api/src/routers/svc/webhooks.ts
@@ -0,0 +1,86 @@
+/* eslint-disable @typescript-eslint/no-misused-promises */
+/* eslint-disable @typescript-eslint/no-unsafe-assignment */
+/* eslint-disable @typescript-eslint/explicit-module-boundary-types */
+import express from 'express'
+import { readPushSubscription } from '../../datalayer/pubsub'
+import { getRepository } from '../../entity/utils'
+import { Webhook } from '../../entity/webhook'
+import axios, { Method } from 'axios'
+
+export function webhooksServiceRouter() {
+  const router = express.Router()
+
+  router.post('/:action', async (req, res) => {
+    console.log('trigger webhook of action', req.params.action)
+    const { message: msgStr, expired } = readPushSubscription(req)
+
+    if (!msgStr) {
+      res.status(400).send('Bad Request')
+      return
+    }
+
+    if (expired) {
+      console.log('discarding expired message')
+      res.status(200).send('Expired')
+      return
+    }
+
+    try {
+      const data = JSON.parse(msgStr)
+      const { userId, type } = data
+      if (!userId || !type) {
+        console.log('No userId or type found in message')
+        res.status(400).send('Bad Request')
+        return
+      }
+
+      // example: PAGE_CREATED
+      const eventType = `${type as string}_${req.params.action}`.toUpperCase()
+      const webhooks = await getRepository(Webhook)
+        .createQueryBuilder()
+        .where('userId = :userId', { userId })
+        .andWhere(':eventType = ANY(eventTypes)', { eventType })
+        .andWhere('enabled = true')
+        .getMany()
+
+      if (webhooks.length <= 0) {
+        console.log(
+          'No active webhook found for user',
+          userId,
+          'and eventType',
+          eventType
+        )
+        res.status(200).send('No webhook found')
+        return
+      }
+
+      // trigger webhooks
+      for (const webhook of webhooks) {
+        const url = webhook.url
+        const method = webhook.method as Method
+        const body = JSON.stringify({
+          action: req.params.action,
+          userId,
+          [type]: data,
+        })
+
+        console.log('triggering webhook', url, method, body)
+        await axios.request({
+          url,
+          method,
+          headers: {
+            'Content-Type': webhook.contentType,
+          },
+          data: body,
+        })
+      }
+
+      res.status(200).send('OK')
+    } catch (err) {
+      console.log('trigger webhook failed', err)
+      res.status(500).send(err)
+    }
+  })
+
+  return router
+}
diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts
index 87e642d4e..d65bbcdd8 100755
--- a/packages/api/src/server.ts
+++ b/packages/api/src/server.ts
@@ -42,6 +42,7 @@ import { corsConfig } from './utils/corsConfig'
 import { initElasticsearch } from './elastic'
 import { uploadServiceRouter } from './routers/svc/upload'
 import rateLimit from 'express-rate-limit'
+import { webhooksServiceRouter } from './routers/svc/webhooks'
 
 const PORT = process.env.PORT || 4000
 
@@ -111,6 +112,7 @@ export const createApp = (): {
   app.use('/svc/pubsub/newsletters', newsletterServiceRouter())
   app.use('/svc/pubsub/emails', emailsServiceRouter())
   app.use('/svc/pubsub/upload', uploadServiceRouter())
+  app.use('/svc/pubsub/webhooks', webhooksServiceRouter())
   app.use('/svc/reminders', remindersServiceRouter())
   app.use('/svc/pdf-attachments', pdfAttachmentsRouter())