Add notification cloud run

packages/api/src/services/rules.ts

@@ -1,6 +1,7 @@
 import { RuleAction, RuleActionType } from '../generated/graphql'
 import { CreateSubscriptionOptions } from '@google-cloud/pubsub'
 import { env } from '../env'
+import { getDeviceTokensByUserId } from './user_device_tokens'
 
 enum RuleTrigger {
   ON_PAGE_UPDATE,
@@ -40,12 +41,12 @@ export const getPubSubSubscriptionName = (
   return `${topicName}-${userId}-rule-${ruleName}`
 }
 
-export const getPubSubSubscriptionOptions = (
+export const getPubSubSubscriptionOptions = async (
   userId: string,
   ruleName: string,
   filter: string,
   action: RuleAction
-): CreateSubscriptionOptions => {
+): Promise<CreateSubscriptionOptions> => {
   const options: CreateSubscriptionOptions = {
     messageRetentionDuration: 60 * 10, // 10 minutes
     expirationPolicy: {
@@ -70,10 +71,17 @@ export const getPubSubSubscriptionOptions = (
         throw new Error('Missing notification messages')
       }
 
+      const deviceTokens = await getDeviceTokensByUserId(userId)
+      if (!deviceTokens || deviceTokens.length === 0) {
+        throw new Error('No device tokens found')
+      }
+
       options.pushConfig = {
-        pushEndpoint: `${env.queue.notificationEndpoint}/${userId}`,
+        pushEndpoint: `${env.queue.notificationEndpoint}/${userId}?token=${env.queue.verificationToken}`,
         attributes: {
+          filter,
           messages: JSON.stringify(params),
+          tokens: JSON.stringify(deviceTokens.map((t) => t.token)),
         },
       }
       break

packages/api/src/util.ts

@@ -66,6 +66,7 @@ interface BackendEnv {
     integrationTaskHandlerUrl: string
     textToSpeechTaskHandlerUrl: string
     notificationEndpoint: string
+    verificationToken: string
   }
   fileUpload: {
     gcsUploadBucket: string
@@ -154,6 +155,7 @@ const nullableEnvVars = [
   'AZURE_SPEECH_REGION',
   'GCP_LOCATION',
   'NOTIFICATION_ENDPOINT',
+  'PUBSUB_VERIFICATION_TOKEN',
 ] // Allow some vars to be null/empty
 
 /* If not in GAE and Prod/QA/Demo env (f.e. on localhost/dev env), allow following env vars to be null */
@@ -240,6 +242,7 @@ export function getEnv(): BackendEnv {
     integrationTaskHandlerUrl: parse('INTEGRATION_TASK_HANDLER_URL'),
     textToSpeechTaskHandlerUrl: parse('TEXT_TO_SPEECH_TASK_HANDLER_URL'),
     notificationEndpoint: parse('NOTIFICATION_ENDPOINT'),
+    verificationToken: parse('PUBSUB_VERIFICATION_TOKEN'),
   }
   const imageProxy = {
     url: parse('IMAGE_PROXY_URL'),

packages/notification/.dockerignore

@@ -0,0 +1,5 @@
+node_modules
+build
+.env*
+Dockerfile
+.dockerignore

packages/notification/.eslintignore

@@ -0,0 +1,2 @@
+node_modules/
+build/

packages/notification/.eslintrc

@@ -0,0 +1,6 @@
+{
+  "extends": "../../.eslintrc",
+  "parserOptions": {
+    "project": "tsconfig.json"
+  }
+}

packages/notification/.gcloudignore

@@ -0,0 +1,16 @@
+# This file specifies files that are *not* uploaded to Google Cloud Platform
+# using gcloud. It follows the same syntax as .gitignore, with the addition of
+# "#!include" directives (which insert the entries of the given .gitignore-style
+# file at that point).
+#
+# For more information, run:
+#   $ gcloud topic gcloudignore
+#
+.gcloudignore
+# If you would like to upload your .git directory, .gitignore file or files
+# from your .gitignore file, remove the corresponding line
+# below:
+.git
+.gitignore
+
+node_modules

packages/notification/.npmignore

@@ -0,0 +1 @@
+/test/

packages/notification/Dockerfile

@@ -0,0 +1,26 @@
+FROM node:14.18-alpine
+
+# Run everything after as non-privileged user.
+WORKDIR /app
+
+COPY package.json .
+COPY yarn.lock .
+COPY tsconfig.json .
+COPY .eslintrc .
+
+COPY /packages/notification/package.json ./packages/notification/package.json
+
+RUN yarn install --pure-lockfile
+
+ADD /packages/notification ./packages/notification
+RUN yarn workspace @omnivore/notification build
+
+# After building, fetch the production dependencies
+RUN rm -rf /app/packages/notification/node_modules
+RUN rm -rf /app/node_modules
+RUN yarn install --pure-lockfile --production
+
+EXPOSE 8080
+
+CMD ["yarn", "workspace", "@omnivore/notification", "start"]
+

packages/notification/mocha-config.json

@@ -0,0 +1,5 @@
+{
+    "extension": ["ts"],
+    "spec": "test/**/*.test.ts",
+    "require": "test/babel-register.js"
+  }

packages/notification/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "@omnivore/notification",
+  "version": "1.0.0",
+  "main": "build/src/index.js",
+  "files": [
+    "build/src"
+  ],
+  "license": "Apache-2.0",
+  "scripts": {
+    "test": "yarn mocha -r ts-node/register --config mocha-config.json",
+    "lint": "eslint src --ext ts,js,tsx,jsx",
+    "compile": "tsc",
+    "build": "tsc",
+    "start": "functions-framework --target=notification",
+    "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\"",
+    "gcloud-deploy": "gcloud functions deploy notification --gen2 --entry-point=notification --trigger-http --allow-unauthenticated --region=us-west2 --runtime nodejs14",
+    "deploy": "yarn build && yarn gcloud-deploy"
+  },
+  "devDependencies": {
+    "chai": "^4.3.6",
+    "eslint-plugin-prettier": "^4.0.0",
+    "mocha": "^10.0.0"
+  },
+  "dependencies": {
+    "@google-cloud/functions-framework": "3.1.2",
+    "@google-cloud/pubsub": "^3.2.1",
+    "firebase-admin": "^10.0.2",
+    "@sentry/serverless": "^6.16.1"
+  }
+}

packages/notification/src/index.ts

@@ -0,0 +1,84 @@
+import * as Sentry from '@sentry/serverless'
+import { Request, Response } from 'express'
+import { sendBatchPushNotifications } from './sendNotification'
+import { Message } from 'firebase-admin/lib/messaging'
+
+interface SubscriptionData {
+  attributes?: string
+  data: string
+}
+
+const readPushSubscription = (req: Request): SubscriptionData | null => {
+  console.debug('request query', req.body)
+
+  if (req.query.token !== process.env.PUBSUB_VERIFICATION_TOKEN) {
+    console.log('query does not include valid pubsub token')
+    return null
+  }
+
+  // GCP PubSub sends the request as a base64 encoded string
+  if (!('message' in req.body)) {
+    console.log('Invalid pubsub message: message not in body')
+    return null
+  }
+
+  const body = req.body as { message: { data: string }; attributes?: string }
+  const data = Buffer.from(body.message.data, 'base64').toString('utf-8')
+
+  return {
+    data,
+    attributes: body.attributes,
+  }
+}
+
+const getBatchMessages = (messages: string[], tokens: string[]): Message[] => {
+  const batchMessages: Message[] = []
+  messages.forEach((message) => {
+    tokens.forEach((token) => {
+      batchMessages.push({
+        token,
+        notification: {
+          body: message,
+        },
+      })
+    })
+  })
+
+  return batchMessages
+}
+
+export const notification = Sentry.GCPFunction.wrapHttpFunction(
+  async (req: Request, res: Response) => {
+    const subscriptionData = readPushSubscription(req)
+    if (!subscriptionData) {
+      res.status(400).send('Invalid request')
+      return
+    }
+
+    const { attributes } = subscriptionData
+    if (!attributes) {
+      res.status(400).send('Invalid request')
+      return
+    }
+
+    const { messages, tokens } = JSON.parse(attributes) as {
+      messages: string[]
+      tokens: string[]
+    }
+    if (!messages || messages.length === 0 || !tokens || tokens.length === 0) {
+      res.status(400).send('Invalid request')
+      return
+    }
+
+    const batchMessages = getBatchMessages(messages, tokens)
+
+    try {
+      await sendBatchPushNotifications(batchMessages)
+
+      res.status(200).send('OK')
+    } catch (error) {
+      console.error(error)
+      res.status(500).send('Internal server error')
+    }
+  }
+)

packages/notification/src/sendNotification.ts

@@ -0,0 +1,31 @@
+import { initializeApp } from 'firebase-admin/app'
+import {
+  BatchResponse,
+  getMessaging,
+  Message,
+  MulticastMessage,
+} from 'firebase-admin/messaging'
+
+// getting credentials from App Engine
+initializeApp()
+
+export const sendPushNotification = async (
+  message: Message
+): Promise<string | undefined> => {
+  return getMessaging().send(message)
+}
+
+export const sendMulticastPushNotifications = async (
+  message: MulticastMessage
+): Promise<BatchResponse | undefined> => {
+  return getMessaging().sendMulticast(message)
+}
+
+export const sendBatchPushNotifications = async (
+  messages: Message[]
+): Promise<BatchResponse | undefined> => {
+  const res = await getMessaging().sendAll(messages)
+  console.debug('success count: ', res.successCount)
+
+  return res
+}

packages/notification/test/babel-register.js

@@ -0,0 +1,3 @@
+const register = require('@babel/register').default
+
+register({ extensions: ['.ts', '.tsx', '.js', '.jsx'] })

packages/notification/tsconfig.json

@@ -0,0 +1,11 @@
+{
+  "extends": "@tsconfig/node14/tsconfig.json",
+  "compilerOptions": {
+    "outDir": "build",
+    "rootDir": ".",
+    "lib": ["dom"],
+    // Generate d.ts files
+    "declaration": true
+  },
+  "include": ["src"],
+}