From ddddf8234ec37b8699e3559966740499bab340da Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Tue, 30 Jan 2024 23:53:05 +0800
Subject: [PATCH 1/8] create a job for db trigger

---
 packages/api/src/jobs/update_db.ts   | 31 ++++++++++++++++++++++++++++
 packages/api/src/queue-processor.ts  | 10 +++++++--
 packages/api/src/services/labels.ts  |  6 ++++++
 packages/api/src/utils/createTask.ts | 17 +++++++++++++++
 4 files changed, 62 insertions(+), 2 deletions(-)
 create mode 100644 packages/api/src/jobs/update_db.ts

diff --git a/packages/api/src/jobs/update_db.ts b/packages/api/src/jobs/update_db.ts
new file mode 100644
index 000000000..096111606
--- /dev/null
+++ b/packages/api/src/jobs/update_db.ts
@@ -0,0 +1,31 @@
+import { authTrx } from '../repository'
+
+export const UPDATE_LABELS_IN_LIBRARY_ITEM_JOB = 'update-labels-in-library-item'
+
+export interface UpdateLabelsInLibraryItemData {
+  libraryItemId: string
+  userId: string
+}
+
+export const updateLabelsInLibraryItem = async (
+  data: UpdateLabelsInLibraryItemData
+) => {
+  return authTrx(
+    async (tx) =>
+      tx.query(
+        `WITH labels_agg AS (
+          SELECT array_agg(l.name) AS names_agg
+          FROM omnivore.labels l
+          INNER JOIN omnivore.entity_labels el ON el.label_id = l.id 
+          LEFT JOIN omnivore.highlight h ON h.id = el.highlight_id 
+          WHERE el.library_item_id = $1 OR h.library_item_id = $1
+        )
+        UPDATE omnivore.library_item li
+        SET label_names = COALESCE((SELECT names_agg FROM labels_agg), ARRAY[]::TEXT[])
+        WHERE li.id = $1;`,
+        [data.libraryItemId]
+      ),
+    undefined,
+    data.userId
+  )
+}
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index cd89e1c69..c2379c432 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -2,7 +2,7 @@
 /* eslint-disable @typescript-eslint/restrict-template-expressions */
 /* eslint-disable @typescript-eslint/require-await */
 /* eslint-disable @typescript-eslint/no-misused-promises */
-import { Job, Queue, QueueEvents, Worker, JobType } from 'bullmq'
+import { Job, JobType, Queue, QueueEvents, Worker } from 'bullmq'
 import express, { Express } from 'express'
 import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
 import { appDataSource } from './data_source'
@@ -11,10 +11,14 @@ import { findThumbnail, THUMBNAIL_JOB } from './jobs/find_thumbnail'
 import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
+import { triggerRule, TRIGGER_RULE_JOB_NAME } from './jobs/trigger_rule'
+import {
+  updateLabelsInLibraryItem,
+  UPDATE_LABELS_IN_LIBRARY_ITEM_JOB,
+} from './jobs/update_db'
 import { updatePDFContentJob } from './jobs/update_pdf_content'
 import { redisDataSource } from './redis_data_source'
 import { CustomTypeOrmLogger } from './utils/logger'
-import { triggerRule, TRIGGER_RULE_JOB_NAME } from './jobs/trigger_rule'
 
 export const QUEUE_NAME = 'omnivore-backend-queue'
 
@@ -125,6 +129,8 @@ const main = async () => {
           return findThumbnail(job.data)
         case TRIGGER_RULE_JOB_NAME:
           return triggerRule(job.data)
+        case UPDATE_LABELS_IN_LIBRARY_ITEM_JOB:
+          return updateLabelsInLibraryItem(job.data)
       }
     },
     {
diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index 564a1e2da..03f9eec7b 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -5,6 +5,8 @@ import { Label } from '../entity/label'
 import { createPubSubClient, EntityType, PubsubClient } from '../pubsub'
 import { authTrx } from '../repository'
 import { CreateLabelInput, labelRepository } from '../repository/label'
+import { enqueueUpdateLabelsInLibraryItem } from '../utils/createTask'
+import { logger } from '../utils/logger'
 
 type AddLabelsToLibraryItemEvent = {
   pageId: string
@@ -120,6 +122,10 @@ export const saveLabelsInLibraryItem = async (
       userId
     )
   }
+
+  // update labels in library item
+  const job = await enqueueUpdateLabelsInLibraryItem({ libraryItemId, userId })
+  logger.info('update labels in library item job created', job)
 }
 
 export const addLabelsToLibraryItem = async (
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index 4e3ed31ce..5196f5d31 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -17,6 +17,10 @@ import {
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
 import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
+import {
+  UpdateLabelsInLibraryItemData,
+  UPDATE_LABELS_IN_LIBRARY_ITEM_JOB,
+} from '../jobs/update_db'
 import { getBackendQueue } from '../queue-processor'
 import { redisDataSource } from '../redis_data_source'
 import { signFeatureToken } from '../services/features'
@@ -660,4 +664,17 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => {
   })
 }
 
+export const enqueueUpdateLabelsInLibraryItem = async (
+  data: UpdateLabelsInLibraryItemData
+) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  return queue.add(UPDATE_LABELS_IN_LIBRARY_ITEM_JOB, data, {
+    priority: 1,
+  })
+}
+
 export default createHttpTaskWithToken

From 5ed35bbeff5dfd985f5573698881c6fa260de02b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 13:32:37 +0800
Subject: [PATCH 2/8] create a job for update labels in library item

---
 packages/api/src/jobs/update_db.ts         | 10 ++---
 packages/api/src/queue-processor.ts        |  9 ++---
 packages/api/src/resolvers/labels/index.ts | 12 +++---
 packages/api/src/services/labels.ts        | 47 ++++++++++++++++++++--
 packages/api/src/services/library_item.ts  | 33 +++++++++++++++
 packages/api/src/utils/createTask.ts       | 25 +++++++-----
 6 files changed, 103 insertions(+), 33 deletions(-)

diff --git a/packages/api/src/jobs/update_db.ts b/packages/api/src/jobs/update_db.ts
index 096111606..03e670b25 100644
--- a/packages/api/src/jobs/update_db.ts
+++ b/packages/api/src/jobs/update_db.ts
@@ -1,20 +1,18 @@
 import { authTrx } from '../repository'
 
-export const UPDATE_LABELS_IN_LIBRARY_ITEM_JOB = 'update-labels-in-library-item'
+export const UPDATE_LABELS_JOB = 'update-labels'
 
