From c05721c87e572ce314ce4763190ab4c431509937 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Sun, 17 Mar 2024 11:48:58 +0800
Subject: [PATCH 01/14] validate the filter before creating rules

---
 packages/api/src/resolvers/rules/index.ts | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/packages/api/src/resolvers/rules/index.ts b/packages/api/src/resolvers/rules/index.ts
index 49f9c7537..0efa67750 100644
--- a/packages/api/src/resolvers/rules/index.ts
+++ b/packages/api/src/resolvers/rules/index.ts
@@ -15,6 +15,7 @@ import {
 } from '../../generated/graphql'
 import { deleteRule } from '../../services/rules'
 import { authorized } from '../../utils/gql-utils'
+import { parseSearchQuery } from '../../utils/search'
 
 export const setRuleResolver = authorized<
   SetRuleSuccess,
@@ -22,6 +23,9 @@ export const setRuleResolver = authorized<
   MutationSetRuleArgs
 >(async (_, { input }, { authTrx, uid, log }) => {
   try {
+    // validate filter
+    parseSearchQuery(input.filter)
+
     const rule = await authTrx((t) =>
       t.getRepository(Rule).save({
         ...input,

From 2fc6fd6e9365f5523955f3a2c8229635750cd320 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Sun, 17 Mar 2024 11:56:06 +0800
Subject: [PATCH 02/14] Show error message if failed to creating rules

---
 packages/web/pages/settings/rules.tsx | 20 +++++++++++++-------
 1 file changed, 13 insertions(+), 7 deletions(-)

diff --git a/packages/web/pages/settings/rules.tsx b/packages/web/pages/settings/rules.tsx
index dd0d52448..d38ee76f1 100644
--- a/packages/web/pages/settings/rules.tsx
+++ b/packages/web/pages/settings/rules.tsx
@@ -32,13 +32,19 @@ const CreateRuleModal = (props: CreateRuleModalProps): JSX.Element => {
     const name = form.getFieldValue('name')
     const filter = form.getFieldValue('filter')
     const eventTypes = form.getFieldValue('eventTypes')
-    await setRuleMutation({
-      name,
-      filter,
-      actions: [],
-      enabled: true,
-      eventTypes,
-    })
+    try {
+      await setRuleMutation({
+        name,
+        filter,
+        actions: [],
+        enabled: true,
+        eventTypes,
+      })
+    } catch (error) {
+      showErrorToast('Error creating rule')
+      return
+    }
+
     form.resetFields()
     props.setIsModalOpen(false)
     props.revalidate()

From 9af4235233c17d7cfd2e059330165fe2c811bb8f Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Sun, 17 Mar 2024 12:38:23 +0800
Subject: [PATCH 03/14] skip failed to parse rules and show the timestamp in
 the UI

---
 packages/api/src/entity/rule.ts               |  3 +++
 packages/api/src/generated/graphql.ts         |  2 ++
 packages/api/src/generated/schema.graphql     |  1 +
 packages/api/src/jobs/trigger_rule.ts         | 25 ++++++++++++++-----
 packages/api/src/schema.ts                    |  1 +
 packages/api/src/services/rules.ts            | 14 ++++++++++-
 .../0168.do.add_failed_at_to_rule.sql         |  9 +++++++
 .../0168.undo.add_failed_at_to_rule.sql       |  9 +++++++
 .../networking/queries/useGetRulesQuery.tsx   |  2 ++
 packages/web/pages/settings/rules.tsx         |  6 +++++
 10 files changed, 65 insertions(+), 7 deletions(-)
 create mode 100755 packages/db/migrations/0168.do.add_failed_at_to_rule.sql
 create mode 100755 packages/db/migrations/0168.undo.add_failed_at_to_rule.sql

diff --git a/packages/api/src/entity/rule.ts b/packages/api/src/entity/rule.ts
index 0147cbd19..51c63b972 100644
--- a/packages/api/src/entity/rule.ts
+++ b/packages/api/src/entity/rule.ts
@@ -59,4 +59,7 @@ export class Rule {
 
   @UpdateDateColumn({ default: () => 'CURRENT_TIMESTAMP' })
   updatedAt!: Date
+
+  @Column('timestamptz')
+  failedAt?: Date
 }
diff --git a/packages/api/src/generated/graphql.ts b/packages/api/src/generated/graphql.ts
index 4e9c0735d..a2a4d7ef8 100644
--- a/packages/api/src/generated/graphql.ts
+++ b/packages/api/src/generated/graphql.ts
@@ -2434,6 +2434,7 @@ export type Rule = {
   createdAt: Scalars['Date'];
   enabled: Scalars['Boolean'];
   eventTypes: Array<RuleEventType>;
+  failedAt?: Maybe<Scalars['Date']>;
   filter: Scalars['String'];
   id: Scalars['ID'];
   name: Scalars['String'];
@@ -6322,6 +6323,7 @@ export type RuleResolvers<ContextType = ResolverContext, ParentType extends Reso
   createdAt?: Resolver<ResolversTypes['Date'], ParentType, ContextType>;
   enabled?: Resolver<ResolversTypes['Boolean'], ParentType, ContextType>;
   eventTypes?: Resolver<Array<ResolversTypes['RuleEventType']>, ParentType, ContextType>;
+  failedAt?: Resolver<Maybe<ResolversTypes['Date']>, ParentType, ContextType>;
   filter?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   id?: Resolver<ResolversTypes['ID'], ParentType, ContextType>;
   name?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
diff --git a/packages/api/src/generated/schema.graphql b/packages/api/src/generated/schema.graphql
index 60ad805cd..ab001939f 100644
--- a/packages/api/src/generated/schema.graphql
+++ b/packages/api/src/generated/schema.graphql
@@ -1825,6 +1825,7 @@ type Rule {
   createdAt: Date!
   enabled: Boolean!
   eventTypes: [RuleEventType!]!
+  failedAt: Date
   filter: String!
   id: ID!
   name: String!
diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts
index 260b7868f..3247e3fa9 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -8,7 +8,7 @@ import {
   softDeleteLibraryItem,
   updateLibraryItem,
 } from '../services/library_item'
-import { findEnabledRules } from '../services/rules'
+import { findEnabledRules, markRuleAsFailed } from '../services/rules'
 import { sendPushNotifications } from '../services/user'
 import { logger } from '../utils/logger'
 
@@ -112,14 +112,27 @@ const triggerActions = async (
       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}`)
+    let libraryItem: LibraryItem
+
+    try {
+      const { libraryItems, count } = await searchLibraryItems(
+        searchArgs,
+        userId
+      )
+      if (count === 0) {
+        logger.info(`No pages found for rule ${rule.id}`)
+        continue
+      }
+
+      libraryItem = libraryItems[0]
+    } catch (error) {
+      // failed to search for library items, mark rule as failed
+      logger.error('Error parsing filter in rules', error)
+      await markRuleAsFailed(rule.id, userId)
+
       continue
     }
 
-    const libraryItem = libraryItems.libraryItems[0]
-
     for (const action of rule.actions) {
       const actionFunc = getRuleAction(action.type)
       const actionObj: RuleActionObj = {
diff --git a/packages/api/src/schema.ts b/packages/api/src/schema.ts
index 6c20fb51f..e7eb46a62 100755
--- a/packages/api/src/schema.ts
+++ b/packages/api/src/schema.ts
@@ -2152,6 +2152,7 @@ const schema = gql`
     createdAt: Date!
     updatedAt: Date
     eventTypes: [RuleEventType!]!
+    failedAt: Date
   }
 
   type RuleAction {
diff --git a/packages/api/src/services/rules.ts b/packages/api/src/services/rules.ts
index bf3637be7..6393a8c40 100644
--- a/packages/api/src/services/rules.ts
+++ b/packages/api/src/services/rules.ts
@@ -1,4 +1,4 @@
-import { ArrayContains, ILike } from 'typeorm'
+import { ArrayContains, ILike, IsNull, Not } from 'typeorm'
 import { Rule, RuleAction, RuleEventType } from '../entity/rule'
 import { authTrx, getRepository } from '../repository'
 
@@ -62,5 +62,17 @@ export const findEnabledRules = async (
     user: { id: userId },
     enabled: true,
     eventTypes: ArrayContains([eventType]),
+    failedAt: IsNull(), // only rules that have not failed
   })
 }
+
+export const markRuleAsFailed = async (id: string, userId: string) => {
+  return authTrx(
+    (t) =>
+      t.getRepository(Rule).update(id, {
+        failedAt: new Date(),
+      }),
+    undefined,
+    userId
+  )
+}
diff --git a/packages/db/migrations/0168.do.add_failed_at_to_rule.sql b/packages/db/migrations/0168.do.add_failed_at_to_rule.sql
new file mode 100755
index 000000000..6f77f166c
--- /dev/null
+++ b/packages/db/migrations/0168.do.add_failed_at_to_rule.sql
@@ -0,0 +1,9 @@
+-- Type: DO
+-- Name: add_failed_at_to_rule
+-- Description: Add failed_at column to rules table
+
+BEGIN;
+
+ALTER TABLE omnivore.rules ADD COLUMN failed_at timestamptz;
+
+COMMIT;
diff --git a/packages/db/migrations/0168.undo.add_failed_at_to_rule.sql b/packages/db/migrations/0168.undo.add_failed_at_to_rule.sql
new file mode 100755
index 000000000..de2fc769d
--- /dev/null
+++ b/packages/db/migrations/0168.undo.add_failed_at_to_rule.sql
@@ -0,0 +1,9 @@
+-- Type: UNDO
+-- Name: add_failed_at_to_rule
+-- Description: Add failed_at column to rules table
+
+BEGIN;
+
+ALTER TABLE omnivore.rules DROP COLUMN failed_at;
+
+COMMIT;
diff --git a/packages/web/lib/networking/queries/useGetRulesQuery.tsx b/packages/web/lib/networking/queries/useGetRulesQuery.tsx
index f7619cdcf..2da718072 100644
--- a/packages/web/lib/networking/queries/useGetRulesQuery.tsx
+++ b/packages/web/lib/networking/queries/useGetRulesQuery.tsx
@@ -29,6 +29,7 @@ export interface Rule {
   createdAt: Date
   updatedAt: Date
   eventTypes: RuleEventType[]
+  failedAt?: Date
 }
 
 interface RulesQueryResponse {
@@ -62,6 +63,7 @@ export function useGetRulesQuery(): RulesQueryResponse {
             createdAt
             updatedAt
             eventTypes
+            failedAt
           }
         }
         ... on RulesError {
diff --git a/packages/web/pages/settings/rules.tsx b/packages/web/pages/settings/rules.tsx
index d38ee76f1..5cf3e1636 100644
--- a/packages/web/pages/settings/rules.tsx
+++ b/packages/web/pages/settings/rules.tsx
@@ -235,6 +235,7 @@ export default function Rules(): JSX.Element {
         filter: rule.filter,
         actions: rule.actions,
         eventTypes: rule.eventTypes,
+        failedAt: rule.failedAt,
       }
     })
   }, [rules])
@@ -318,6 +319,11 @@ export default function Rules(): JSX.Element {
         </>
       ),
     },
+    {
+      title: 'Failed At',
+      dataIndex: 'failedAt',
+      key: 'failedAt',
+    },
     {
       title: '',
       key: 'tools',

From 208a5895ef165a45a6cfe1120343e99c15d313b4 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 11:06:24 +0800
Subject: [PATCH 04/14] remove unnecessary fields from item events

---
 packages/api/src/pubsub.ts                    | 37 ++------------
 .../api/src/services/integrations/notion.ts   |  2 +-
 .../api/src/services/integrations/readwise.ts |  4 +-
 packages/api/src/services/library_item.ts     | 49 +++++++++++++++----
 packages/api/src/utils/helpers.ts             |  4 --
 5 files changed, 47 insertions(+), 49 deletions(-)

diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts
index f13bfe5e4..53f95878a 100644
--- a/packages/api/src/pubsub.ts
+++ b/packages/api/src/pubsub.ts
@@ -3,6 +3,7 @@ import express from 'express'
 import { RuleEventType } from './entity/rule'
 import { env } from './env'
 import { ReportType } from './generated/graphql'
+import { FeatureName, findFeatureByName } from './services/features'
 import { Merge } from './util'
 import {
   enqueueAISummarizeJob,
@@ -11,13 +12,7 @@ import {
   enqueueTriggerRuleJob,
   enqueueWebhookJob,
 } from './utils/createTask'
-import { deepDelete } from './utils/helpers'
 import { buildLogger } from './utils/logger'
-import {
-  FeatureName,
-  findFeatureByName,
-  getFeatureName,
-} from './services/features'
 import { processYouTubeVideo } from './jobs/process-youtube-video'
 
 const logger = buildLogger('pubsub')
@@ -42,8 +37,6 @@ const isYouTubeVideoURL = (url: string | undefined): boolean => {
 }
 
 export const createPubSubClient = (): PubsubClient => {
-  const fieldsToDelete = ['user'] as const
-
   const publish = (topicName: string, msg: Buffer): Promise<void> => {
     if (env.dev.isLocal) {
       logger.info(`Publishing ${topicName}: ${msg.toString()}`)
@@ -93,11 +86,6 @@ export const createPubSubClient = (): PubsubClient => {
         libraryItemIds: [libraryItemId],
       })
 
-      const cleanData = deepDelete(
-        data as EntityData<T> & Record<typeof fieldsToDelete[number], unknown>,
-        [...fieldsToDelete]
-      )
-
       await enqueueWebhookJob({
         userId,
         type,
@@ -121,11 +109,6 @@ export const createPubSubClient = (): PubsubClient => {
           libraryItemId,
         })
       }
-
-      return publish(
-        'entityCreated',
-        Buffer.from(JSON.stringify({ type, userId, ...cleanData }))
-      )
     },
     entityUpdated: async <T extends Record<string, any>>(
       type: EntityType,
@@ -148,32 +131,20 @@ export const createPubSubClient = (): PubsubClient => {
         libraryItemIds: [libraryItemId],
       })
 
-      const cleanData = deepDelete(
-        data as EntityData<T> & Record<typeof fieldsToDelete[number], unknown>,
-        [...fieldsToDelete]
-      )
-
       await enqueueWebhookJob({
         userId,
         type,
         action: 'updated',
         data,
       })
-
-      return publish(
-        'entityUpdated',
-        Buffer.from(JSON.stringify({ type, userId, ...cleanData }))
-      )
     },
-    entityDeleted: (
+    entityDeleted: async (
       type: EntityType,
       id: string,
       userId: string
     ): Promise<void> => {
-      return publish(
-        'entityDeleted',
-        Buffer.from(JSON.stringify({ type, id, userId }))
-      )
+      logger.info(`entityDeleted: ${type} ${id} ${userId}`)
+      await Promise.resolve()
     },
     reportSubmitted: (
       submitterId: string,
diff --git a/packages/api/src/services/integrations/notion.ts b/packages/api/src/services/integrations/notion.ts
index 4d823e031..de517dca6 100644
--- a/packages/api/src/services/integrations/notion.ts
+++ b/packages/api/src/services/integrations/notion.ts
@@ -5,8 +5,8 @@ import { Integration } from '../../entity/integration'
 import { LibraryItem } from '../../entity/library_item'
 import { env } from '../../env'
 import { Merge } from '../../util'
-import { highlightUrl } from '../../utils/helpers'
 import { logger } from '../../utils/logger'
+import { getHighlightUrl } from '../highlights'
 import { IntegrationClient } from './integration'
 
 type AnnotationColor =
diff --git a/packages/api/src/services/integrations/readwise.ts b/packages/api/src/services/integrations/readwise.ts
index dfc5b43db..0c204b630 100644
--- a/packages/api/src/services/integrations/readwise.ts
+++ b/packages/api/src/services/integrations/readwise.ts
@@ -1,7 +1,7 @@
 import axios from 'axios'
 import { LibraryItem } from '../../entity/library_item'
-import { highlightUrl } from '../../utils/helpers'
 import { logger } from '../../utils/logger'
+import { getHighlightUrl } from '../highlights'
 import { IntegrationClient } from './integration'
 
 interface ReadwiseHighlight {
@@ -98,7 +98,7 @@ export class ReadwiseClient implements IntegrationClient {
           text: highlight.quote,
           title: item.title,
           author: item.author || undefined,
-          highlight_url: highlightUrl(item.slug, highlight.id),
+          highlight_url: getHighlightUrl(item.slug, highlight.id),
           highlighted_at: new Date(highlight.createdAt).toISOString(),
           category,
           image_url: item.thumbnail || undefined,
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 52d44f7ff..7d13e2fa8 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -29,6 +29,26 @@ import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
 import { addLabelsToLibraryItem } from './labels'
 
+type ItemEvent = { libraryItemId: string; userId: string }
+type IgnoredFields =
+  | 'user'
+  | 'uploadFile'
+  | 'labelNames'
+  | 'highlightAnnotations'
+  | 'previewContentType'
+  | 'links'
+  | 'recommenderNames'
+  | 'textContentHash'
+
+type CreateItemEvent = Merge<
+  Omit<DeepPartial<LibraryItem>, IgnoredFields>,
+  ItemEvent
+>
+type UpdateItemEvent = Merge<
+  Omit<QueryDeepPartialEntity<LibraryItem>, IgnoredFields>,
+  ItemEvent
+>
+
 enum ReadFilter {
   ALL = 'all',
   READ = 'read',
@@ -833,19 +853,32 @@ export const updateLibraryItem = async (
     userId
   )
 
-  if (skipPubSub) {
+  if (skipPubSub || libraryItem.state === LibraryItemState.Processing) {
     return updatedLibraryItem
   }
 
-  await pubsub.entityUpdated<QueryDeepPartialEntity<LibraryItem>>(
+  if (libraryItem.state === LibraryItemState.Succeeded) {
+    // send create event if the item was created
+    await pubsub.entityCreated<CreateItemEvent>(
+      EntityType.PAGE,
+      {
+        ...updatedLibraryItem,
+        libraryItemId: id,
+        userId,
+      },
+      userId
+    )
+
+    return updatedLibraryItem
+  }
+
+  await pubsub.entityUpdated<UpdateItemEvent>(
     EntityType.PAGE,
     {
       ...libraryItem,
       id,
       libraryItemId: id,
-      // don't send original content and readable content
-      originalContent: undefined,
-      readableContent: undefined,
+      userId,
     },
     userId
   )
@@ -999,14 +1032,12 @@ export const createOrUpdateLibraryItem = async (
     return newLibraryItem
   }
 
-  await pubsub.entityCreated<DeepPartial<LibraryItem>>(
+  await pubsub.entityCreated<CreateItemEvent>(
     EntityType.PAGE,
     {
       ...newLibraryItem,
       libraryItemId: newLibraryItem.id,
-      // don't send original content and readable content
-      originalContent: undefined,
-      readableContent: undefined,
+      userId,
     },
     userId
   )
diff --git a/packages/api/src/utils/helpers.ts b/packages/api/src/utils/helpers.ts
index 9e3e18cd3..d2fec6520 100644
--- a/packages/api/src/utils/helpers.ts
+++ b/packages/api/src/utils/helpers.ts
@@ -10,7 +10,6 @@ import { Highlight as HighlightData } from '../entity/highlight'
 import { LibraryItem, LibraryItemState } from '../entity/library_item'
 import { Recommendation as RecommendationData } from '../entity/recommendation'
 import { RegistrationType, User } from '../entity/user'
-import { env } from '../env'
 import {
   Article,
   ArticleSavingRequest,
@@ -404,6 +403,3 @@ export const setRecentlySavedItemInRedis = async (
     })
   }
 }
-
-export const highlightUrl = (slug: string, highlightId: string): string =>
-  `${env.client.url}/me/${slug}#${highlightId}`

From a49366b4cc249bec2f2eec55fcc47f53ab8891c8 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 11:24:16 +0800
Subject: [PATCH 05/14] remove unnecessary fields from item events

---
 packages/api/src/jobs/trigger_rule.ts     |  1 +
 packages/api/src/pubsub.ts                | 31 ++++++++---------
 packages/api/src/services/highlights.ts   | 11 +++---
 packages/api/src/services/labels.ts       | 12 +++----
 packages/api/src/services/library_item.ts | 41 +++++++----------------
 5 files changed, 41 insertions(+), 55 deletions(-)

diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts
index 3247e3fa9..a75fb75f5 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -16,6 +16,7 @@ export interface TriggerRuleJobData {
   libraryItemId: string
   userId: string
   ruleEventType: RuleEventType
+  data: unknown
 }
 
 interface RuleActionObj {
diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts
index 53f95878a..b210fbfa5 100644
--- a/packages/api/src/pubsub.ts
+++ b/packages/api/src/pubsub.ts
@@ -4,7 +4,6 @@ import { RuleEventType } from './entity/rule'
 import { env } from './env'
 import { ReportType } from './generated/graphql'
 import { FeatureName, findFeatureByName } from './services/features'
-import { Merge } from './util'
 import {
   enqueueAISummarizeJob,
   enqueueExportItem,
@@ -19,11 +18,6 @@ const logger = buildLogger('pubsub')
 
 const client = new PubSub()
 
-type EntityData<T extends Record<string, any>> = Merge<
-  T,
-  { libraryItemId: string }
->
-
 const isYouTubeVideoURL = (url: string | undefined): boolean => {
   if (!url) {
     return false
@@ -68,16 +62,17 @@ export const createPubSubClient = (): PubsubClient => {
     },
     entityCreated: async <T extends Record<string, any>>(
       type: EntityType,
-      data: EntityData<T>,
-      userId: string
+      data: T,
+      userId: string,
+      libraryItemId: string
     ): Promise<void> => {
-      const libraryItemId = data.libraryItemId
       // queue trigger rule job
       if (type === EntityType.PAGE) {
         await enqueueTriggerRuleJob({
           userId,
           ruleEventType: RuleEventType.PageCreated,
           libraryItemId,
+          data,
         })
       }
       // queue export item job
@@ -112,17 +107,17 @@ export const createPubSubClient = (): PubsubClient => {
     },
     entityUpdated: async <T extends Record<string, any>>(
       type: EntityType,
-      data: EntityData<T>,
-      userId: string
+      data: T,
+      userId: string,
+      libraryItemId: string
     ): Promise<void> => {
-      const libraryItemId = data.libraryItemId
-
       // queue trigger rule job
       if (type === EntityType.PAGE) {
         await enqueueTriggerRuleJob({
           userId,
           ruleEventType: RuleEventType.PageUpdated,
           libraryItemId,
+          data,
         })
       }
       // queue export item job
@@ -178,13 +173,15 @@ export interface PubsubClient {
   ) => Promise<void>
   entityCreated: <T extends Record<string, any>>(
     type: EntityType,
-    data: EntityData<T>,
-    userId: string
+    data: T,
+    userId: string,
+    libraryItemId: string
   ) => Promise<void>
   entityUpdated: <T extends Record<string, any>>(
     type: EntityType,
-    data: EntityData<T>,
-    userId: string
+    data: T,
+    userId: string,
+    libraryItemId: string
   ) => Promise<void>
   entityDeleted: (type: EntityType, id: string, userId: string) => Promise<void>
   reportSubmitted(
diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts
index 5883fb7cb..babf4d4a4 100644
--- a/packages/api/src/services/highlights.ts
+++ b/packages/api/src/services/highlights.ts
@@ -59,7 +59,8 @@ export const createHighlight = async (
   await pubsub.entityCreated<CreateHighlightEvent>(
     EntityType.HIGHLIGHT,
     { ...newHighlight, pageId: libraryItemId },
-    userId
+    userId,
+    libraryItemId
   )
 
   await enqueueUpdateHighlight({
@@ -106,7 +107,8 @@ export const mergeHighlights = async (
   await pubsub.entityCreated<CreateHighlightEvent>(
     EntityType.HIGHLIGHT,
     { ...newHighlight, pageId: libraryItemId },
-    userId
+    userId,
+    libraryItemId
   )
 
   await enqueueUpdateHighlight({
@@ -139,8 +141,9 @@ export const updateHighlight = async (
   const libraryItemId = updatedHighlight.libraryItem.id
   await pubsub.entityUpdated<UpdateHighlightEvent>(
     EntityType.HIGHLIGHT,
-    { ...highlight, id: highlightId, pageId: libraryItemId, libraryItemId },
-    userId
+    { ...highlight, id: highlightId, pageId: libraryItemId },
+    userId,
+    libraryItemId
   )
 
   await enqueueUpdateHighlight({
diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index 285338745..f4bae13b7 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -11,13 +11,11 @@ import { findHighlightById } from './highlights'
 import { findLibraryItemIdsByLabelId } from './library_item'
 
 type AddLabelsToLibraryItemEvent = {
-  libraryItemId: string
   pageId: string
   labels: DeepPartial<Label>[]
   source?: LabelSource
 }
 type AddLabelsToHighlightEvent = {
-  libraryItemId: string
   highlightId: string
   labels: DeepPartial<Label>[]
 }
@@ -148,8 +146,9 @@ export const saveLabelsInLibraryItem = async (
     // create pubsub event
     await pubsub.entityCreated<AddLabelsToLibraryItemEvent>(
       EntityType.LABEL,
-      { pageId: libraryItemId, labels, source, libraryItemId },
-      userId
+      { pageId: libraryItemId, labels, source },
+      userId,
+      libraryItemId
     )
   }
 
@@ -218,8 +217,9 @@ export const saveLabelsInHighlight = async (
   // create pubsub event
   await pubsub.entityCreated<AddLabelsToHighlightEvent>(
     EntityType.LABEL,
-    { highlightId, labels, libraryItemId },
-    userId
+    { highlightId, labels },
+    userId,
+    libraryItemId
   )
 
   // update labels in library item
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 7d13e2fa8..b2f2fa049 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -29,7 +29,6 @@ import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
 import { addLabelsToLibraryItem } from './labels'
 
-type ItemEvent = { libraryItemId: string; userId: string }
 type IgnoredFields =
   | 'user'
   | 'uploadFile'
@@ -39,14 +38,10 @@ type IgnoredFields =
   | 'links'
   | 'recommenderNames'
   | 'textContentHash'
-
-type CreateItemEvent = Merge<
-  Omit<DeepPartial<LibraryItem>, IgnoredFields>,
-  ItemEvent
->
-type UpdateItemEvent = Merge<
-  Omit<QueryDeepPartialEntity<LibraryItem>, IgnoredFields>,
-  ItemEvent
+export type CreateItemEvent = Omit<DeepPartial<LibraryItem>, IgnoredFields>
+export type UpdateItemEvent = Omit<
+  QueryDeepPartialEntity<LibraryItem>,
+  IgnoredFields
 >
 
 enum ReadFilter {
@@ -861,12 +856,9 @@ export const updateLibraryItem = async (
     // send create event if the item was created
     await pubsub.entityCreated<CreateItemEvent>(
       EntityType.PAGE,
-      {
-        ...updatedLibraryItem,
-        libraryItemId: id,
-        userId,
-      },
-      userId
+      updatedLibraryItem,
+      userId,
+      id
     )
 
     return updatedLibraryItem
@@ -874,13 +866,9 @@ export const updateLibraryItem = async (
 
   await pubsub.entityUpdated<UpdateItemEvent>(
     EntityType.PAGE,
-    {
-      ...libraryItem,
-      id,
-      libraryItemId: id,
-      userId,
-    },
-    userId
+    libraryItem,
+    userId,
+    id
   )
 
   return updatedLibraryItem
@@ -1034,12 +1022,9 @@ export const createOrUpdateLibraryItem = async (
 
   await pubsub.entityCreated<CreateItemEvent>(
     EntityType.PAGE,
-    {
-      ...newLibraryItem,
-      libraryItemId: newLibraryItem.id,
-      userId,
-    },
-    userId
+    newLibraryItem,
+    userId,
+    newLibraryItem.id
   )
 
   return newLibraryItem

From e87e6dd102b8c9f0a8f0d5cbe5f215eabb7d8fdd Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 12:04:17 +0800
Subject: [PATCH 06/14] filter data locally

---
 packages/api/src/jobs/trigger_rule.ts | 68 +++++++++++++--------------
 1 file changed, 32 insertions(+), 36 deletions(-)

diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts
index a75fb75f5..d00d6ded8 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -1,28 +1,32 @@
+import { filter } from '@omnivore/liqe'
 import { ReadingProgressDataSource } from '../datasources/reading_progress_data_source'
-import { LibraryItem, LibraryItemState } from '../entity/library_item'
+import { LibraryItemState } from '../entity/library_item'
 import { Rule, RuleAction, RuleActionType, RuleEventType } from '../entity/rule'
 import { addLabelsToLibraryItem } from '../services/labels'
 import {
-  SearchArgs,
-  searchLibraryItems,
+  CreateItemEvent,
   softDeleteLibraryItem,
+  UpdateItemEvent,
   updateLibraryItem,
 } from '../services/library_item'
 import { findEnabledRules, markRuleAsFailed } from '../services/rules'
 import { sendPushNotifications } from '../services/user'
 import { logger } from '../utils/logger'
+import { parseSearchQuery } from '../utils/search'
 
+type Data = CreateItemEvent | UpdateItemEvent
 export interface TriggerRuleJobData {
   libraryItemId: string
   userId: string
   ruleEventType: RuleEventType
-  data: unknown
+  data: Data
 }
 
 interface RuleActionObj {
+  libraryItemId: string
   userId: string
   action: RuleAction
-  libraryItem: LibraryItem
+  data: Data
 }
 type RuleActionFunc = (obj: RuleActionObj) => Promise<unknown>
 
@@ -34,19 +38,19 @@ const addLabels = async (obj: RuleActionObj) => {
 
   return addLabelsToLibraryItem(
     labelIds,
-    obj.libraryItem.id,
+    obj.libraryItemId,
     obj.userId,
     'system'
   )
 }
 
 const deleteLibraryItem = async (obj: RuleActionObj) => {
-  return softDeleteLibraryItem(obj.libraryItem.id, obj.userId)
+  return softDeleteLibraryItem(obj.libraryItemId, obj.userId)
 }
 
 const archivePage = async (obj: RuleActionObj) => {
   return updateLibraryItem(
-    obj.libraryItem.id,
+    obj.libraryItemId,
     { archivedAt: new Date(), state: LibraryItemState.Archived },
     obj.userId,
     undefined,
@@ -57,7 +61,7 @@ const archivePage = async (obj: RuleActionObj) => {
 const markPageAsRead = async (obj: RuleActionObj) => {
   return readingProgressDataSource.updateReadingProgress(
     obj.userId,
-    obj.libraryItem.id,
+    obj.libraryItemId,
     {
       readingProgressPercent: 100,
       readingProgressTopPercent: 100,
@@ -67,15 +71,15 @@ const markPageAsRead = async (obj: RuleActionObj) => {
 }
 
 const sendNotification = async (obj: RuleActionObj) => {
-  const item = obj.libraryItem
+  const item = obj.data
   const message = {
-    title: item.author || item.siteName || 'Omnivore',
-    body: item.title,
+    title: item.author?.toString() || item.siteName?.toString() || 'Omnivore',
+    body: item.title?.toString(),
     image: item.thumbnail,
   }
   const data = {
-    folder: item.folder,
-    libraryItemId: item.id,
+    folder: item.folder?.toString() || 'inbox',
+    libraryItemId: obj.libraryItemId,
   }
 
   return sendPushNotifications(obj.userId, message, 'rule', data)
@@ -97,35 +101,26 @@ const getRuleAction = (actionType: RuleActionType): RuleActionFunc => {
 }
 
 const triggerActions = async (
+  libraryItemId: string,
   userId: string,
   rules: Rule[],
-  data: TriggerRuleJobData
+  data: Data
 ) => {
   const actionPromises: Promise<unknown>[] = []
 
   for (const rule of rules) {
-    const itemId = data.libraryItemId
-    const searchArgs: SearchArgs = {
-      includeContent: false,
-      includeDeleted: false,
-      includePending: false,
-      size: 1,
-      query: `(${rule.filter}) AND includes:${itemId}`,
-    }
-
-    let libraryItem: LibraryItem
+    let filteredData: Data
 
     try {
-      const { libraryItems, count } = await searchLibraryItems(
-        searchArgs,
-        userId
-      )
-      if (count === 0) {
-        logger.info(`No pages found for rule ${rule.id}`)
+      const ast = parseSearchQuery(rule.filter)
+      // filter library item by rule filter
+      const results = filter(ast, [data])
+      if (results.length === 0) {
+        logger.info(`No items found for rule ${rule.id}`)
         continue
       }
 
-      libraryItem = libraryItems[0]
+      filteredData = results[0]
     } catch (error) {
       // failed to search for library items, mark rule as failed
       logger.error('Error parsing filter in rules', error)
@@ -137,9 +132,10 @@ const triggerActions = async (
     for (const action of rule.actions) {
       const actionFunc = getRuleAction(action.type)
       const actionObj: RuleActionObj = {
+        libraryItemId,
         userId,
         action,
-        libraryItem,
+        data: filteredData,
       }
 
       actionPromises.push(actionFunc(actionObj))
@@ -153,8 +149,8 @@ const triggerActions = async (
   }
 }
 
-export const triggerRule = async (data: TriggerRuleJobData) => {
-  const { userId, ruleEventType } = data
+export const triggerRule = async (jobData: TriggerRuleJobData) => {
+  const { userId, ruleEventType, data, libraryItemId } = jobData
 
   // get rules by calling api
   const rules = await findEnabledRules(userId, ruleEventType)
@@ -163,7 +159,7 @@ export const triggerRule = async (data: TriggerRuleJobData) => {
     return false
   }
 
-  await triggerActions(userId, rules, data)
+  await triggerActions(libraryItemId, userId, rules, data)
 
   return true
 }

From f1eff130af21388aff236b902ab8bca0492b7672 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 17:23:42 +0800
Subject: [PATCH 07/14] filter data locally with liqe

---
 packages/api/src/jobs/trigger_rule.ts     |  12 +-
 packages/api/src/services/library_item.ts | 299 +++++++++++++++++++++-
 2 files changed, 301 insertions(+), 10 deletions(-)

diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts
index d00d6ded8..f94e3c37a 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -4,9 +4,8 @@ import { LibraryItemState } from '../entity/library_item'
 import { Rule, RuleAction, RuleActionType, RuleEventType } from '../entity/rule'
 import { addLabelsToLibraryItem } from '../services/labels'
 import {
-  CreateItemEvent,
+  ItemEvent,
   softDeleteLibraryItem,
-  UpdateItemEvent,
   updateLibraryItem,
 } from '../services/library_item'
 import { findEnabledRules, markRuleAsFailed } from '../services/rules'
@@ -14,19 +13,18 @@ import { sendPushNotifications } from '../services/user'
 import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
 
-type Data = CreateItemEvent | UpdateItemEvent
 export interface TriggerRuleJobData {
   libraryItemId: string
   userId: string
   ruleEventType: RuleEventType
-  data: Data
+  data: ItemEvent
 }
 
 interface RuleActionObj {
   libraryItemId: string
   userId: string
   action: RuleAction
-  data: Data
+  data: ItemEvent
 }
 type RuleActionFunc = (obj: RuleActionObj) => Promise<unknown>
 
@@ -104,12 +102,12 @@ const triggerActions = async (
   libraryItemId: string,
   userId: string,
   rules: Rule[],
-  data: Data
+  data: ItemEvent
 ) => {
   const actionPromises: Promise<unknown>[] = []
 
   for (const rule of rules) {
-    let filteredData: Data
+    let filteredData: ItemEvent
 
     try {
       const ast = parseSearchQuery(rule.filter)
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index b2f2fa049..dcea18b29 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1,4 +1,5 @@
 import { ExpressionToken, LiqeQuery } from '@omnivore/liqe'
+import { camelCase } from 'lodash'
 import { DateTime } from 'luxon'
 import {
   DeepPartial,
@@ -32,12 +33,10 @@ import { addLabelsToLibraryItem } from './labels'
 type IgnoredFields =
   | 'user'
   | 'uploadFile'
-  | 'labelNames'
-  | 'highlightAnnotations'
   | 'previewContentType'
   | 'links'
-  | 'recommenderNames'
   | 'textContentHash'
+export type ItemEvent = CreateItemEvent | UpdateItemEvent
 export type CreateItemEvent = Omit<DeepPartial<LibraryItem>, IgnoredFields>
 export type UpdateItemEvent = Omit<
   QueryDeepPartialEntity<LibraryItem>,
@@ -1295,3 +1294,297 @@ export const findLibraryItemIdsByLabelId = async (
     userId
   )
 }
+
+export const filterItemEvents = (
+  ast: LiqeQuery,
+  events: readonly ItemEvent[]
+): readonly ItemEvent[] => {
+  const testNo = (value: string, event: ItemEvent) => {
+    const keywordRegexMap: Record<string, RegExp> = {
+      highlightAnnotations: /^highlight(s)?$/i,
+      labelNames: /^label(s)?$/i,
+      subscription: /^subscription(s)?$/i,
+    }
+
+    const matchingKeyword = Object.keys(keywordRegexMap).find((keyword) =>
+      value.match(keywordRegexMap[keyword])
+    )
+
+    if (!matchingKeyword) {
+      throw new Error(`Unexpected keyword: ${value}`)
+    }
+
+    const key = matchingKeyword as
+      | 'highlightAnnotations'
+      | 'labelNames'
+      | 'subscription'
+    const eventValue = event[key] as string | string[]
+
+    return !eventValue || (Array.isArray(eventValue) && eventValue.length === 0)
+  }
+
+  const testEvent = (ast: LiqeQuery, event: ItemEvent) => {
+    if (ast.type !== 'Tag') {
+      throw new Error('Expected a tag expression.')
+    }
+
+    const { field, expression } = ast
+
+    if (field.type === 'ImplicitField') {
+      // TODO: Implement full text search
+      throw new Error('Full text search is not supported.')
+    }
+    if (expression.type !== 'LiteralExpression') {
+      // ignore empty values
+      throw new Error('Expected a literal expression.')
+    }
+
+    const value = expression.value?.toString()
+    if (!value) {
+      // ignore empty values
+      throw new Error('Expected a non-empty value.')
+    }
+
+    switch (field.name.toLowerCase()) {
+      case 'in': {
+        switch (value.toLowerCase()) {
+          case InFilter.ALL:
+            return true
+          case InFilter.ARCHIVE:
+            return event.state === LibraryItemState.Archived
+          case InFilter.TRASH:
+            return event.state === LibraryItemState.Deleted
+          default:
+            return event.state === LibraryItemState.Succeeded
+        }
+      }
+
+      case 'is': {
+        switch (value.toLowerCase()) {
+          case ReadFilter.READ:
+            return (
+              event.readingProgressBottomPercent &&
+              event.readingProgressBottomPercent > 98
+            )
+          case ReadFilter.READING:
+            return (
+              event.readingProgressBottomPercent &&
+              event.readingProgressBottomPercent >= 2 &&
+              event.readingProgressBottomPercent <= 98
+            )
+          case ReadFilter.UNREAD:
+            return (
+              !event.readingProgressBottomPercent ||
+              event.readingProgressBottomPercent < 2
+            )
+          default:
+            throw new Error(`Unexpected keyword: ${value}`)
+        }
+      }
+      case 'type': {
+        return event.itemType?.toString().toLowerCase() === value.toLowerCase()
+      }
+      case 'label': {
+        const labels = event.labelNames as string[]
+        const labelsToTest = value.toLowerCase().split(',')
+        return labelsToTest.some((label) => {
+          const hasWildcard = label.includes('*')
+          if (hasWildcard) {
+            return labels.some((l) => l.match(new RegExp(label, 'i')))
+          }
+
+          return labels.some((l) => l.toLowerCase() === label)
+        })
+      }
+      case 'has':
+        return !testNo(value, event)
+      case 'read':
+      case 'updated':
+      case 'published': {
+        let startDate: Date | undefined
+        let endDate: Date | undefined
+        // check for special date filters
+        switch (value.toLowerCase()) {
+          case 'today':
+            startDate = DateTime.local().startOf('day').toJSDate()
+            break
+          case 'yesterday': {
+            const yesterday = DateTime.local().minus({ days: 1 })
+            startDate = yesterday.startOf('day').toJSDate()
+            endDate = yesterday.endOf('day').toJSDate()
+            break
+          }
+          case 'this week':
+            startDate = DateTime.local().startOf('week').toJSDate()
+            break
+          case 'this month':
+            startDate = DateTime.local().startOf('month').toJSDate()
+            break
+          default: {
+            // check for date ranges
+            const [start, end] = value.split('..')
+            // validate date
+            if (start && start !== '*') {
+              startDate = new Date(start)
+              if (isNaN(startDate.getTime())) {
+                throw new Error('Invalid start date')
+              }
+            }
+
+            if (end && end !== '*') {
+              endDate = new Date(end)
+              if (isNaN(endDate.getTime())) {
+                throw new Error('Invalid end date')
+              }
+            }
+          }
+        }
+
+        const start = startDate ?? new Date(0)
+        const end = endDate ?? new Date()
+        const key = `${field.name.toLowerCase()}At` as
+          | 'readAt'
+          | 'updatedAt'
+          | 'publishedAt'
+        const eventValue = event[key] as Date
+
+        return eventValue >= start && eventValue <= end
+      }
+      // term filters
+      case 'subscription':
+      case 'rss':
+      case 'language': {
+        const columnName = getColumnName(field.name)
+        // get camel case column name
+        const key = camelCase(columnName) as 'subscription' | 'itemLanguage'
+
+        return event[key]?.toString().toLowerCase() === value.toLowerCase()
+      }
+      // match filters
+      case 'author':
+      case 'title':
+      case 'description':
+      case 'note':
+      case 'site': {
+        const columnName = getColumnName(field.name)
+        const key = camelCase(columnName) as
+          | 'author'
+          | 'title'
+          | 'description'
+          | 'note'
+          | 'siteName'
+
+        // TODO: Implement full text search
+        return event[key]?.toString().match(new RegExp(value, 'i'))
+      }
+      case 'includes': {
+        const ids = value.split(',')
+        if (!ids || ids.length === 0) {
+          throw new Error('Expected ids')
+        }
+
+        return event.id && ids.includes(event.id.toString())
+      }
+      case 'recommendedby': {
+        if (value === '*') {
+          // select all if * is provided
+          return event.recommenderNames && event.recommenderNames.length > 0
+        }
+
+        return (
+          event.recommenderNames &&
+          (event.recommenderNames as string[]).some(
+            (name) => name.toLowerCase() === value.toLowerCase()
+          )
+        )
+      }
+      case 'no':
+        return testNo(value, event)
+      case 'use':
+      case 'mode':
+      case 'event':
+        // mode is ignored and used only by the frontend
+        return true
+      case 'readposition':
+      case 'wordscount': {
+        const operatorRegex = /([<>]=?)/
+        const operator = value.match(operatorRegex)?.[0]
+        if (!operator) {
+          throw new Error('Expected operator')
+        }
+
+        const newValue = value.replace(operatorRegex, '')
+        const intValue = parseInt(newValue, 10)
+
+        const column = getColumnName(field.name)
+        const key = camelCase(column) as
+          | 'wordCount'
+          | 'readingProgressBottomPercent'
+        const eventValue = event[key] as number
+
+        switch (operator) {
+          case '>':
+            return eventValue > intValue
+          case '>=':
+            return eventValue >= intValue
+          case '<':
+            return eventValue < intValue
+          case '<=':
+            return eventValue <= intValue
+          default:
+            throw new Error('Unexpected operator')
+        }
+      }
+      default:
+        // TODO: Implement full text search
+        return false
+      // treat unknown fields as implicit fields
+      // return serializeImplicitField({
+      //   ...expression,
+      //   value: `${field.name}:${value}`,
+      // })
+    }
+  }
+
+  if (ast.type === 'Tag') {
+    return events.filter((event) => {
+      return testEvent(ast, event)
+    })
+  }
+
+  if (ast.type === 'UnaryOperator') {
+    const removeRows = filterItemEvents(ast.operand, events)
+
+    return events.filter((row) => {
+      return !removeRows.includes(row)
+    })
+  }
+
+  if (ast.type === 'ParenthesizedExpression') {
+    return filterItemEvents(ast.expression, events)
+  }
+
+  if (!ast.left) {
+    throw new Error('Expected left to be defined.')
+  }
+
+  const leftRows = filterItemEvents(ast.left, events)
+
+  if (!ast.right) {
+    throw new Error('Expected right to be defined.')
+  }
+
+  if (ast.type !== 'LogicalExpression') {
+    throw new Error('Expected a tag expression.')
+  }
+
+  if (ast.operator.operator === 'OR') {
+    const rightRows = filterItemEvents(ast.right, events)
+
+    return Array.from(new Set([...leftRows, ...rightRows]))
+  } else if (ast.operator.operator === 'AND') {
+    return filterItemEvents(ast.right, leftRows)
+  }
+
+  throw new Error('Unexpected state.')
+}

From de28a9ad5f69f8a150d3424f50cedaf02d540bd1 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 18:41:14 +0800
Subject: [PATCH 08/14] fix bug

---
 packages/api/src/jobs/trigger_rule.ts     |  4 +-
 packages/api/src/services/library_item.ts | 77 +++++++++++++----------
 2 files changed, 47 insertions(+), 34 deletions(-)

diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts
index f94e3c37a..8a532eb07 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -1,9 +1,9 @@
-import { filter } from '@omnivore/liqe'
 import { ReadingProgressDataSource } from '../datasources/reading_progress_data_source'
 import { LibraryItemState } from '../entity/library_item'
 import { Rule, RuleAction, RuleActionType, RuleEventType } from '../entity/rule'
 import { addLabelsToLibraryItem } from '../services/labels'
 import {
+  filterItemEvents,
   ItemEvent,
   softDeleteLibraryItem,
   updateLibraryItem,
@@ -112,7 +112,7 @@ const triggerActions = async (
     try {
       const ast = parseSearchQuery(rule.filter)
       // filter library item by rule filter
-      const results = filter(ast, [data])
+      const results = filterItemEvents(ast, [data])
       if (results.length === 0) {
         logger.info(`No items found for rule ${rule.id}`)
         continue
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index dcea18b29..2a0b2de68 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -878,7 +878,8 @@ export const updateLibraryItemReadingProgress = async (
   userId: string,
   bottomPercent: number,
   topPercent: number | null = null,
-  anchorIndex: number | null = null
+  anchorIndex: number | null = null,
+  pubsub = createPubSubClient()
 ): Promise<LibraryItem | null> => {
   // If we have a top percent, we only save it if it's greater than the current top percent
   // or set to zero if the top percent is zero.
@@ -925,6 +926,13 @@ export const updateLibraryItemReadingProgress = async (
   }
 
   const updatedItem = result[0][0]
+  await pubsub.entityUpdated<UpdateItemEvent>(
+    EntityType.PAGE,
+    updatedItem,
+    userId,
+    id
+  )
+
   return updatedItem
 }
 
@@ -1330,24 +1338,32 @@ export const filterItemEvents = (
 
     const { field, expression } = ast
 
-    if (field.type === 'ImplicitField') {
-      // TODO: Implement full text search
-      throw new Error('Full text search is not supported.')
-    }
     if (expression.type !== 'LiteralExpression') {
       // ignore empty values
       throw new Error('Expected a literal expression.')
     }
 
-    const value = expression.value?.toString()
-    if (!value) {
+    const lowercasedValue = expression.value?.toString().toLowerCase()
+
+    if (field.type === 'ImplicitField') {
+      if (!lowercasedValue) {
+        return true
+      }
+
+      // TODO: Implement full text search
+      return event.readableContent
+        ?.toString()
+        ?.match(new RegExp(lowercasedValue, 'i'))
+    }
+
+    if (!lowercasedValue) {
       // ignore empty values
       throw new Error('Expected a non-empty value.')
     }
 
     switch (field.name.toLowerCase()) {
       case 'in': {
-        switch (value.toLowerCase()) {
+        switch (lowercasedValue) {
           case InFilter.ALL:
             return true
           case InFilter.ARCHIVE:
@@ -1355,12 +1371,15 @@ export const filterItemEvents = (
           case InFilter.TRASH:
             return event.state === LibraryItemState.Deleted
           default:
-            return event.state === LibraryItemState.Succeeded
+            return (
+              event.state != LibraryItemState.Archived &&
+              event.state != LibraryItemState.Deleted
+            )
         }
       }
 
       case 'is': {
-        switch (value.toLowerCase()) {
+        switch (lowercasedValue) {
           case ReadFilter.READ:
             return (
               event.readingProgressBottomPercent &&
@@ -1378,15 +1397,15 @@ export const filterItemEvents = (
               event.readingProgressBottomPercent < 2
             )
           default:
-            throw new Error(`Unexpected keyword: ${value}`)
+            throw new Error(`Unexpected keyword: ${lowercasedValue}`)
         }
       }
       case 'type': {
-        return event.itemType?.toString().toLowerCase() === value.toLowerCase()
+        return event.itemType?.toString().toLowerCase() === lowercasedValue
       }
       case 'label': {
         const labels = event.labelNames as string[]
-        const labelsToTest = value.toLowerCase().split(',')
+        const labelsToTest = lowercasedValue.split(',')
         return labelsToTest.some((label) => {
           const hasWildcard = label.includes('*')
           if (hasWildcard) {
@@ -1397,14 +1416,14 @@ export const filterItemEvents = (
         })
       }
       case 'has':
-        return !testNo(value, event)
+        return !testNo(lowercasedValue, event)
       case 'read':
       case 'updated':
       case 'published': {
         let startDate: Date | undefined
         let endDate: Date | undefined
         // check for special date filters
-        switch (value.toLowerCase()) {
+        switch (lowercasedValue) {
           case 'today':
             startDate = DateTime.local().startOf('day').toJSDate()
             break
@@ -1422,7 +1441,7 @@ export const filterItemEvents = (
             break
           default: {
             // check for date ranges
-            const [start, end] = value.split('..')
+            const [start, end] = lowercasedValue.split('..')
             // validate date
             if (start && start !== '*') {
               startDate = new Date(start)
@@ -1458,7 +1477,7 @@ export const filterItemEvents = (
         // get camel case column name
         const key = camelCase(columnName) as 'subscription' | 'itemLanguage'
 
-        return event[key]?.toString().toLowerCase() === value.toLowerCase()
+        return event[key]?.toString().toLowerCase() === lowercasedValue
       }
       // match filters
       case 'author':
@@ -1475,10 +1494,10 @@ export const filterItemEvents = (
           | 'siteName'
 
         // TODO: Implement full text search
-        return event[key]?.toString().match(new RegExp(value, 'i'))
+        return event[key]?.toString().match(new RegExp(lowercasedValue, 'i'))
       }
       case 'includes': {
-        const ids = value.split(',')
+        const ids = lowercasedValue.split(',')
         if (!ids || ids.length === 0) {
           throw new Error('Expected ids')
         }
@@ -1486,7 +1505,7 @@ export const filterItemEvents = (
         return event.id && ids.includes(event.id.toString())
       }
       case 'recommendedby': {
-        if (value === '*') {
+        if (lowercasedValue === '*') {
           // select all if * is provided
           return event.recommenderNames && event.recommenderNames.length > 0
         }
@@ -1494,12 +1513,12 @@ export const filterItemEvents = (
         return (
           event.recommenderNames &&
           (event.recommenderNames as string[]).some(
-            (name) => name.toLowerCase() === value.toLowerCase()
+            (name) => name.toLowerCase() === lowercasedValue
           )
         )
       }
       case 'no':
-        return testNo(value, event)
+        return testNo(lowercasedValue, event)
       case 'use':
       case 'mode':
       case 'event':
@@ -1508,12 +1527,12 @@ export const filterItemEvents = (
       case 'readposition':
       case 'wordscount': {
         const operatorRegex = /([<>]=?)/
-        const operator = value.match(operatorRegex)?.[0]
+        const operator = lowercasedValue.match(operatorRegex)?.[0]
         if (!operator) {
           throw new Error('Expected operator')
         }
 
-        const newValue = value.replace(operatorRegex, '')
+        const newValue = lowercasedValue.replace(operatorRegex, '')
         const intValue = parseInt(newValue, 10)
 
         const column = getColumnName(field.name)
@@ -1536,13 +1555,7 @@ export const filterItemEvents = (
         }
       }
       default:
-        // TODO: Implement full text search
         return false
-      // treat unknown fields as implicit fields
-      // return serializeImplicitField({
-      //   ...expression,
-      //   value: `${field.name}:${value}`,
-      // })
     }
   }
 
@@ -1555,8 +1568,8 @@ export const filterItemEvents = (
   if (ast.type === 'UnaryOperator') {
     const removeRows = filterItemEvents(ast.operand, events)
 
-    return events.filter((row) => {
-      return !removeRows.includes(row)
+    return events.filter((event) => {
+      return !removeRows.includes(event)
     })
   }
 

From edeafc9a7196e57c4af93c373f34c6c988519d1e Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 19:07:31 +0800
Subject: [PATCH 09/14] fix bug

---
 packages/api/src/pubsub.ts | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts
index b210fbfa5..c64b45900 100644
--- a/packages/api/src/pubsub.ts
+++ b/packages/api/src/pubsub.ts
@@ -5,14 +5,12 @@ import { env } from './env'
 import { ReportType } from './generated/graphql'
 import { FeatureName, findFeatureByName } from './services/features'
 import {
-  enqueueAISummarizeJob,
   enqueueExportItem,
   enqueueProcessYouTubeVideo,
   enqueueTriggerRuleJob,
   enqueueWebhookJob,
 } from './utils/createTask'
 import { buildLogger } from './utils/logger'
-import { processYouTubeVideo } from './jobs/process-youtube-video'
 
 const logger = buildLogger('pubsub')
 
@@ -95,10 +93,11 @@ export const createPubSubClient = (): PubsubClient => {
         // })
       }
 
-      if (
-        'originalUrl' in data &&
-        isYouTubeVideoURL(data['originalUrl'] as string | undefined)
-      ) {
+      const isYoutubeVideo = (data: any): data is { originalUrl: string } => {
+        return 'originalUrl' in data
+      }
+
+      if (isYoutubeVideo(data) && isYouTubeVideoURL(data['originalUrl'])) {
         await enqueueProcessYouTubeVideo({
           userId,
           libraryItemId,

From 7f224cc1e1c0733168b3dd06d46d72c2b77d37ef Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Mon, 18 Mar 2024 22:29:46 +0800
Subject: [PATCH 10/14] running text search on all text fields

---
 packages/api/src/services/library_item.ts | 28 +++++++++++++----------
 1 file changed, 16 insertions(+), 12 deletions(-)

diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 2a0b2de68..8cc80bd6a 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1322,11 +1322,7 @@ export const filterItemEvents = (
       throw new Error(`Unexpected keyword: ${value}`)
     }
 
-    const key = matchingKeyword as
-      | 'highlightAnnotations'
-      | 'labelNames'
-      | 'subscription'
-    const eventValue = event[key] as string | string[]
+    const eventValue = event[matchingKeyword as keyof ItemEvent]
 
     return !eventValue || (Array.isArray(eventValue) && eventValue.length === 0)
   }
@@ -1350,10 +1346,21 @@ export const filterItemEvents = (
         return true
       }
 
+      const textFields = [
+        'author',
+        'title',
+        'description',
+        'note',
+        'siteName',
+        'readableContent',
+        'originalUrl',
+      ]
+      const text = textFields
+        .map((field) => event[field as keyof ItemEvent])
+        .join(' ')
+
       // TODO: Implement full text search
-      return event.readableContent
-        ?.toString()
-        ?.match(new RegExp(lowercasedValue, 'i'))
+      return text.match(new RegExp(lowercasedValue, 'i'))
     }
 
     if (!lowercasedValue) {
@@ -1461,10 +1468,7 @@ export const filterItemEvents = (
 
         const start = startDate ?? new Date(0)
         const end = endDate ?? new Date()
-        const key = `${field.name.toLowerCase()}At` as
-          | 'readAt'
-          | 'updatedAt'
-          | 'publishedAt'
+        const key = `${field.name.toLowerCase()}At` as keyof ItemEvent
         const eventValue = event[key] as Date
 
         return eventValue >= start && eventValue <= end

From abf9706539bd37b2e99e9d27be5813636196eb38 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Tue, 19 Mar 2024 11:53:43 +0800
Subject: [PATCH 11/14] resolve conflicts after rebasing

---
 packages/api/src/services/integrations/notion.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/api/src/services/integrations/notion.ts b/packages/api/src/services/integrations/notion.ts
index de517dca6..7e891287d 100644
--- a/packages/api/src/services/integrations/notion.ts
+++ b/packages/api/src/services/integrations/notion.ts
@@ -256,7 +256,7 @@ export class NotionClient implements IntegrationClient {
                       text: {
                         content: highlight.quote || '',
                         link: {
-                          url: highlightUrl(item.slug, highlight.id),
+                          url: getHighlightUrl(item.slug, highlight.id),
                         },
                       },
                       annotations: {

From 2e374557bf9a459202338898754f7e59b55bed46 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Tue, 19 Mar 2024 13:03:45 +0800
Subject: [PATCH 12/14] send labels in the page_created events

---
 packages/api/src/services/save_file.ts |  5 +++--
 packages/api/src/services/save_page.ts |  4 ++++
 packages/api/src/utils/createTask.ts   | 29 +++++++++++++++++---------
 3 files changed, 26 insertions(+), 12 deletions(-)

diff --git a/packages/api/src/services/save_file.ts b/packages/api/src/services/save_file.ts
index a004f1558..4ada19787 100644
--- a/packages/api/src/services/save_file.ts
+++ b/packages/api/src/services/save_file.ts
@@ -3,7 +3,7 @@ import { User } from '../entity/user'
 import { homePageURL } from '../env'
 import { SaveErrorCode, SaveFileInput, SaveResult } from '../generated/graphql'
 import { getStorageFileDetails } from '../utils/uploads'
-import { createAndSaveLabelsInLibraryItem } from './labels'
+import { createAndAddLabelsToLibraryItem } from './labels'
 import { updateLibraryItem } from './library_item'
 import { findUploadFileById, setFileUploadComplete } from './upload_file'
 
@@ -39,12 +39,13 @@ export const saveFile = async (
       folder: input.folder || undefined,
       savedAt: input.savedAt ? new Date(input.savedAt) : undefined,
       publishedAt: input.publishedAt ? new Date(input.publishedAt) : undefined,
+      labelNames: input.labels?.map((label) => label.name) || undefined,
     },
     user.id
   )
 
   // add labels to item
-  await createAndSaveLabelsInLibraryItem(
+  await createAndAddLabelsToLibraryItem(
     input.clientRequestId,
     user.id,
     input.labels,
diff --git a/packages/api/src/services/save_page.ts b/packages/api/src/services/save_page.ts
index 232ef3cad..e54b3d026 100644
--- a/packages/api/src/services/save_page.ts
+++ b/packages/api/src/services/save_page.ts
@@ -134,6 +134,7 @@ export const savePage = async (
     feedContent: input.feedContent,
     dir: parseResult.parsedContent?.dir,
     preparedDocument,
+    labelNames: input.labels?.map((label) => label.name),
   })
   const isImported =
     input.source === 'csv-importer' || input.source === 'pocket'
@@ -213,6 +214,7 @@ export const parsedContentToLibraryItem = ({
   folder,
   feedContent,
   dir,
+  labelNames,
 }: {
   url: string
   userId: string
@@ -234,6 +236,7 @@ export const parsedContentToLibraryItem = ({
   folder?: string | null
   feedContent?: string | null
   dir?: string | null
+  labelNames?: string[]
 }): DeepPartial<LibraryItem> & { originalUrl: string } => {
   logger.info('save_page', { url, state, itemId })
   return {
@@ -281,5 +284,6 @@ export const parsedContentToLibraryItem = ({
       dir?.toLowerCase() === 'rtl'
         ? DirectionalityType.RTL
         : DirectionalityType.LTR, // default to LTR
+    labelNames,
   }
 }
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index 6d6c30baf..9dffb7d5e 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -15,6 +15,7 @@ import {
   ArticleSavingRequestStatus,
   CreateLabelInput,
 } from '../generated/graphql'
+import { AISummarizeJobData, AI_SUMMARIZE_JOB_NAME } from '../jobs/ai-summarize'
 import { BulkActionData, BULK_ACTION_JOB_NAME } from '../jobs/bulk_action'
 import { CallWebhookJobData, CALL_WEBHOOK_JOB_NAME } from '../jobs/call_webhook'
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
@@ -23,6 +24,12 @@ import {
   ExportItemJobData,
   EXPORT_ITEM_JOB_NAME,
 } from '../jobs/integration/export_item'
+import {
+  ProcessYouTubeTranscriptJobData,
+  ProcessYouTubeVideoJobData,
+  PROCESS_YOUTUBE_TRANSCRIPT_JOB_NAME,
+  PROCESS_YOUTUBE_VIDEO_JOB_NAME,
+} from '../jobs/process-youtube-video'
 import {
   queueRSSRefreshFeedJob,
   REFRESH_ALL_FEEDS_JOB_NAME,
@@ -44,13 +51,6 @@ import { CreateTaskError } from './errors'
 import { stringToHash } from './helpers'
 import { logger } from './logger'
 import View = google.cloud.tasks.v2.Task.View
-import { AISummarizeJobData, AI_SUMMARIZE_JOB_NAME } from '../jobs/ai-summarize'
-import {
-  PROCESS_YOUTUBE_TRANSCRIPT_JOB_NAME,
-  PROCESS_YOUTUBE_VIDEO_JOB_NAME,
-  ProcessYouTubeTranscriptJobData,
-  ProcessYouTubeVideoJobData,
-} from '../jobs/process-youtube-video'
 
 // Instantiates a client.
 const client = new CloudTasksClient()
@@ -132,9 +132,18 @@ const createHttpTaskWithToken = async ({
 > => {
   // If there is no Google Cloud Project Id exposed, it means that we are in local environment
   if (env.dev.isLocal || !project) {
-    logger.error(
-      'error: attempting to create a cloud task but not running in google cloud.'
-    )
+    setTimeout(() => {
+      axios
+        .post(taskHandlerUrl, payload, {
+          headers: {
+            'Content-Type': 'application/json',
+            ...requestHeaders,
+          },
+        })
+        .catch((error) => {
+          logError(error)
+        })
+    })
     return null
   }
 

From 689eeaeacdff9c8e9cca82c3318ec27eff9c617e Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 20 Mar 2024 10:46:13 +0800
Subject: [PATCH 13/14] disable sending pubsub events for rss feed temporarily
 to resolve conflicts

---
 .../api/src/resolvers/discover_feeds/add.ts   | 29 ++++++++++---------
 1 file changed, 15 insertions(+), 14 deletions(-)

diff --git a/packages/api/src/resolvers/discover_feeds/add.ts b/packages/api/src/resolvers/discover_feeds/add.ts
index 7cc72d729..5420bc8b5 100644
--- a/packages/api/src/resolvers/discover_feeds/add.ts
+++ b/packages/api/src/resolvers/discover_feeds/add.ts
@@ -2,7 +2,11 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 /* eslint-disable @typescript-eslint/require-await */
-import { authorized } from '../../utils/gql-utils'
+import axios from 'axios'
+import { XMLParser } from 'fast-xml-parser'
+import { QueryRunner } from 'typeorm'
+import { v4 } from 'uuid'
+import { appDataSource } from '../../data_source'
 import {
   AddDiscoverFeedError,
   AddDiscoverFeedErrorCode,
@@ -10,13 +14,9 @@ import {
   DiscoverFeed,
   MutationAddDiscoverFeedArgs,
 } from '../../generated/graphql'
-import { appDataSource } from '../../data_source'
-import { QueryRunner } from 'typeorm'
-import axios from 'axios'
-import { RSS_PARSER_CONFIG } from '../../utils/parser'
-import { XMLParser } from 'fast-xml-parser'
 import { EntityType } from '../../pubsub'
-import { v4 } from 'uuid'
+import { authorized } from '../../utils/gql-utils'
+import { RSS_PARSER_CONFIG } from '../../utils/parser'
 
 const parser = new XMLParser({
   ignoreAttributes: false,
@@ -183,13 +183,14 @@ export const addDiscoverFeedResolver = authorized<
     }
 
     const result = await addNewSubscription(queryRunner, url, uid)
-    if (result.__typename == 'AddDiscoverFeedSuccess') {
-      await pubsub.entityCreated(
-        EntityType.RSS_FEED,
-        { feed: result.feed, libraryItemId: 'NA' },
-        uid
-      )
-    }
+    // TODO: Add pubsub for new feed
+    // if (result.__typename == 'AddDiscoverFeedSuccess') {
+    //   await pubsub.entityCreated(
+    //     EntityType.RSS_FEED,
+    //     { feed: result.feed, libraryItemId: 'NA' },
+    //     uid
+    //   )
+    // }
 
     return result
   } catch (error) {

From 647a8f6e24691f844125bfdfba67c36c20d1a8d6 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 20 Mar 2024 11:00:26 +0800
Subject: [PATCH 14/14] update the migration version number to resolve
 conflicts

---
 ...dd_failed_at_to_rule.sql => 0169.do.add_failed_at_to_rule.sql} | 0
 ..._failed_at_to_rule.sql => 0169.undo.add_failed_at_to_rule.sql} | 0
 2 files changed, 0 insertions(+), 0 deletions(-)
 rename packages/db/migrations/{0168.do.add_failed_at_to_rule.sql => 0169.do.add_failed_at_to_rule.sql} (100%)
 rename packages/db/migrations/{0168.undo.add_failed_at_to_rule.sql => 0169.undo.add_failed_at_to_rule.sql} (100%)

diff --git a/packages/db/migrations/0168.do.add_failed_at_to_rule.sql b/packages/db/migrations/0169.do.add_failed_at_to_rule.sql
similarity index 100%
rename from packages/db/migrations/0168.do.add_failed_at_to_rule.sql
rename to packages/db/migrations/0169.do.add_failed_at_to_rule.sql
diff --git a/packages/db/migrations/0168.undo.add_failed_at_to_rule.sql b/packages/db/migrations/0169.undo.add_failed_at_to_rule.sql
similarity index 100%
rename from packages/db/migrations/0168.undo.add_failed_at_to_rule.sql
rename to packages/db/migrations/0169.undo.add_failed_at_to_rule.sql