From 7556c6da5f617b7f97cbc0cc4651f9a125e3b421 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 26 Jan 2024 17:15:39 +0800 Subject: [PATCH 1/3] move trigger rule into a job --- packages/api/src/jobs/trigger_rule.ts | 138 ++++++++++++++++++++ packages/api/src/resolvers/article/index.ts | 9 +- packages/api/src/services/labels.ts | 40 ++---- packages/api/src/services/rules.ts | 17 ++- packages/api/src/services/user.ts | 26 ++++ packages/api/src/utils/sendNotification.ts | 4 +- 6 files changed, 199 insertions(+), 35 deletions(-) create mode 100644 packages/api/src/jobs/trigger_rule.ts diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts new file mode 100644 index 000000000..3ba72fd4d --- /dev/null +++ b/packages/api/src/jobs/trigger_rule.ts @@ -0,0 +1,138 @@ +import { LibraryItem, LibraryItemState } from '../entity/library_item' +import { Rule, RuleAction, RuleActionType, RuleEventType } from '../entity/rule' +import { addLabelsToLibraryItem } from '../services/labels' +import { + SearchArgs, + searchLibraryItems, + updateLibraryItem, +} from '../services/library_item' +import { findEnabledRules } from '../services/rules' +import { sendPushNotifications } from '../services/user' +import { logger } from '../utils/logger' + +interface Data { + id: string + userId: string + ruleEventType: RuleEventType + subscription: string + image: string + content: string + readingProgressPercent: number +} + +interface RuleActionObj { + userId: string + action: RuleAction + libraryItem: LibraryItem +} + +type RuleActionFunc = (obj: RuleActionObj) => Promise + +const addLabels = async (obj: RuleActionObj) => { + const labelIds = obj.action.params + + return addLabelsToLibraryItem( + labelIds, + obj.libraryItem.id, + obj.userId, + 'system' + ) +} + +const archivePage = async (obj: RuleActionObj) => { + return updateLibraryItem( + obj.libraryItem.id, + { archivedAt: new Date(), state: LibraryItemState.Archived }, + obj.userId + ) +} + +const markPageAsRead = async (obj: RuleActionObj) => { + return updateLibraryItem( + obj.libraryItem.id, + { + readingProgressTopPercent: 100, + readingProgressBottomPercent: 100, + readAt: new Date(), + }, + obj.userId + ) +} + +const sendNotification = async (obj: RuleActionObj) => { + const item = obj.libraryItem + const message = { + title: item.author || item.siteName || 'Omnivore', + body: item.title, + } + + return sendPushNotifications(obj.userId, message, 'rule') +} + +const getRuleAction = (actionType: RuleActionType): RuleActionFunc => { + switch (actionType) { + case RuleActionType.AddLabel: + return addLabels + case RuleActionType.Archive: + return archivePage + case RuleActionType.MarkAsRead: + return markPageAsRead + case RuleActionType.SendNotification: + return sendNotification + } +} + +const triggerActions = async (userId: string, rules: Rule[], data: Data) => { + const actionPromises: Promise[] = [] + + for (const rule of rules) { + const itemId = data.id + const searchArgs: SearchArgs = { + includeContent: false, + includeDeleted: false, + includePending: false, + size: 1, + query: `(${rule.filter}) AND includes:${itemId}`, + } + + const libraryItems = await searchLibraryItems(searchArgs, userId) + if (libraryItems.count === 0) { + logger.info(`No pages found for rule ${rule.id}`) + continue + } + + const libraryItem = libraryItems.libraryItems[0] + + for (const action of rule.actions) { + const actionFunc = getRuleAction(action.type) + const actionObj: RuleActionObj = { + userId, + action, + libraryItem, + } + + actionPromises.push(actionFunc(actionObj)) + } + } + + try { + await Promise.all(actionPromises) + } catch (error) { + logger.error(error) + } +} + +export const triggerRule = async (data: Data) => { + const { userId, ruleEventType } = data + + // get rules by calling api + const rules = await findEnabledRules(userId, ruleEventType) + if (rules.length === 0) { + console.log('No rules found') + return false + } + + await triggerActions(userId, rules, data) + + return true +} diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 03cce990e..9eaf31c59 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -92,6 +92,7 @@ import { import { traceAs } from '../../tracing' import { analytics } from '../../utils/analytics' import { isSiteBlockedForParse } from '../../utils/blocked' +import { authorized } from '../../utils/gql-utils' import { cleanUrl, errorHandler, @@ -102,7 +103,6 @@ import { titleForFilePath, userDataToUser, } from '../../utils/helpers' -import { authorized } from '../../utils/gql-utils' import { contentConverter, getDistillerResult, @@ -908,7 +908,12 @@ export const setFavoriteArticleResolver = authorized< const labels = await findOrCreateLabels([label], uid) // adds Favorites label to item - await addLabelsToLibraryItem(labels, id, uid) + await addLabelsToLibraryItem( + labels.map((l) => l.id), + id, + uid, + 'user' + ) return { success: true, diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts index 46ec8eb0a..564a1e2da 100644 --- a/packages/api/src/services/labels.ts +++ b/packages/api/src/services/labels.ts @@ -5,7 +5,6 @@ import { Label } from '../entity/label' import { createPubSubClient, EntityType, PubsubClient } from '../pubsub' import { authTrx } from '../repository' import { CreateLabelInput, labelRepository } from '../repository/label' -import { libraryItemRepository } from '../repository/library_item' type AddLabelsToLibraryItemEvent = { pageId: string @@ -124,43 +123,28 @@ export const saveLabelsInLibraryItem = async ( } export const addLabelsToLibraryItem = async ( - labels: Label[], + labelIds: string[], libraryItemId: string, userId: string, - source: LabelSource = 'user', - pubsub = createPubSubClient() + source: LabelSource = 'user' ) => { await authTrx( async (tx) => { - const libraryItem = await tx - .withRepository(libraryItemRepository) - .findOneByOrFail({ id: libraryItemId, user: { id: userId } }) - - if (libraryItem.labels) { - labels.push(...libraryItem.labels) - } - - // save new labels - await tx.getRepository(EntityLabel).save( - labels.map((l) => ({ - labelId: l.id, - libraryItemId, - source, - })) + await tx.query( + `INSERT INTO omnivore.entity_labels (label_id, library_item_id, source) + SELECT id, $1, $2 FROM omnivore.labels + WHERE id = ANY($3) + AND NOT EXISTS ( + SELECT 1 FROM omnivore.entity_labels + WHERE label_id = labels.id + AND library_item_id = $1 + )`, + [libraryItemId, source, labelIds] ) }, undefined, userId ) - - if (source === 'user') { - // create pubsub event - await pubsub.entityCreated( - EntityType.LABEL, - { pageId: libraryItemId, labels, source }, - userId - ) - } } export const saveLabelsInHighlight = async ( diff --git a/packages/api/src/services/rules.ts b/packages/api/src/services/rules.ts index 01f86c8ce..5f27a3839 100644 --- a/packages/api/src/services/rules.ts +++ b/packages/api/src/services/rules.ts @@ -1,6 +1,6 @@ -import { ILike } from 'typeorm' -import { Rule, RuleAction } from '../entity/rule' -import { authTrx } from '../repository' +import { ArrayContainedBy, ArrayContains, ILike } from 'typeorm' +import { Rule, RuleAction, RuleEventType } from '../entity/rule' +import { authTrx, getRepository } from '../repository' export const createRule = async ( userId: string, @@ -53,3 +53,14 @@ export const deleteRules = async (userId: string) => { userId ) } + +export const findEnabledRules = async ( + userId: string, + eventType: RuleEventType +) => { + return getRepository(Rule).findBy({ + user: { id: userId }, + enabled: true, + eventTypes: ArrayContainedBy([eventType]), + }) +} diff --git a/packages/api/src/services/user.ts b/packages/api/src/services/user.ts index 1a4dc58d1..1a74500bb 100644 --- a/packages/api/src/services/user.ts +++ b/packages/api/src/services/user.ts @@ -1,9 +1,15 @@ +import { Notification } from 'firebase-admin/messaging' import { DeepPartial, FindOptionsWhere, In } from 'typeorm' import { Profile } from '../entity/profile' import { StatusType, User } from '../entity/user' import { authTrx, getRepository, queryBuilderToRawSql } from '../repository' import { userRepository } from '../repository/user' import { SetClaimsRole } from '../utils/dictionary' +import { + PushNotificationType, + sendMulticastPushNotifications, +} from '../utils/sendNotification' +import { findDeviceTokensByUserId } from './user_device_tokens' export const deleteUser = async (userId: string) => { await authTrx( @@ -120,3 +126,23 @@ export const batchDelete = async (criteria: FindOptionsWhere) => { SetClaimsRole.ADMIN ) } + +export const sendPushNotifications = async ( + userId: string, + notification: Notification, + notificationType: PushNotificationType, + data?: { [key: string]: string } +) => { + const tokens = await findDeviceTokensByUserId(userId) + if (tokens.length === 0) { + throw new Error('No device tokens found') + } + + const message = { + notification, + data, + tokens: tokens.map((token) => token.token), + } + + return sendMulticastPushNotifications(userId, message, notificationType) +} diff --git a/packages/api/src/utils/sendNotification.ts b/packages/api/src/utils/sendNotification.ts index 9c6550653..26adf675c 100644 --- a/packages/api/src/utils/sendNotification.ts +++ b/packages/api/src/utils/sendNotification.ts @@ -60,7 +60,7 @@ export const sendMulticastPushNotifications = async ( }) logger.info('sending multicast message: ', message) - const res = await getMessaging().sendMulticast(message) + const res = await getMessaging().sendEachForMulticast(message) logger.info('send notification result: ', res.responses) return res @@ -75,7 +75,7 @@ export const sendBatchPushNotifications = async ( messages: Message[] ): Promise => { try { - const res = await getMessaging().sendAll(messages) + const res = await getMessaging().sendEach(messages) logger.info(`success count: ${res.successCount}`) return res From e03895c8b72949ba22da43c402bedc040e6c0f89 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 26 Jan 2024 17:40:51 +0800 Subject: [PATCH 2/3] enqueue trigger rule job once page is created or updated --- packages/api/src/jobs/trigger_rule.ts | 28 ++++++++++++++--------- packages/api/src/pubsub.ts | 26 +++++++++++++++++++-- packages/api/src/queue-processor.ts | 3 +++ packages/api/src/services/library_item.ts | 7 +++++- packages/api/src/utils/createTask.ts | 14 ++++++++++++ 5 files changed, 64 insertions(+), 14 deletions(-) diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts index 3ba72fd4d..72d902a90 100644 --- a/packages/api/src/jobs/trigger_rule.ts +++ b/packages/api/src/jobs/trigger_rule.ts @@ -10,14 +10,10 @@ import { findEnabledRules } from '../services/rules' import { sendPushNotifications } from '../services/user' import { logger } from '../utils/logger' -interface Data { - id: string +export interface TriggerRuleJobData { + libraryItemId: string userId: string ruleEventType: RuleEventType - subscription: string - image: string - content: string - readingProgressPercent: number } interface RuleActionObj { @@ -26,6 +22,8 @@ interface RuleActionObj { libraryItem: LibraryItem } +export const TRIGGER_RULE_JOB_NAME = 'trigger-rule' + type RuleActionFunc = (obj: RuleActionObj) => Promise const addLabels = async (obj: RuleActionObj) => { @@ -43,7 +41,9 @@ const archivePage = async (obj: RuleActionObj) => { return updateLibraryItem( obj.libraryItem.id, { archivedAt: new Date(), state: LibraryItemState.Archived }, - obj.userId + obj.userId, + undefined, + true ) } @@ -55,7 +55,9 @@ const markPageAsRead = async (obj: RuleActionObj) => { readingProgressBottomPercent: 100, readAt: new Date(), }, - obj.userId + obj.userId, + undefined, + true ) } @@ -82,11 +84,15 @@ const getRuleAction = (actionType: RuleActionType): RuleActionFunc => { } } -const triggerActions = async (userId: string, rules: Rule[], data: Data) => { +const triggerActions = async ( + userId: string, + rules: Rule[], + data: TriggerRuleJobData +) => { const actionPromises: Promise[] = [] for (const rule of rules) { - const itemId = data.id + const itemId = data.libraryItemId const searchArgs: SearchArgs = { includeContent: false, includeDeleted: false, @@ -122,7 +128,7 @@ const triggerActions = async (userId: string, rules: Rule[], data: Data) => { } } -export const triggerRule = async (data: Data) => { +export const triggerRule = async (data: TriggerRuleJobData) => { const { userId, ruleEventType } = data // get rules by calling api diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts index ea1bcf987..02c35f3d4 100644 --- a/packages/api/src/pubsub.ts +++ b/packages/api/src/pubsub.ts @@ -1,7 +1,9 @@ import { PubSub } from '@google-cloud/pubsub' import express from 'express' +import { RuleEventType } from './entity/rule' import { env } from './env' import { ReportType } from './generated/graphql' +import { enqueueTriggerRuleJob } from './utils/createTask' import { deepDelete } from './utils/helpers' import { buildLogger } from './utils/logger' @@ -41,11 +43,21 @@ export const createPubSubClient = (): PubsubClient => { Buffer.from(JSON.stringify({ userId, email, name, username })) ) }, - entityCreated: ( + entityCreated: async ( type: EntityType, data: T, userId: string ): Promise => { + // queue trigger rule job + if (type === EntityType.PAGE) { + const libraryItemId = (data as T & { id: string }).id + await enqueueTriggerRuleJob({ + userId, + ruleEventType: RuleEventType.PageCreated, + libraryItemId, + }) + } + const cleanData = deepDelete( data as T & Record, [...fieldsToDelete] @@ -56,11 +68,21 @@ export const createPubSubClient = (): PubsubClient => { Buffer.from(JSON.stringify({ type, userId, ...cleanData })) ) }, - entityUpdated: ( + entityUpdated: async ( type: EntityType, data: T, userId: string ): Promise => { + // queue trigger rule job + if (type === EntityType.PAGE) { + const libraryItemId = (data as T & { id: string }).id + await enqueueTriggerRuleJob({ + userId, + ruleEventType: RuleEventType.PageUpdated, + libraryItemId, + }) + } + const cleanData = deepDelete( data as T & Record, [...fieldsToDelete] diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index d99d79f10..cd89e1c69 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -14,6 +14,7 @@ import { savePageJob } from './jobs/save_page' import { updatePDFContentJob } from './jobs/update_pdf_content' import { redisDataSource } from './redis_data_source' import { CustomTypeOrmLogger } from './utils/logger' +import { triggerRule, TRIGGER_RULE_JOB_NAME } from './jobs/trigger_rule' export const QUEUE_NAME = 'omnivore-backend-queue' @@ -122,6 +123,8 @@ const main = async () => { } case THUMBNAIL_JOB: return findThumbnail(job.data) + case TRIGGER_RULE_JOB_NAME: + return triggerRule(job.data) } }, { diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 11a7ee92a..73e9fde77 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -698,7 +698,8 @@ export const updateLibraryItem = async ( id: string, libraryItem: QueryDeepPartialEntity, userId: string, - pubsub = createPubSubClient() + pubsub = createPubSubClient(), + skipPubSub = false ): Promise => { const updatedLibraryItem = await authTrx( async (tx) => { @@ -726,6 +727,10 @@ export const updateLibraryItem = async ( userId ) + if (skipPubSub) { + return updatedLibraryItem + } + await pubsub.entityUpdated>( EntityType.PAGE, { diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts index 94233a454..60e74da5d 100644 --- a/packages/api/src/utils/createTask.ts +++ b/packages/api/src/utils/createTask.ts @@ -17,6 +17,8 @@ import { import { THUMBNAIL_JOB } from '../jobs/find_thumbnail' import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds' import { getBackendQueue } from '../queue-processor' +import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule' +import { getBackendQueue } from '../queue-processor' import { redisDataSource } from '../redis_data_source' import { signFeatureToken } from '../services/features' import { OmnivoreAuthorizationHeader } from './auth' @@ -648,4 +650,16 @@ export const enqueueRssFeedFetch = async ( } } +export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => { + const queue = await getBackendQueue() + if (!queue) { + return undefined + } + + return queue.add(TRIGGER_RULE_JOB_NAME, data, { + removeOnComplete: true, + removeOnFail: true, + }) +} + export default createHttpTaskWithToken From 30fef30384195bae4f07f28ebb0e93623051262f Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 26 Jan 2024 20:32:56 +0800 Subject: [PATCH 3/3] resolve conflicts --- packages/api/src/utils/createTask.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts index 60e74da5d..fc213d33e 100644 --- a/packages/api/src/utils/createTask.ts +++ b/packages/api/src/utils/createTask.ts @@ -16,7 +16,6 @@ import { } from '../generated/graphql' import { THUMBNAIL_JOB } from '../jobs/find_thumbnail' import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds' -import { getBackendQueue } from '../queue-processor' import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule' import { getBackendQueue } from '../queue-processor' import { redisDataSource } from '../redis_data_source'