-export interface UpdateLabelsInLibraryItemData {
+export interface UpdateLabelsData {
   libraryItemId: string
   userId: string
 }
 
-export const updateLabelsInLibraryItem = async (
-  data: UpdateLabelsInLibraryItemData
-) => {
+export const updateLabels = async (data: UpdateLabelsData) => {
   return authTrx(
     async (tx) =>
       tx.query(
         `WITH labels_agg AS (
-          SELECT array_agg(l.name) AS names_agg
+          SELECT array_agg(DISTINCT l.name) AS names_agg
           FROM omnivore.labels l
           INNER JOIN omnivore.entity_labels el ON el.label_id = l.id 
           LEFT JOIN omnivore.highlight h ON h.id = el.highlight_id 
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index c2379c432..f4458e53e 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -12,10 +12,7 @@ import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
 import { triggerRule, TRIGGER_RULE_JOB_NAME } from './jobs/trigger_rule'
-import {
-  updateLabelsInLibraryItem,
-  UPDATE_LABELS_IN_LIBRARY_ITEM_JOB,
-} from './jobs/update_db'
+import { updateLabels, UPDATE_LABELS_JOB } from './jobs/update_db'
 import { updatePDFContentJob } from './jobs/update_pdf_content'
 import { redisDataSource } from './redis_data_source'
 import { CustomTypeOrmLogger } from './utils/logger'
@@ -129,8 +126,8 @@ const main = async () => {
           return findThumbnail(job.data)
         case TRIGGER_RULE_JOB_NAME:
           return triggerRule(job.data)
-        case UPDATE_LABELS_IN_LIBRARY_ITEM_JOB:
-          return updateLabelsInLibraryItem(job.data)
+        case UPDATE_LABELS_JOB:
+          return updateLabels(job.data)
       }
     },
     {
diff --git a/packages/api/src/resolvers/labels/index.ts b/packages/api/src/resolvers/labels/index.ts
index 6a6860940..cf9d6e3b6 100644
--- a/packages/api/src/resolvers/labels/index.ts
+++ b/packages/api/src/resolvers/labels/index.ts
@@ -31,6 +31,7 @@ import {
 import { labelRepository } from '../../repository/label'
 import { userRepository } from '../../repository/user'
 import {
+  deleteLabelById,
   findOrCreateLabels,
   saveLabelsInHighlight,
   saveLabelsInLibraryItem,
@@ -118,13 +119,10 @@ export const deleteLabelResolver = authorized<
   DeleteLabelSuccess,
   DeleteLabelError,
   MutationDeleteLabelArgs
->(async (_, { id: labelId }, { authTrx, log, uid }) => {
+>(async (_, { id: labelId }, { log, uid }) => {
   try {
-    const deleteResult = await authTrx(async (tx) => {
-      return tx.withRepository(labelRepository).deleteById(labelId)
-    })
-
-    if (!deleteResult.affected) {
+    const deleted = await deleteLabelById(labelId, uid)
+    if (!deleted) {
       return {
         errorCodes: [DeleteLabelErrorCode.NotFound],
       }
@@ -281,7 +279,7 @@ export const setLabelsForHighlightResolver = authorized<
       }
     }
 
-    // save labels in the library item
+    // save labels in the highlight
     await saveLabelsInHighlight(labelsSet, input.highlightId, uid, pubsub)
 
     analytics.track({
diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index 03f9eec7b..d4d1aff16 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -5,8 +5,9 @@ import { Label } from '../entity/label'
 import { createPubSubClient, EntityType, PubsubClient } from '../pubsub'
 import { authTrx } from '../repository'
 import { CreateLabelInput, labelRepository } from '../repository/label'
-import { enqueueUpdateLabelsInLibraryItem } from '../utils/createTask'
+import { bulkEnqueueUpdateLabels } from '../utils/createTask'
 import { logger } from '../utils/logger'
+import { findLibraryItemIdsByLabelId } from './library_item'
 
 type AddLabelsToLibraryItemEvent = {
   pageId: string
@@ -124,8 +125,8 @@ export const saveLabelsInLibraryItem = async (
   }
 
   // update labels in library item
-  const job = await enqueueUpdateLabelsInLibraryItem({ libraryItemId, userId })
-  logger.info('update labels in library item job created', job)
+  const jobs = await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
+  logger.info('update labels jobs enqueued', jobs)
 }
 
 export const addLabelsToLibraryItem = async (
@@ -151,6 +152,10 @@ export const addLabelsToLibraryItem = async (
     undefined,
     userId
   )
+
+  // update labels in library item
+  const jobs = await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
+  logger.info('update labels jobs enqueued', jobs)
 }
 
 export const saveLabelsInHighlight = async (
@@ -224,12 +229,33 @@ export const deleteLabels = async (
   )
 }
 
+export const deleteLabelById = async (labelId: string, userId: string) => {
+  const libraryItemIds = await findLibraryItemIdsByLabelId(labelId, userId)
+
+  const deleteResult = await authTrx(async (tx) => {
+    return tx.withRepository(labelRepository).deleteById(labelId)
+  })
+
+  if (!deleteResult.affected) {
+    return false
+  }
+
+  const data = libraryItemIds.map((libraryItemId) => ({
+    libraryItemId,
+    userId,
+  }))
+  const jobs = await bulkEnqueueUpdateLabels(data)
+  logger.info('update labels jobs enqueued', jobs)
+
+  return true
+}
+
 export const updateLabel = async (
   id: string,
   label: QueryDeepPartialEntity<Label>,
   userId: string
 ) => {
-  return authTrx(
+  const updatedLabel = await authTrx(
     async (t) => {
       const repo = t.withRepository(labelRepository)
       await repo.updateLabel(id, label)
@@ -239,6 +265,19 @@ export const updateLabel = async (
     undefined,
     userId
   )
+
+  if (label.name) {
+    const libraryItemIds = await findLibraryItemIdsByLabelId(id, userId)
+
+    const data = libraryItemIds.map((libraryItemId) => ({
+      libraryItemId,
+      userId,
+    }))
+    const jobs = await bulkEnqueueUpdateLabels(data)
+    logger.info('update labels jobs enqueued', jobs)
+  }
+
+  return updatedLabel
 }
 
 export const findLabelsByUserId = async (userId: string): Promise<Label[]> => {
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 73e9fde77..0f3906e63 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1132,3 +1132,36 @@ export const batchDelete = async (criteria: FindOptionsWhere<LibraryItem>) => {
 
   return authTrx(async (t) => t.query(sql))
 }
+
+export const findLibraryItemIdsByLabelId = async (
+  labelId: string,
+  userId: string
+) => {
+  return authTrx(
+    async (tx) => {
+      // find library items have the label or have highlights with the label
+      const result = (await tx.query(
+        `
+        SELECT library_item_id
+        FROM (
+            SELECT library_item_id
+            FROM omnivore.entity_labels
+            WHERE label_id = $1
+                  AND library_item_id IS NOT NULL
+            UNION
+            SELECT h.library_item_id
+            FROM omnivore.highlight h
+            INNER JOIN omnivore.entity_labels ON entity_labels.highlight_id = h.id
+            WHERE label_id = $1
+                  AND highlight_id IS NOT NULL
+        ) AS combined_results
+      `,
+        [labelId]
+      )) as { library_item_id: string }[]
+
+      return result.map((r) => r.library_item_id)
+    },
+    undefined,
+    userId
+  )
+}
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index 5196f5d31..a662262f1 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -17,10 +17,7 @@ import {
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
 import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
-import {
-  UpdateLabelsInLibraryItemData,
-  UPDATE_LABELS_IN_LIBRARY_ITEM_JOB,
-} from '../jobs/update_db'
+import { UpdateLabelsData, UPDATE_LABELS_JOB } from '../jobs/update_db'
 import { getBackendQueue } from '../queue-processor'
 import { redisDataSource } from '../redis_data_source'
 import { signFeatureToken } from '../services/features'
@@ -664,17 +661,25 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => {
   })
 }
 
-export const enqueueUpdateLabelsInLibraryItem = async (
-  data: UpdateLabelsInLibraryItemData
-) => {
+export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
   const queue = await getBackendQueue()
   if (!queue) {
     return undefined
   }
 
-  return queue.add(UPDATE_LABELS_IN_LIBRARY_ITEM_JOB, data, {
-    priority: 1,
-  })
+  const jobs = data.map((d) => ({
+    name: UPDATE_LABELS_JOB,
+    data: d,
+    opts: {
+      priority: 1,
+    },
+  }))
+
+  try {
+    return queue.addBulk(jobs)
+  } catch (error) {
+    logger.error('error enqueuing update labels jobs', error)
+  }
 }
 
 export default createHttpTaskWithToken

From 43855247b8207077dfec5f623310e8d1d72a5d87 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 13:56:56 +0800
Subject: [PATCH 3/8] update bulk action to add labels

---
 packages/api/src/services/labels.ts       | 15 +++++++++++++++
 packages/api/src/services/library_item.ts | 13 ++++++++++++-
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index d4d1aff16..8ad7c24a6 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -1,6 +1,7 @@
 import { DeepPartial, FindOptionsWhere, In } from 'typeorm'
 import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'
 import { EntityLabel, LabelSource } from '../entity/entity_label'
+import { Highlight } from '../entity/highlight'
 import { Label } from '../entity/label'
 import { createPubSubClient, EntityType, PubsubClient } from '../pubsub'
 import { authTrx } from '../repository'
@@ -187,6 +188,20 @@ export const saveLabelsInHighlight = async (
     { highlightId, labels },
     userId
   )
+
+  const highlight = await authTrx(async (tx) =>
+    tx.getRepository(Highlight).findOne({
+      where: { id: highlightId },
+      relations: ['libraryItem'],
+    })
+  )
+  if (highlight) {
+    // update labels in library item
+    const jobs = await bulkEnqueueUpdateLabels([
+      { libraryItemId: highlight.libraryItem.id, userId },
+    ])
+    logger.info('update labels jobs enqueued', jobs)
+  }
 }
 
 export const findLabelsByIds = async (
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 0f3906e63..7151c8d2b 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -17,6 +17,7 @@ import {
   valuesToRawSql,
 } from '../repository'
 import { libraryItemRepository } from '../repository/library_item'
+import { bulkEnqueueUpdateLabels } from '../utils/createTask'
 import { setRecentlySavedItemInRedis, wordsCount } from '../utils/helpers'
 import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
@@ -1020,7 +1021,17 @@ export const batchUpdateLibraryItems = async (
             return !existingLabel
           })
       )
-      return tx.getRepository(EntityLabel).save(labelsToAdd)
+      const labelsAdded = await tx.getRepository(EntityLabel).save(labelsToAdd)
+
+      const data = labelsAdded.map((label) => ({
+        libraryItemId: label.libraryItemId,
+        userId,
+      }))
+      // update labels in library item
+      const jobs = await bulkEnqueueUpdateLabels(data)
+      logger.info('update labels jobs enqueued', jobs)
+
+      return
     }
 
     // generate raw sql because postgres doesn't support prepared statements in DO blocks

From 5646413b59e01d6bda8ed493181d7b617571c330 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 14:15:33 +0800
Subject: [PATCH 4/8] add a job for updating highlight annotations in
 library_item

---
 packages/api/src/jobs/update_db.ts      | 27 +++++++++++++++-
 packages/api/src/queue-processor.ts     |  9 +++++-
 packages/api/src/services/highlights.ts | 42 +++++++++++++++++++++++--
 packages/api/src/utils/createTask.ts    | 22 ++++++++++++-
 4 files changed, 95 insertions(+), 5 deletions(-)

diff --git a/packages/api/src/jobs/update_db.ts b/packages/api/src/jobs/update_db.ts
index 03e670b25..3dc7deb8b 100644
--- a/packages/api/src/jobs/update_db.ts
+++ b/packages/api/src/jobs/update_db.ts
@@ -1,6 +1,7 @@
 import { authTrx } from '../repository'
 
 export const UPDATE_LABELS_JOB = 'update-labels'
+export const UPDATE_HIGHLIGHT_JOB = 'update-highlight'
 
 export interface UpdateLabelsData {
   libraryItemId: string
@@ -20,7 +21,31 @@ export const updateLabels = async (data: UpdateLabelsData) => {
         )
         UPDATE omnivore.library_item li
         SET label_names = COALESCE((SELECT names_agg FROM labels_agg), ARRAY[]::TEXT[])
-        WHERE li.id = $1;`,
+        WHERE li.id = $1`,
+        [data.libraryItemId]
+      ),
+    undefined,
+    data.userId
+  )
+}
+
+export interface UpdateHighlightData {
+  libraryItemId: string
+  userId: string
+}
+
+export const updateHighlight = async (data: UpdateHighlightData) => {
+  return authTrx(
+    async (tx) =>
+      tx.query(
+        `WITH highlight_agg AS (
+          SELECT array_agg(COALESCE(annotation, '')) AS annotation_agg
+          FROM omnivore.highlight
+          WHERE library_item_id = $1
+        )
+        UPDATE omnivore.library_item
+        SET highlight_annotations = COALESCE((SELECT annotation_agg FROM highlight_agg), ARRAY[]::TEXT[])
+        WHERE id = $1`,
         [data.libraryItemId]
       ),
     undefined,
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index f4458e53e..f80a9d8ce 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -12,7 +12,12 @@ import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
 import { triggerRule, TRIGGER_RULE_JOB_NAME } from './jobs/trigger_rule'
-import { updateLabels, UPDATE_LABELS_JOB } from './jobs/update_db'
+import {
+  updateHighlight,
+  updateLabels,
+  UPDATE_HIGHLIGHT_JOB,
+  UPDATE_LABELS_JOB,
+} from './jobs/update_db'
 import { updatePDFContentJob } from './jobs/update_pdf_content'
 import { redisDataSource } from './redis_data_source'
 import { CustomTypeOrmLogger } from './utils/logger'
@@ -128,6 +133,8 @@ const main = async () => {
           return triggerRule(job.data)
         case UPDATE_LABELS_JOB:
           return updateLabels(job.data)
+        case UPDATE_HIGHLIGHT_JOB:
+          return updateHighlight(job.data)
       }
     },
     {
diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts
index 2d062418f..509f666b4 100644
--- a/packages/api/src/services/highlights.ts
+++ b/packages/api/src/services/highlights.ts
@@ -8,6 +8,8 @@ import { homePageURL } from '../env'
 import { createPubSubClient, EntityType } from '../pubsub'
 import { authTrx } from '../repository'
 import { highlightRepository } from '../repository/highlight'
+import { enqueueUpdateHighlight } from '../utils/createTask'
+import { logger } from '../utils/logger'
 
 type HighlightEvent = { id: string; pageId: string }
 type CreateHighlightEvent = DeepPartial<Highlight> & HighlightEvent
@@ -61,6 +63,14 @@ export const createHighlight = async (
     userId
   )
 
+  if (newHighlight.annotation) {
+    const job = await enqueueUpdateHighlight({
+      libraryItemId,
+      userId,
+    })
+    logger.info('update highlight job enqueued', job)
+  }
+
   return newHighlight
 }
 
@@ -103,6 +113,14 @@ export const mergeHighlights = async (
     userId
   )
 
+  if (newHighlight.annotation) {
+    const job = await enqueueUpdateHighlight({
+      libraryItemId,
+      userId,
+    })
+    logger.info('update highlight job enqueued', job)
+  }
+
   return newHighlight
 }
 
@@ -125,28 +143,48 @@ export const updateHighlight = async (
     })
   })
 
+  const libraryItemId = updatedHighlight.libraryItem.id
   await pubsub.entityUpdated<UpdateHighlightEvent>(
     EntityType.HIGHLIGHT,
-    { ...highlight, id: highlightId, pageId: updatedHighlight.libraryItem.id },
+    { ...highlight, id: highlightId, pageId: libraryItemId },
     userId
   )
 
+  if (highlight.annotation) {
+    const job = await enqueueUpdateHighlight({
+      libraryItemId,
+      userId,
+    })
+    logger.info('update highlight job enqueued', job)
+  }
+
   return updatedHighlight
 }
 
 export const deleteHighlightById = async (highlightId: string) => {
-  return authTrx(async (tx) => {
+  const deletedHighlight = await authTrx(async (tx) => {
     const highlightRepo = tx.withRepository(highlightRepository)
     const highlight = await highlightRepo.findOneOrFail({
       where: { id: highlightId },
       relations: {
         user: true,
+        libraryItem: true,
       },
     })
 
     await highlightRepo.delete(highlightId)
     return highlight
   })
+
+  if (deletedHighlight.annotation) {
+    const job = await enqueueUpdateHighlight({
+      libraryItemId: deletedHighlight.libraryItem.id,
+      userId: deletedHighlight.user.id,
+    })
+    logger.info('update highlight job enqueued', job)
+  }
+
+  return deletedHighlight
 }
 
 export const findHighlightById = async (
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index a662262f1..3b845c492 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -17,7 +17,12 @@ import {
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
 import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
-import { UpdateLabelsData, UPDATE_LABELS_JOB } from '../jobs/update_db'
+import {
+  UpdateHighlightData,
+  UpdateLabelsData,
+  UPDATE_HIGHLIGHT_JOB,
+  UPDATE_LABELS_JOB,
+} from '../jobs/update_db'
 import { getBackendQueue } from '../queue-processor'
 import { redisDataSource } from '../redis_data_source'
 import { signFeatureToken } from '../services/features'
@@ -682,4 +687,19 @@ export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
   }
 }
 
+export const enqueueUpdateHighlight = async (data: UpdateHighlightData) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  try {
+    return queue.add(UPDATE_HIGHLIGHT_JOB, data, {
+      priority: 1,
+    })
+  } catch (error) {
+    logger.error('error enqueuing update highlight job', error)
+  }
+}
+
 export default createHttpTaskWithToken

From cf898299e452f61c25583acc59d9b7fea72a5f08 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 14:18:34 +0800
Subject: [PATCH 5/8] some lint

---
 packages/api/src/jobs/update_db.ts | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/packages/api/src/jobs/update_db.ts b/packages/api/src/jobs/update_db.ts
index 3dc7deb8b..dcde84383 100644
--- a/packages/api/src/jobs/update_db.ts
+++ b/packages/api/src/jobs/update_db.ts
@@ -8,6 +8,11 @@ export interface UpdateLabelsData {
   userId: string
 }
 
+export interface UpdateHighlightData {
+  libraryItemId: string
+  userId: string
+}
+
 export const updateLabels = async (data: UpdateLabelsData) => {
   return authTrx(
     async (tx) =>
@@ -29,11 +34,6 @@ export const updateLabels = async (data: UpdateLabelsData) => {
   )
 }
 
-export interface UpdateHighlightData {
-  libraryItemId: string
-  userId: string
-}
-
 export const updateHighlight = async (data: UpdateHighlightData) => {
   return authTrx(
     async (tx) =>

From 5e883cb2ba12ff0ae44cd3ef5608794fb0411546 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 17:45:49 +0800
Subject: [PATCH 6/8] running worker in the test

---
 .github/workflows/run-tests.yaml              |  1 +
 packages/api/src/entity/highlight.ts          |  3 +
 packages/api/src/entity/library_item.ts       |  9 --
 packages/api/src/queue-processor.ts           | 82 +++++++++++--------
 packages/api/src/resolvers/article/index.ts   | 13 +--
 .../api/src/resolvers/function_resolvers.ts   | 25 ++----
 packages/api/src/services/highlights.ts       | 46 ++++-------
 packages/api/src/services/labels.ts           | 45 ++++------
 packages/api/src/services/library_item.ts     | 28 +++----
 packages/api/test/global-setup.ts             |  7 +-
 packages/api/test/global-teardown.ts          |  7 +-
 packages/api/test/resolvers/article.test.ts   | 15 +++-
 packages/api/test/resolvers/highlight.test.ts | 11 ++-
 packages/api/test/util.ts                     | 15 ++++
 14 files changed, 152 insertions(+), 155 deletions(-)

diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml
index 70714c14c..42196640c 100644
--- a/.github/workflows/run-tests.yaml
+++ b/.github/workflows/run-tests.yaml
@@ -93,6 +93,7 @@ jobs:
           PG_DB: omnivore_test
           PG_LOGGER: debug
           REDIS_URL: redis://localhost:${{ job.services.redis.ports[6379] }}
+          MQ_REDIS_URL: redis://localhost:${{ job.services.redis.ports[6379] }}
   build-docker-images:
     name: Build docker images
     runs-on: ubuntu-latest
diff --git a/packages/api/src/entity/highlight.ts b/packages/api/src/entity/highlight.ts
index 356c8bddc..f34ba69e8 100644
--- a/packages/api/src/entity/highlight.ts
+++ b/packages/api/src/entity/highlight.ts
@@ -35,6 +35,9 @@ export class Highlight {
   @JoinColumn({ name: 'library_item_id' })
   libraryItem!: LibraryItem
 
+  @Column('uuid')
+  libraryItemId!: string
+
   @Column('text')
   quote?: string | null
 
diff --git a/packages/api/src/entity/library_item.ts b/packages/api/src/entity/library_item.ts
index 5765c1469..ce803891f 100644
--- a/packages/api/src/entity/library_item.ts
+++ b/packages/api/src/entity/library_item.ts
@@ -183,15 +183,6 @@ export class LibraryItem {
   })
   highlights?: Highlight[]
 
-  @Column('text', { nullable: true })
-  labelNames?: string[] | null
-
-  @Column('text', { nullable: true })
-  highlightLabels?: string[] | null
-
-  @Column('text', { nullable: true })
-  highlightAnnotations?: string[] | null
-
   @Column('text', { nullable: true })
   note?: string | null
 
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index f80a9d8ce..ebf6fcfdf 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -2,7 +2,14 @@
 /* eslint-disable @typescript-eslint/restrict-template-expressions */
 /* eslint-disable @typescript-eslint/require-await */
 /* eslint-disable @typescript-eslint/no-misused-promises */
-import { Job, JobType, Queue, QueueEvents, Worker } from 'bullmq'
+import {
+  ConnectionOptions,
+  Job,
+  JobType,
+  Queue,
+  QueueEvents,
+  Worker,
+} from 'bullmq'
 import express, { Express } from 'express'
 import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
 import { appDataSource } from './data_source'
@@ -40,6 +47,43 @@ export const getBackendQueue = async (): Promise<Queue | undefined> => {
   return backendQueue
 }
 
+export const createWorker = (connection: ConnectionOptions) =>
+  new Worker(
+    QUEUE_NAME,
+    async (job: Job) => {
+      switch (job.name) {
+        case 'refresh-all-feeds': {
+          const queue = await getBackendQueue()
+          const counts = await queue?.getJobCounts('prioritized')
+          if (counts && counts.wait > 1000) {
+            return
+          }
+          return await refreshAllFeeds(appDataSource)
+        }
+        case 'refresh-feed': {
+          return await refreshFeed(job.data)
+        }
+        case 'save-page': {
+          return savePageJob(job.data, job.attemptsMade)
+        }
+        case 'update-pdf-content': {
+          return updatePDFContentJob(job.data)
+        }
+        case THUMBNAIL_JOB:
+          return findThumbnail(job.data)
+        case TRIGGER_RULE_JOB_NAME:
+          return triggerRule(job.data)
+        case UPDATE_LABELS_JOB:
+          return updateLabels(job.data)
+        case UPDATE_HIGHLIGHT_JOB:
+          return updateHighlight(job.data)
+      }
+    },
+    {
+      connection,
+    }
+  )
+
 const main = async () => {
   console.log('[queue-processor]: starting queue processor')
 
@@ -106,41 +150,7 @@ const main = async () => {
     throw '[queue-processor] error redis is not initialized'
   }
 
-  const worker = new Worker(
-    QUEUE_NAME,
-    async (job: Job) => {
-      switch (job.name) {
-        case 'refresh-all-feeds': {
-          const queue = await getBackendQueue()
-          const counts = await queue?.getJobCounts('prioritized')
-          if (counts && counts.wait > 1000) {
-            return
-          }
-          return await refreshAllFeeds(appDataSource)
-        }
-        case 'refresh-feed': {
-          return await refreshFeed(job.data)
-        }
-        case 'save-page': {
-          return savePageJob(job.data, job.attemptsMade)
-        }
-        case 'update-pdf-content': {
-          return updatePDFContentJob(job.data)
-        }
-        case THUMBNAIL_JOB:
-          return findThumbnail(job.data)
-        case TRIGGER_RULE_JOB_NAME:
-          return triggerRule(job.data)
-        case UPDATE_LABELS_JOB:
-          return updateLabels(job.data)
-        case UPDATE_HIGHLIGHT_JOB:
-          return updateHighlight(job.data)
-      }
-    },
-    {
-      connection: workerRedisClient,
-    }
-  )
+  const worker = createWorker(workerRedisClient)
 
   const queueEvents = new QueueEvents(QUEUE_NAME, {
     connection: workerRedisClient,
diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts
index 93710f98f..921a01cf8 100644
--- a/packages/api/src/resolvers/article/index.ts
+++ b/packages/api/src/resolvers/article/index.ts
@@ -701,15 +701,10 @@ export const searchResolver = authorized<
 
   const edges = await Promise.all(
     libraryItems.map(async (libraryItem) => {
-      if (
-        libraryItem.highlightAnnotations &&
-        libraryItem.highlightAnnotations.length > 0
-      ) {
-        libraryItem.highlights = await findHighlightsByLibraryItemId(
-          libraryItem.id,
-          uid
-        )
-      }
+      libraryItem.highlights = await findHighlightsByLibraryItemId(
+        libraryItem.id,
+        uid
+      )
 
       if (params.includeContent && libraryItem.readableContent) {
         // convert html to the requested format
diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts
index ad3e3ac96..871782068 100644
--- a/packages/api/src/resolvers/function_resolvers.ts
+++ b/packages/api/src/resolvers/function_resolvers.ts
@@ -334,17 +334,13 @@ export const functionResolvers = {
       return article.content ? wordsCount(article.content) : undefined
     },
     async labels(
-      article: { id: string; labels?: Label[]; labelNames?: string[] | null },
+      article: { id: string; labels?: Label[] },
       _: unknown,
       ctx: WithDataSourcesContext
     ) {
       if (article.labels) return article.labels
 
-      if (article.labelNames && article.labelNames.length > 0) {
-        return findLabelsByLibraryItemId(article.id, ctx.uid)
-      }
-
-      return []
+      return findLabelsByLibraryItemId(article.id, ctx.uid)
     },
   },
   Highlight: {
@@ -409,17 +405,13 @@ export const functionResolvers = {
       return item.siteIcon
     },
     async labels(
-      item: { id: string; labels?: Label[]; labelNames?: string[] | null },
+      item: { id: string; labels?: Label[] },
       _: unknown,
       ctx: WithDataSourcesContext
     ) {
       if (item.labels) return item.labels
 
-      if (item.labelNames && item.labelNames.length > 0) {
-        return findLabelsByLibraryItemId(item.id, ctx.uid)
-      }
-
-      return []
+      return findLabelsByLibraryItemId(item.id, ctx.uid)
     },
     async recommendations(
       item: {
@@ -446,19 +438,14 @@ export const functionResolvers = {
       item: {
         id: string
         highlights?: Highlight[]
-        highlightAnnotations?: string[] | null
       },
       _: unknown,
       ctx: WithDataSourcesContext
     ) {
       if (item.highlights) return item.highlights
 
-      if (item.highlightAnnotations && item.highlightAnnotations.length > 0) {
-        const highlights = await findHighlightsByLibraryItemId(item.id, ctx.uid)
-        return highlights.map(highlightDataToHighlight)
-      }
-
-      return []
+      const highlights = await findHighlightsByLibraryItemId(item.id, ctx.uid)
+      return highlights.map(highlightDataToHighlight)
     },
   },
   Subscription: {
diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts
index 509f666b4..436a3dd1a 100644
--- a/packages/api/src/services/highlights.ts
+++ b/packages/api/src/services/highlights.ts
@@ -9,7 +9,6 @@ import { createPubSubClient, EntityType } from '../pubsub'
 import { authTrx } from '../repository'
 import { highlightRepository } from '../repository/highlight'
 import { enqueueUpdateHighlight } from '../utils/createTask'
-import { logger } from '../utils/logger'
 
 type HighlightEvent = { id: string; pageId: string }
 type CreateHighlightEvent = DeepPartial<Highlight> & HighlightEvent
@@ -63,13 +62,10 @@ export const createHighlight = async (
     userId
   )
 
-  if (newHighlight.annotation) {
-    const job = await enqueueUpdateHighlight({
-      libraryItemId,
-      userId,
-    })
-    logger.info('update highlight job enqueued', job)
-  }
+  await enqueueUpdateHighlight({
+    libraryItemId,
+    userId,
+  })
 
   return newHighlight
 }
@@ -113,13 +109,10 @@ export const mergeHighlights = async (
     userId
   )
 
-  if (newHighlight.annotation) {
-    const job = await enqueueUpdateHighlight({
-      libraryItemId,
-      userId,
-    })
-    logger.info('update highlight job enqueued', job)
-  }
+  await enqueueUpdateHighlight({
+    libraryItemId,
+    userId,
+  })
 
   return newHighlight
 }
@@ -150,13 +143,10 @@ export const updateHighlight = async (
     userId
   )
 
-  if (highlight.annotation) {
-    const job = await enqueueUpdateHighlight({
-      libraryItemId,
-      userId,
-    })
-    logger.info('update highlight job enqueued', job)
-  }
+  await enqueueUpdateHighlight({
+    libraryItemId,
+    userId,
+  })
 
   return updatedHighlight
 }
@@ -168,7 +158,6 @@ export const deleteHighlightById = async (highlightId: string) => {
       where: { id: highlightId },
       relations: {
         user: true,
-        libraryItem: true,
       },
     })
 
@@ -176,13 +165,10 @@ export const deleteHighlightById = async (highlightId: string) => {
     return highlight
   })
 
-  if (deletedHighlight.annotation) {
-    const job = await enqueueUpdateHighlight({
-      libraryItemId: deletedHighlight.libraryItem.id,
-      userId: deletedHighlight.user.id,
-    })
-    logger.info('update highlight job enqueued', job)
-  }
+  await enqueueUpdateHighlight({
+    libraryItemId: deletedHighlight.libraryItemId,
+    userId: deletedHighlight.user.id,
+  })
 
   return deletedHighlight
 }
diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index 8ad7c24a6..8d98514c2 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -1,13 +1,12 @@
 import { DeepPartial, FindOptionsWhere, In } from 'typeorm'
 import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'
 import { EntityLabel, LabelSource } from '../entity/entity_label'
-import { Highlight } from '../entity/highlight'
 import { Label } from '../entity/label'
 import { createPubSubClient, EntityType, PubsubClient } from '../pubsub'
 import { authTrx } from '../repository'
 import { CreateLabelInput, labelRepository } from '../repository/label'
 import { bulkEnqueueUpdateLabels } from '../utils/createTask'
-import { logger } from '../utils/logger'
+import { findHighlightById } from './highlights'
 import { findLibraryItemIdsByLabelId } from './library_item'
 
 type AddLabelsToLibraryItemEvent = {
@@ -126,8 +125,7 @@ export const saveLabelsInLibraryItem = async (
   }
 
   // update labels in library item
-  const jobs = await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
-  logger.info('update labels jobs enqueued', jobs)
+  await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
 }
 
 export const addLabelsToLibraryItem = async (
@@ -155,8 +153,7 @@ export const addLabelsToLibraryItem = async (
   )
 
   // update labels in library item
-  const jobs = await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
-  logger.info('update labels jobs enqueued', jobs)
+  await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
 }
 
 export const saveLabelsInHighlight = async (
@@ -189,19 +186,11 @@ export const saveLabelsInHighlight = async (
     userId
   )
 
-  const highlight = await authTrx(async (tx) =>
-    tx.getRepository(Highlight).findOne({
-      where: { id: highlightId },
-      relations: ['libraryItem'],
-    })
-  )
-  if (highlight) {
-    // update labels in library item
-    const jobs = await bulkEnqueueUpdateLabels([
-      { libraryItemId: highlight.libraryItem.id, userId },
-    ])
-    logger.info('update labels jobs enqueued', jobs)
-  }
+  const highlight = await findHighlightById(highlightId, userId)
+  // update labels in library item
+  await bulkEnqueueUpdateLabels([
+    { libraryItemId: highlight.libraryItemId, userId },
+  ])
 }
 
 export const findLabelsByIds = async (
@@ -259,8 +248,7 @@ export const deleteLabelById = async (labelId: string, userId: string) => {
     libraryItemId,
     userId,
   }))
-  const jobs = await bulkEnqueueUpdateLabels(data)
-  logger.info('update labels jobs enqueued', jobs)
+  await bulkEnqueueUpdateLabels(data)
 
   return true
 }
@@ -281,16 +269,13 @@ export const updateLabel = async (
     userId
   )
 
-  if (label.name) {
-    const libraryItemIds = await findLibraryItemIdsByLabelId(id, userId)
+  const libraryItemIds = await findLibraryItemIdsByLabelId(id, userId)
 
-    const data = libraryItemIds.map((libraryItemId) => ({
-      libraryItemId,
-      userId,
-    }))
-    const jobs = await bulkEnqueueUpdateLabels(data)
-    logger.info('update labels jobs enqueued', jobs)
-  }
+  const data = libraryItemIds.map((libraryItemId) => ({
+    libraryItemId,
+    userId,
+  }))
+  await bulkEnqueueUpdateLabels(data)
 
   return updatedLabel
 }
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 7151c8d2b..d8c927151 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1008,30 +1008,24 @@ export const batchUpdateLibraryItems = async (
       const libraryItems = await queryBuilder.getMany()
       // add labels in library items
       const labelsToAdd = libraryItems.flatMap((libraryItem) =>
-        labels
-          .map((label) => ({
-            labelId: label.id,
-            libraryItemId: libraryItem.id,
-            name: label.name,
-          }))
-          .filter((entityLabel) => {
-            const existingLabel = libraryItem.labelNames?.find(
-              (l) => l.toLowerCase() === entityLabel.name.toLowerCase()
-            )
-            return !existingLabel
-          })
+        labels.map((label) => ({
+          labelId: label.id,
+          libraryItemId: libraryItem.id,
+          name: label.name,
+          // put an zero uuid for highlight to avoid unique constraint violation
+          highlightId: '00000000-0000-0000-0000-000000000000',
+        }))
       )
       const labelsAdded = await tx.getRepository(EntityLabel).save(labelsToAdd)
 
-      const data = labelsAdded.map((label) => ({
-        libraryItemId: label.libraryItemId,
+      const data = libraryItems.map((item) => ({
+        libraryItemId: item.id,
         userId,
       }))
       // update labels in library item
-      const jobs = await bulkEnqueueUpdateLabels(data)
-      logger.info('update labels jobs enqueued', jobs)
+      await bulkEnqueueUpdateLabels(data)
 
-      return
+      return labelsAdded
     }
 
     // generate raw sql because postgres doesn't support prepared statements in DO blocks
diff --git a/packages/api/test/global-setup.ts b/packages/api/test/global-setup.ts
index 31f9a3d86..56c82a57f 100644
--- a/packages/api/test/global-setup.ts
+++ b/packages/api/test/global-setup.ts
@@ -1,7 +1,7 @@
 import { env } from '../src/env'
 import { redisDataSource } from '../src/redis_data_source'
 import { createTestConnection } from './db'
-import { startApolloServer } from './util'
+import { startApolloServer, startWorker } from './util'
 
 export const mochaGlobalSetup = async () => {
   await createTestConnection()
@@ -10,6 +10,11 @@ export const mochaGlobalSetup = async () => {
   if (env.redis.cache.url) {
     await redisDataSource.initialize()
     console.log('redis connection created')
+
+    if (redisDataSource.workerRedisClient) {
+      startWorker(redisDataSource.workerRedisClient)
+      console.log('worker started')
+    }
   }
 
   await startApolloServer()
diff --git a/packages/api/test/global-teardown.ts b/packages/api/test/global-teardown.ts
index 2e476c55e..45b0760b6 100644
--- a/packages/api/test/global-teardown.ts
+++ b/packages/api/test/global-teardown.ts
@@ -1,7 +1,7 @@
 import { appDataSource } from '../src/data_source'
 import { env } from '../src/env'
 import { redisDataSource } from '../src/redis_data_source'
-import { stopApolloServer } from './util'
+import { stopApolloServer, stopWorker } from './util'
 
 export const mochaGlobalTeardown = async () => {
   await stopApolloServer()
@@ -13,5 +13,10 @@ export const mochaGlobalTeardown = async () => {
   if (env.redis.cache.url) {
     await redisDataSource.shutdown()
     console.log('redis connection closed')
+
+    if (redisDataSource.workerRedisClient) {
+      stopWorker()
+      console.log('worker closed')
+    }
   }
 }
diff --git a/packages/api/test/resolvers/article.test.ts b/packages/api/test/resolvers/article.test.ts
index c25d1447f..3fb6405f6 100644
--- a/packages/api/test/resolvers/article.test.ts
+++ b/packages/api/test/resolvers/article.test.ts
@@ -42,7 +42,12 @@ import { deleteUser } from '../../src/services/user'
 import * as createTask from '../../src/utils/createTask'
 import * as uploads from '../../src/utils/uploads'
 import { createTestLibraryItem, createTestUser } from '../db'
-import { generateFakeUuid, graphqlRequest, request } from '../util'
+import {
+  generateFakeUuid,
+  graphqlRequest,
+  request,
+  waitUntilJobsDone,
+} from '../util'
 
 chai.use(chaiString)
 
@@ -942,6 +947,8 @@ describe('Article API', () => {
         )
         highlights.push(highlight)
       }
+
+      await waitUntilJobsDone()
     })
 
     beforeEach(async () => {
@@ -1287,6 +1294,8 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
+
+        await waitUntilJobsDone()
       })
 
       after(async () => {
@@ -1784,6 +1793,8 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
+
+        await waitUntilJobsDone()
       })
 
       after(async () => {
@@ -1842,6 +1853,8 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
+
+        await waitUntilJobsDone()
       })
 
       after(async () => {
diff --git a/packages/api/test/resolvers/highlight.test.ts b/packages/api/test/resolvers/highlight.test.ts
index e0fe8a668..1d839d515 100644
--- a/packages/api/test/resolvers/highlight.test.ts
+++ b/packages/api/test/resolvers/highlight.test.ts
@@ -2,7 +2,9 @@ import * as chai from 'chai'
 import { expect } from 'chai'
 import chaiString from 'chai-string'
 import 'mocha'
+import { Highlight } from '../../src/entity/highlight'
 import { User } from '../../src/entity/user'
+import { getRepository } from '../../src/repository'
 import {
   createHighlight,
   deleteHighlightById,
@@ -11,7 +13,12 @@ import {
 import { createLabel, saveLabelsInHighlight } from '../../src/services/labels'
 import { deleteUser } from '../../src/services/user'
 import { createTestLibraryItem, createTestUser } from '../db'
-import { generateFakeShortId, generateFakeUuid, graphqlRequest, request } from '../util'
+import {
+  generateFakeShortId,
+  generateFakeUuid,
+  graphqlRequest,
+  request,
+} from '../util'
 
 chai.use(chaiString)
 
@@ -281,7 +288,7 @@ describe('Highlights API', () => {
         itemId,
         newHighlightId,
         newShortHighlightId,
-        [highlightId],
+        [highlightId]
       )
       const res = await graphqlRequest(query, authToken).expect(200)
 
diff --git a/packages/api/test/util.ts b/packages/api/test/util.ts
index 5d2567f83..cfd2976a3 100644
--- a/packages/api/test/util.ts
+++ b/packages/api/test/util.ts
@@ -1,11 +1,14 @@
+import { ConnectionOptions, Worker } from 'bullmq'
 import { nanoid } from 'nanoid'
 import supertest from 'supertest'
 import { v4 } from 'uuid'
+import { createWorker } from '../src/queue-processor'
 import { createApp } from '../src/server'
 import { corsConfig } from '../src/utils/corsConfig'
 
 const { app, apollo } = createApp()
 export const request = supertest(app)
+let worker: Worker | undefined
 
 export const startApolloServer = async () => {
   await apollo.start()
@@ -16,6 +19,18 @@ export const stopApolloServer = async () => {
   await apollo.stop()
 }
 
+export const startWorker = async (connection: ConnectionOptions) => {
+  worker = createWorker(connection)
+}
+
+export const stopWorker = async () => {
+  worker?.close()
+}
+
+export const waitUntilJobsDone = async () => {
+  await worker?.waitUntilReady()
+}
+
 export const graphqlRequest = (
   query: string,
   authToken: string,

From 1b8111b90783911dfb2658da654d0c5184e2548d Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 19:10:57 +0800
Subject: [PATCH 7/8] wait until all jobs are done in test

---
 packages/api/src/services/labels.ts         |  2 +-
 packages/api/src/utils/createTask.ts        |  3 +-
 packages/api/test/db.ts                     | 77 ++++++++++++++++++++-
 packages/api/test/resolvers/article.test.ts | 27 ++------
 packages/api/test/util.ts                   | 17 +++--
 5 files changed, 95 insertions(+), 31 deletions(-)

diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts
index 8d98514c2..c4f8d4d13 100644
--- a/packages/api/src/services/labels.ts
+++ b/packages/api/src/services/labels.ts
@@ -125,7 +125,7 @@ export const saveLabelsInLibraryItem = async (
   }
 
   // update labels in library item
-  await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
+  return bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
 }
 
 export const addLabelsToLibraryItem = async (
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index 3b845c492..1b06be0a3 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -669,7 +669,7 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => {
 export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
   const queue = await getBackendQueue()
   if (!queue) {
-    return undefined
+    return []
   }
 
   const jobs = data.map((d) => ({
@@ -684,6 +684,7 @@ export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
     return queue.addBulk(jobs)
   } catch (error) {
     logger.error('error enqueuing update labels jobs', error)
+    return []
   }
 }
 
diff --git a/packages/api/test/db.ts b/packages/api/test/db.ts
index 26a02e965..01c9f9b6a 100644
--- a/packages/api/test/db.ts
+++ b/packages/api/test/db.ts
@@ -1,18 +1,24 @@
 import { DeepPartial } from 'typeorm'
 import { appDataSource } from '../src/data_source'
+import { EntityLabel, LabelSource } from '../src/entity/entity_label'
 import { Filter } from '../src/entity/filter'
+import { Highlight } from '../src/entity/highlight'
 import { Label } from '../src/entity/label'
 import { LibraryItem } from '../src/entity/library_item'
 import { Reminder } from '../src/entity/reminder'
 import { User } from '../src/entity/user'
 import { UserDeviceToken } from '../src/entity/user_device_tokens'
-import { getRepository, setClaims } from '../src/repository'
+import { authTrx, getRepository, setClaims } from '../src/repository'
+import { highlightRepository } from '../src/repository/highlight'
 import { userRepository } from '../src/repository/user'
 import { createUser } from '../src/services/create_user'
-import { saveLabelsInLibraryItem } from '../src/services/labels'
 import { createLibraryItem } from '../src/services/library_item'
 import { createDeviceToken } from '../src/services/user_device_tokens'
-import { generateFakeUuid } from './util'
+import {
+  bulkEnqueueUpdateLabels,
+  enqueueUpdateHighlight,
+} from '../src/utils/createTask'
+import { generateFakeUuid, waitUntilJobsDone } from './util'
 
 export const createTestConnection = async (): Promise<void> => {
   appDataSource.setOptions({
@@ -121,3 +127,68 @@ export const createTestLibraryItem = async (
 
   return createdItem
 }
+
+export const saveLabelsInLibraryItem = async (
+  labels: Label[],
+  libraryItemId: string,
+  userId: string,
+  source: LabelSource = 'user'
+) => {
+  await authTrx(
+    async (tx) => {
+      const repo = tx.getRepository(EntityLabel)
+
+      // delete existing labels
+      await repo.delete({
+        libraryItemId,
+      })
+
+      // save new labels
+      await repo.save(
+        labels.map((l) => ({
+          labelId: l.id,
+          libraryItemId,
+          source,
+        }))
+      )
+    },
+    undefined,
+    userId
+  )
+
+  // update labels in library item
+  const jobs = await bulkEnqueueUpdateLabels([{ libraryItemId, userId }])
+
+  await waitUntilJobsDone(jobs)
+}
+
+export const createHighlight = async (
+  highlight: DeepPartial<Highlight>,
+  libraryItemId: string,
+  userId: string
+) => {
+  const newHighlight = await authTrx(
+    async (tx) => {
+      const repo = tx.withRepository(highlightRepository)
+      const newHighlight = await repo.createAndSave(highlight)
+      return repo.findOneOrFail({
+        where: { id: newHighlight.id },
+        relations: {
+          user: true,
+        },
+      })
+    },
+    undefined,
+    userId
+  )
+
+  const job = await enqueueUpdateHighlight({
+    libraryItemId,
+    userId,
+  })
+  if (job) {
+    await waitUntilJobsDone([job])
+  }
+
+  return newHighlight
+}
diff --git a/packages/api/test/resolvers/article.test.ts b/packages/api/test/resolvers/article.test.ts
index 3fb6405f6..1b49a5a83 100644
--- a/packages/api/test/resolvers/article.test.ts
+++ b/packages/api/test/resolvers/article.test.ts
@@ -21,12 +21,7 @@ import {
 } from '../../src/generated/graphql'
 import { getRepository } from '../../src/repository'
 import { createGroup, deleteGroup } from '../../src/services/groups'
-import { createHighlight } from '../../src/services/highlights'
-import {
-  createLabel,
-  deleteLabels,
-  saveLabelsInLibraryItem,
-} from '../../src/services/labels'
+import { createLabel, deleteLabels } from '../../src/services/labels'
 import {
   createLibraryItem,
   createLibraryItems,
@@ -41,13 +36,13 @@ import {
 import { deleteUser } from '../../src/services/user'
 import * as createTask from '../../src/utils/createTask'
 import * as uploads from '../../src/utils/uploads'
-import { createTestLibraryItem, createTestUser } from '../db'
 import {
-  generateFakeUuid,
-  graphqlRequest,
-  request,
-  waitUntilJobsDone,
-} from '../util'
+  createHighlight,
+  createTestLibraryItem,
+  createTestUser,
+  saveLabelsInLibraryItem,
+} from '../db'
+import { generateFakeUuid, graphqlRequest, request } from '../util'
 
 chai.use(chaiString)
 
@@ -947,8 +942,6 @@ describe('Article API', () => {
         )
         highlights.push(highlight)
       }
-
-      await waitUntilJobsDone()
     })
 
     beforeEach(async () => {
@@ -1294,8 +1287,6 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
-
-        await waitUntilJobsDone()
       })
 
       after(async () => {
@@ -1793,8 +1784,6 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
-
-        await waitUntilJobsDone()
       })
 
       after(async () => {
@@ -1853,8 +1842,6 @@ describe('Article API', () => {
         )
         await saveLabelsInLibraryItem([label1], items[0].id, user.id)
         await saveLabelsInLibraryItem([label2], items[1].id, user.id)
-
-        await waitUntilJobsDone()
       })
 
       after(async () => {
diff --git a/packages/api/test/util.ts b/packages/api/test/util.ts
index cfd2976a3..2cae36704 100644
--- a/packages/api/test/util.ts
+++ b/packages/api/test/util.ts
@@ -1,14 +1,15 @@
-import { ConnectionOptions, Worker } from 'bullmq'
+import { ConnectionOptions, Job, QueueEvents, Worker } from 'bullmq'
 import { nanoid } from 'nanoid'
 import supertest from 'supertest'
 import { v4 } from 'uuid'
-import { createWorker } from '../src/queue-processor'
+import { createWorker, QUEUE_NAME } from '../src/queue-processor'
 import { createApp } from '../src/server'
 import { corsConfig } from '../src/utils/corsConfig'
 
 const { app, apollo } = createApp()
 export const request = supertest(app)
-let worker: Worker | undefined
+let worker: Worker
+let queueEvents: QueueEvents
 
 export const startApolloServer = async () => {
   await apollo.start()
@@ -21,14 +22,18 @@ export const stopApolloServer = async () => {
 
 export const startWorker = async (connection: ConnectionOptions) => {
   worker = createWorker(connection)
+  queueEvents = new QueueEvents(QUEUE_NAME, {
+    connection,
+  })
 }
 
 export const stopWorker = async () => {
-  worker?.close()
+  queueEvents.close()
+  worker.close()
 }
 
-export const waitUntilJobsDone = async () => {
-  await worker?.waitUntilReady()
+export const waitUntilJobsDone = async (jobs: Job[]) => {
+  await Promise.all(jobs.map((job) => job.waitUntilFinished(queueEvents)))
 }
 
 export const graphqlRequest = (

From 6d5ce451ac300e9c03b3ec62aebb6136bf183334 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 31 Jan 2024 19:24:18 +0800
Subject: [PATCH 8/8] fix bulk action

---
 packages/api/src/services/library_item.ts | 26 ++++++-----------------
 1 file changed, 7 insertions(+), 19 deletions(-)

diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index d8c927151..6aa52eeb1 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -2,7 +2,6 @@ import { ExpressionToken, LiqeQuery } from '@omnivore/liqe'
 import { DateTime } from 'luxon'
 import { DeepPartial, FindOptionsWhere, ObjectLiteral } from 'typeorm'
 import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'
-import { EntityLabel } from '../entity/entity_label'
 import { Highlight } from '../entity/highlight'
 import { Label } from '../entity/label'
 import { LibraryItem, LibraryItemState } from '../entity/library_item'
@@ -17,10 +16,10 @@ import {
   valuesToRawSql,
 } from '../repository'
 import { libraryItemRepository } from '../repository/library_item'
-import { bulkEnqueueUpdateLabels } from '../utils/createTask'
 import { setRecentlySavedItemInRedis, wordsCount } from '../utils/helpers'
 import { logger } from '../utils/logger'
 import { parseSearchQuery } from '../utils/search'
+import { addLabelsToLibraryItem } from './labels'
 
 enum ReadFilter {
   ALL = 'all',
@@ -1005,27 +1004,16 @@ export const batchUpdateLibraryItems = async (
         throw new Error('Labels are required for this action')
       }
 
+      const labelIds = labels.map((label) => label.id)
       const libraryItems = await queryBuilder.getMany()
       // add labels in library items
-      const labelsToAdd = libraryItems.flatMap((libraryItem) =>
-        labels.map((label) => ({
-          labelId: label.id,
-          libraryItemId: libraryItem.id,
-          name: label.name,
-          // put an zero uuid for highlight to avoid unique constraint violation
-          highlightId: '00000000-0000-0000-0000-000000000000',
-        }))
-      )
-      const labelsAdded = await tx.getRepository(EntityLabel).save(labelsToAdd)
+      for (const libraryItem of libraryItems) {
+        await addLabelsToLibraryItem(labelIds, libraryItem.id, userId)
 
-      const data = libraryItems.map((item) => ({
-        libraryItemId: item.id,
-        userId,
-      }))
-      // update labels in library item
-      await bulkEnqueueUpdateLabels(data)
+        libraryItem.labels = labels
+      }
 
-      return labelsAdded
+      return
     }
 
     // generate raw sql because postgres doesn't support prepared statements in DO blocks