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..8a532eb07 100644
--- a/packages/api/src/jobs/trigger_rule.ts
+++ b/packages/api/src/jobs/trigger_rule.ts
@@ -1,27 +1,30 @@
 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,
+  filterItemEvents,
+  ItemEvent,
   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'
+import { parseSearchQuery } from '../utils/search'
 
 export interface TriggerRuleJobData {
   libraryItemId: string
   userId: string
   ruleEventType: RuleEventType
+  data: ItemEvent
 }
 
 interface RuleActionObj {
+  libraryItemId: string
   userId: string
   action: RuleAction
-  libraryItem: LibraryItem
+  data: ItemEvent
 }
 type RuleActionFunc = (obj: RuleActionObj) => Promise<unknown>
 
@@ -33,19 +36,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,
@@ -56,7 +59,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,
@@ -66,15 +69,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)
@@ -96,36 +99,41 @@ const getRuleAction = (actionType: RuleActionType): RuleActionFunc => {
 }
 
 const triggerActions = async (
+  libraryItemId: string,
   userId: string,
   rules: Rule[],
-  data: TriggerRuleJobData
+  data: ItemEvent
 ) => {
   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 filteredData: ItemEvent
+
+    try {
+      const ast = parseSearchQuery(rule.filter)
+      // filter library item by rule filter
+      const results = filterItemEvents(ast, [data])
+      if (results.length === 0) {
+        logger.info(`No items found for rule ${rule.id}`)
+        continue
+      }
+
+      filteredData = results[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)
 
-    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 = {
+        libraryItemId,
         userId,
         action,
-        libraryItem,
+        data: filteredData,
       }
 
       actionPromises.push(actionFunc(actionObj))
@@ -139,8 +147,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)
@@ -149,7 +157,7 @@ export const triggerRule = async (data: TriggerRuleJobData) => {
     return false
   }
 
-  await triggerActions(userId, rules, data)
+  await triggerActions(libraryItemId, userId, rules, data)
 
   return true
 }
diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts
index f13bfe5e4..c64b45900 100644
--- a/packages/api/src/pubsub.ts
+++ b/packages/api/src/pubsub.ts
@@ -3,32 +3,19 @@ import express from 'express'
 import { RuleEventType } from './entity/rule'
 import { env } from './env'
 import { ReportType } from './generated/graphql'
