Add notification endpoint

packages/api/src/datalayer/pubsub.ts

@@ -1,4 +1,4 @@
-import { PubSub } from '@google-cloud/pubsub'
+import { CreateSubscriptionOptions, PubSub } from '@google-cloud/pubsub'
 import { env } from '../env'
 import { ReportType } from '../generated/graphql'
 import express from 'express'
@@ -143,3 +143,38 @@ export const readPushSubscription = (
 
   return { message: message, expired: expired(body) }
 }
+
+export const createPubSubSubscription = async (
+  topicName: string,
+  subscriptionName: string,
+  options?: CreateSubscriptionOptions
+) => {
+  const topic = client.topic(topicName)
+  const [exists] = await topic.exists()
+  if (!exists) {
+    await topic.create()
+  }
+
+  const subscription = topic.subscription(subscriptionName)
+  const [subscriptionExists] = await subscription.exists()
+  if (!subscriptionExists) {
+    await subscription.create(options)
+  }
+}
+
+export const deletePubSubSubscription = async (
+  topicName: string,
+  subscriptionName: string
+) => {
+  const topic = client.topic(topicName)
+  const [exists] = await topic.exists()
+  if (!exists) {
+    return
+  }
+
+  const subscription = topic.subscription(subscriptionName)
+  const [subscriptionExists] = await subscription.exists()
+  if (subscriptionExists) {
+    await subscription.delete()
+  }
+}

packages/api/src/entity/rule.ts

@@ -22,7 +22,7 @@ export class Rule {
   name!: string
 
   @Column('text')
-  query!: string
+  filter!: string
 
   @Column('simple-json')
   actions!: { type: string; params: string[] }[]

packages/api/src/generated/graphql.ts

@@ -1652,9 +1652,9 @@ export type Rule = {
   actions: Array<RuleAction>;
   createdAt: Scalars['Date'];
   enabled: Scalars['Boolean'];
+  filter: Scalars['String'];
   id: Scalars['ID'];
   name: Scalars['String'];
-  query: Scalars['String'];
   updatedAt: Scalars['Date'];
 };
 
@@ -1973,9 +1973,9 @@ export type SetRuleInput = {
   actions: Array<RuleActionInput>;
   description?: InputMaybe<Scalars['String']>;
   enabled: Scalars['Boolean'];
+  filter: Scalars['String'];
   id?: InputMaybe<Scalars['ID']>;
   name: Scalars['String'];
-  query: Scalars['String'];
 };
 
 export type SetRuleResult = SetRuleError | SetRuleSuccess;
@@ -4406,9 +4406,9 @@ export type RuleResolvers<ContextType = ResolverContext, ParentType extends Reso
   actions?: Resolver<Array<ResolversTypes['RuleAction']>, ParentType, ContextType>;
   createdAt?: Resolver<ResolversTypes['Date'], ParentType, ContextType>;
   enabled?: Resolver<ResolversTypes['Boolean'], ParentType, ContextType>;
+  filter?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   id?: Resolver<ResolversTypes['ID'], ParentType, ContextType>;
   name?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
-  query?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   updatedAt?: Resolver<ResolversTypes['Date'], ParentType, ContextType>;
   __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
 };

packages/api/src/generated/schema.graphql

@@ -1160,9 +1160,9 @@ type Rule {
   actions: [RuleAction!]!
   createdAt: Date!
   enabled: Boolean!
+  filter: String!
   id: ID!
   name: String!
-  query: String!
   updatedAt: Date!
 }
 
@@ -1457,9 +1457,9 @@ input SetRuleInput {
   actions: [RuleActionInput!]!
   description: String
   enabled: Boolean!
+  filter: String!
   id: ID
   name: String!
-  query: String!
 }
 
 union SetRuleResult = SetRuleError | SetRuleSuccess

packages/api/src/resolvers/rules/index.ts