-import { Merge } from './util'
+import { FeatureName, findFeatureByName } from './services/features'
 import {
-  enqueueAISummarizeJob,
   enqueueExportItem,
   enqueueProcessYouTubeVideo,
   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')
 
 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
@@ -42,8 +29,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()}`)
@@ -75,16 +60,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
@@ -93,11 +79,6 @@ export const createPubSubClient = (): PubsubClient => {
         libraryItemIds: [libraryItemId],
       })
 
-      const cleanData = deepDelete(
-        data as EntityData<T> & Record<typeof fieldsToDelete[number], unknown>,
-        [...fieldsToDelete]
-      )
-
       await enqueueWebhookJob({
         userId,
         type,
@@ -112,34 +93,30 @@ 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,
         })
       }
-
-      return publish(
-        'entityCreated',
-        Buffer.from(JSON.stringify({ type, userId, ...cleanData }))
-      )
     },
     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
@@ -148,32 +125,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,
@@ -207,13 +172,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/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) {
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,
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/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/integrations/notion.ts b/packages/api/src/services/integrations/notion.ts
index 4d823e031..7e891287d 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 =
@@ -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: {
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/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 52d44f7ff..8cc80bd6a 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,
@@ -29,6 +30,19 @@ import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
 import { addLabelsToLibraryItem } from './labels'
 
+type IgnoredFields =
+  | 'user'
+  | 'uploadFile'
+  | 'previewContentType'
+  | 'links'
+  | 'textContentHash'
+export type ItemEvent = CreateItemEvent | UpdateItemEvent
+export type CreateItemEvent = Omit<DeepPartial<LibraryItem>, IgnoredFields>
+export type UpdateItemEvent = Omit<
+  QueryDeepPartialEntity<LibraryItem>,
+  IgnoredFields
+>
+
 enum ReadFilter {
   ALL = 'all',
   READ = 'read',
@@ -833,21 +847,27 @@ 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,
+      userId,
+      id
+    )
+
+    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
+    libraryItem,
+    userId,
+    id
   )
 
   return updatedLibraryItem
@@ -858,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.
@@ -905,6 +926,13 @@ export const updateLibraryItemReadingProgress = async (
   }
 
   const updatedItem = result[0][0]
+  await pubsub.entityUpdated<UpdateItemEvent>(
+    EntityType.PAGE,
+    updatedItem,
+    userId,
+    id
+  )
+
   return updatedItem
 }
 
@@ -999,16 +1027,11 @@ 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
+    newLibraryItem,
+    userId,
+    newLibraryItem.id
   )
 
   return newLibraryItem
@@ -1279,3 +1302,306 @@ 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 eventValue = event[matchingKeyword as keyof ItemEvent]
+
+    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 (expression.type !== 'LiteralExpression') {
+      // ignore empty values
+      throw new Error('Expected a literal expression.')
+    }
+
+    const lowercasedValue = expression.value?.toString().toLowerCase()
+
+    if (field.type === 'ImplicitField') {
+      if (!lowercasedValue) {
+        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 text.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 (lowercasedValue) {
+          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.Archived &&
+              event.state != LibraryItemState.Deleted
+            )
+        }
+      }
+
+      case 'is': {
+        switch (lowercasedValue) {
+          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: ${lowercasedValue}`)
+        }
+      }
+      case 'type': {
+        return event.itemType?.toString().toLowerCase() === lowercasedValue
+      }
+      case 'label': {
+        const labels = event.labelNames as string[]
+        const labelsToTest = lowercasedValue.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(lowercasedValue, event)
+      case 'read':
+      case 'updated':
+      case 'published': {
+        let startDate: Date | undefined
+        let endDate: Date | undefined
+        // check for special date filters
+        switch (lowercasedValue) {
+          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] = lowercasedValue.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 keyof ItemEvent
+        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() === lowercasedValue
+      }
+      // 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(lowercasedValue, 'i'))
+      }
+      case 'includes': {
+        const ids = lowercasedValue.split(',')
+        if (!ids || ids.length === 0) {
+          throw new Error('Expected ids')
+        }
+
+        return event.id && ids.includes(event.id.toString())
+      }
+      case 'recommendedby': {
+        if (lowercasedValue === '*') {
+          // select all if * is provided
+          return event.recommenderNames && event.recommenderNames.length > 0
+        }
+
+        return (
+          event.recommenderNames &&
+          (event.recommenderNames as string[]).some(
+            (name) => name.toLowerCase() === lowercasedValue
+          )
+        )
+      }
+      case 'no':
+        return testNo(lowercasedValue, 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 = lowercasedValue.match(operatorRegex)?.[0]
+        if (!operator) {
+          throw new Error('Expected operator')
+        }
+
+        const newValue = lowercasedValue.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:
+        return false
+    }
+  }
+
+  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((event) => {
+      return !removeRows.includes(event)
+    })
+  }
+
+  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.')
+}
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/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
   }
 
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}`
diff --git a/packages/db/migrations/0169.do.add_failed_at_to_rule.sql b/packages/db/migrations/0169.do.add_failed_at_to_rule.sql
new file mode 100755
index 000000000..6f77f166c
--- /dev/null
+++ b/packages/db/migrations/0169.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/0169.undo.add_failed_at_to_rule.sql b/packages/db/migrations/0169.undo.add_failed_at_to_rule.sql
new file mode 100755
index 000000000..de2fc769d
--- /dev/null
+++ b/packages/db/migrations/0169.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 dd0d52448..5cf3e1636 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()
@@ -229,6 +235,7 @@ export default function Rules(): JSX.Element {
         filter: rule.filter,
         actions: rule.actions,
         eventTypes: rule.eventTypes,
+        failedAt: rule.failedAt,
       }
     })
   }, [rules])
@@ -312,6 +319,11 @@ export default function Rules(): JSX.Element {
         </>
       ),
     },
+    {
+      title: 'Failed At',
+      dataIndex: 'failedAt',
+      key: 'failedAt',
+    },
     {
       title: '',
       key: 'tools',