@@ -16,7 +16,7 @@ import {
 import {
   createPubSubSubscription,
   deletePubSubSubscription,
-} from '../../utils/pubsub'
+} from '../../datalayer/pubsub'
 
 export const setRuleResolver = authorized<
   SetRuleSuccess,
@@ -32,42 +32,57 @@ export const setRuleResolver = authorized<
     },
   })
 
-  const user = await getRepository(User).findOneBy({ id: claims.uid })
-  if (!user) {
-    return {
-      errorCodes: [SetRuleErrorCode.Unauthorized],
+  try {
+    const user = await getRepository(User).findOneBy({ id: claims.uid })
+    if (!user) {
+      return {
+        errorCodes: [SetRuleErrorCode.Unauthorized],
+      }
     }
-  }
 
-  // create or delete pubsub subscription based on action and enabled state
-  for (const action of input.actions) {
-    const topicName = getPubSubTopicName(action)
-    const subscriptionName = getPubSubSubscriptionName(
-      topicName,
-      user.id,
-      input.name
-    )
-
-    if (input.enabled) {
-      const options = getPubSubSubscriptionOptions(
+    // create or delete pubsub subscription based on action and enabled state
+    for (const action of input.actions) {
+      const topicName = getPubSubTopicName(action)
+      const subscriptionName = getPubSubSubscriptionName(
+        topicName,
         user.id,
-        input.name,
-        input.query,
-        action
+        input.name
       )
-      await createPubSubSubscription(topicName, subscriptionName, options)
-    } else {
-      await deletePubSubSubscription(topicName, subscriptionName)
+
+      if (input.enabled) {
+        const options = getPubSubSubscriptionOptions(
+          user.id,
+          input.name,
+          input.filter,
+          action
+        )
+        await createPubSubSubscription(topicName, subscriptionName, options)
+      } else {
+        await deletePubSubSubscription(topicName, subscriptionName)
+      }
     }
-  }
 
-  const rule = await getRepository(Rule).save({
-    ...input,
-    id: input.id || undefined,
-    user: { id: claims.uid },
-  })
+    const rule = await getRepository(Rule).save({
+      ...input,
+      id: input.id || undefined,
+      user: { id: claims.uid },
+    })
 
-  return {
-    rule,
+    return {
+      rule,
+    }
+  } catch (error) {
+    log.error('Error setting rules', {
+      error,
+      labels: {
+        source: 'resolver',
+        resolver: 'setRulesResolver',
+        uid: claims.uid,
+      },
+    })
+
+    return {
+      errorCodes: [SetRuleErrorCode.BadRequest],
+    }
   }
 })

packages/api/src/schema.ts

@@ -1959,7 +1959,7 @@ const schema = gql`
   type Rule {
     id: ID!
     name: String!
-    query: String!
+    filter: String!
     actions: [RuleAction!]!
     enabled: Boolean!
     createdAt: Date!
@@ -1991,7 +1991,7 @@ const schema = gql`
     id: ID
     name: String!
     description: String
-    query: String!
+    filter: String!
     actions: [RuleActionInput!]!
     enabled: Boolean!
   }

packages/api/src/services/rules.ts

@@ -1,5 +1,6 @@
 import { RuleAction, RuleActionType } from '../generated/graphql'
 import { CreateSubscriptionOptions } from '@google-cloud/pubsub'
+import { env } from '../env'
 
 enum RuleTrigger {
   ON_PAGE_UPDATE,
@@ -42,14 +43,10 @@ export const getPubSubSubscriptionName = (
 export const getPubSubSubscriptionOptions = (
   userId: string,
   ruleName: string,
-  query: string,
+  filter: string,
   action: RuleAction
 ): CreateSubscriptionOptions => {
-  const topic = getPubSubTopicName(action)
-  const name = getPubSubSubscriptionName(topic, userId, ruleName)
   const options: CreateSubscriptionOptions = {
-    name,
-    topic,
     messageRetentionDuration: 60 * 10, // 10 minutes
     expirationPolicy: {
       ttl: null, // never expire
@@ -63,14 +60,24 @@ export const getPubSubSubscriptionOptions = (
         seconds: 600,
       },
     },
-    filter: query,
+    filter,
   }
 
   switch (action.type) {
-    case RuleActionType.SendNotification:
-      options.pushEndpoint = `${process.env
-        .PUSH_NOTIFICATION_ENDPOINT!}?message=${action.params[0]}`
+    case RuleActionType.SendNotification: {
+      const params = action.params
+      if (params.length === 0) {
+        throw new Error('Missing notification messages')
+      }
+
+      options.pushConfig = {
+        pushEndpoint: `${env.queue.notificationEndpoint}/${userId}`,
+        attributes: {
+          messages: JSON.stringify(params),
+        },
+      }
       break
+    }
     // TODO: Add more actions, e.g. RuleActionType.SendEmail
   }
 

packages/api/src/util.ts

@@ -65,6 +65,7 @@ interface BackendEnv {
     reminderTaskHanderUrl: string
     integrationTaskHandlerUrl: string
     textToSpeechTaskHandlerUrl: string
+    notificationEndpoint: string
   }
   fileUpload: {
     gcsUploadBucket: string
@@ -152,6 +153,7 @@ const nullableEnvVars = [
   'AZURE_SPEECH_KEY',
   'AZURE_SPEECH_REGION',
   'GCP_LOCATION',
+  'NOTIFICATION_ENDPOINT',
 ] // 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 */
@@ -237,6 +239,7 @@ export function getEnv(): BackendEnv {
     reminderTaskHanderUrl: parse('REMINDER_TASK_HANDLER_URL'),
     integrationTaskHandlerUrl: parse('INTEGRATION_TASK_HANDLER_URL'),
     textToSpeechTaskHandlerUrl: parse('TEXT_TO_SPEECH_TASK_HANDLER_URL'),
+    notificationEndpoint: parse('NOTIFICATION_ENDPOINT'),
   }
   const imageProxy = {
     url: parse('IMAGE_PROXY_URL'),

packages/api/src/utils/pubsub.ts

@@ -1,39 +0,0 @@
-import { CreateSubscriptionOptions, PubSub } from '@google-cloud/pubsub'
-
-// init pubsub client
-const pubsub = new PubSub()
-
-export const createPubSubSubscription = async (
-  topicName: string,
-  subscriptionName: string,
-  options?: CreateSubscriptionOptions
-) => {
-  const topic = pubsub.topic(topicName)
-  const [exists] = await topic.exists()
-  if (!exists) {
-    await topic.create()
-  }
-
-  const subscription = topic.subscription(subscriptionName)
-  const [subscriptionExists] = await subscription.exists()
-  if (!subscriptionExists) {
-    await subscription.create(options)
-  }
-}
-
-export const deletePubSubSubscription = async (
-  topicName: string,
-  subscriptionName: string
-) => {
-  const topic = pubsub.topic(topicName)
-  const [exists] = await topic.exists()
-  if (!exists) {
-    return
-  }
-
-  const subscription = topic.subscription(subscriptionName)
-  const [subscriptionExists] = await subscription.exists()
-  if (subscriptionExists) {
-    await subscription.delete()
-  }
-}

packages/db/migrations/0099.do.rules.sql

@@ -9,12 +9,12 @@ CREATE TABLE omnivore.rules (
     user_id uuid NOT NULL REFERENCES omnivore.user ON DELETE CASCADE,
     name text NOT NULL,
     description text,
-    query text NOT NULL,
+    filter text NOT NULL,
     actions json NOT NULL, -- array of actions of type {type: 'action_type', params: [action_params]}
     enabled boolean NOT NULL DEFAULT true,
     created_at timestamptz NOT NULL DEFAULT current_timestamp,
     updated_at timestamptz NOT NULL DEFAULT current_timestamp,
-    UNIQUE (user_id, name)
+    UNIQUE (user_id, filter)
 );
 
 CREATE TRIGGER rules_modtime BEFORE UPDATE ON omnivore.rules