Merge pull request #3846 from omnivore-app/fix/bulk-action

fix: export all items job stuck

packages/api/src/jobs/bulk_action.ts

@@ -23,9 +23,8 @@ export const bulkAction = async (data: BulkActionData) => {
     throw new Error('Queue not initialized')
   }
   const now = new Date().toISOString()
-  let offset = 0
 
-  do {
+  for (let offset = 0; offset < count; offset += batchSize) {
     const searchArgs = {
       size: batchSize,
       query: `(${query}) AND updated:*..${now}`, // only process items that have not been updated
@@ -36,9 +35,7 @@ export const bulkAction = async (data: BulkActionData) => {
     } catch (error) {
       logger.error('batch update error', error)
     }
-
-    offset += batchSize
-  } while (offset < count)
+  }
 
   return true
 }

packages/api/src/jobs/integration/export_all_items.ts

@@ -50,9 +50,9 @@ export const exportAllItems = async (jobData: ExportAllItemsJobData) => {
 
   const maxItems = 100
   const limit = 10
-  let offset = 0
+  let exported = 0
   // get max 100 most recent items from the database
-  while (offset < maxItems) {
+  for (let offset = 0; offset < maxItems; offset += limit) {
     const libraryItems = await findRecentLibraryItems(userId, limit, offset)
     if (libraryItems.length === 0) {
       logger.info('no library items found', {
@@ -92,17 +92,17 @@ export const exportAllItems = async (jobData: ExportAllItemsJobData) => {
       updated,
     })
 
-    offset += libraryItems.length
+    exported += libraryItems.length
 
     logger.info('exported items', {
       ...jobData,
-      offset,
+      exported,
     })
   }
 
   logger.info('exported all items', {
     ...jobData,
-    offset,
+    exported,
   })
 
   // clear task name in integration

packages/api/src/queue-processor.ts

@@ -19,6 +19,17 @@ import { aiSummarize, AI_SUMMARIZE_JOB_NAME } from './jobs/ai-summarize'
 import { createDigestJob, CREATE_DIGEST_JOB } from './jobs/ai/create_digest'
 import { bulkAction, BULK_ACTION_JOB_NAME } from './jobs/bulk_action'
 import { callWebhook, CALL_WEBHOOK_JOB_NAME } from './jobs/call_webhook'
+import {
+  confirmEmailJob,
+  CONFIRM_EMAIL_JOB,
+  forwardEmailJob,
+  FORWARD_EMAIL_JOB,
+  saveAttachmentJob,
+  saveNewsletterJob,
+  SAVE_ATTACHMENT_JOB,
+  SAVE_NEWSLETTER_JOB,
+} from './jobs/email/inbound_emails'
+import { sendEmailJob, SEND_EMAIL_JOB } from './jobs/email/send_email'
 import { findThumbnail, THUMBNAIL_JOB } from './jobs/find_thumbnail'
 import {
   exportAllItems,
@@ -37,7 +48,6 @@ import {
 import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
-import { sendEmailJob, SEND_EMAIL_JOB } from './jobs/email/send_email'
 import {
   syncReadPositionsJob,
   SYNC_READ_POSITIONS_JOB_NAME,
@@ -54,16 +64,6 @@ import { redisDataSource } from './redis_data_source'
 import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position'
 import { getJobPriority } from './utils/createTask'
 import { logger } from './utils/logger'
-import {
-  confirmEmailJob,
-  CONFIRM_EMAIL_JOB,
-  forwardEmailJob,
-  FORWARD_EMAIL_JOB,
-  saveAttachmentJob,
-  saveNewsletterJob,
-  SAVE_ATTACHMENT_JOB,
-  SAVE_NEWSLETTER_JOB,
-} from './jobs/email/inbound_emails'
 
 export const QUEUE_NAME = 'omnivore-backend-queue'
 export const JOB_VERSION = 'v001'
@@ -188,6 +188,8 @@ export const createWorker = (connection: ConnectionOptions) =>
     },
     {
       connection,
+      autorun: true, // start processing jobs immediately
+      lockDuration: 60_000, // 1 minute
     }
   )
 
@@ -316,6 +318,10 @@ const main = async () => {
     console.log('completed job: ', job.jobId)
   })
 
+  queueEvents.on('failed', async (job) => {
+    console.log('failed job: ', job.jobId)
+  })
+
   workerRedisClient.on('error', (error) => {
     console.trace('[queue-processor]: redis worker error', { error })
   })
@@ -337,8 +343,14 @@ const main = async () => {
       })
     })
     await worker.close()
+    console.log('[queue-processor]: Worker closed')
+
     await redisDataSource.shutdown()
+    console.log('[queue-processor]: Redis connection closed')
+
     await appDataSource.destroy()
+    console.log('[queue-processor]: DB connection closed')
+
     process.exit(0)
   }
 

packages/api/src/server.ts

@@ -178,9 +178,8 @@ const main = async (): Promise<void> => {
     await apollo.stop()
     console.log('[api]: Express server stopped')
 
-    console.log('[posthog]: flushing events')
     await analytics.shutdownAsync()
-    console.log('[posthog]: events flushed')
+    console.log('[api]: Posthog events flushed')
 
     // Shutdown redis before DB because the quit sequence can
     // cause appDataSource to get reloaded in the callback

packages/api/src/services/library_item.ts

@@ -627,7 +627,9 @@ export const buildQuery = (
   // select all columns except content
   const selects: Select[] = getColumns(libraryItemRepository)
     .filter(
-      (select) => select !== 'readableContent' && select !== 'originalContent'
+      (select) =>
+        select !== 'originalContent' && // exclude original content
+        (args.includeContent || select !== 'readableContent') // exclude content if not requested
     )
     .map((column) => ({ column: `library_item.${column}` }))
 
@@ -647,20 +649,16 @@ export const buildQuery = (
       args.useFolders
     )
   }
-  queryBuilder.where('library_item.user_id = :userId', { userId })
 
   // add select
   selects.forEach((select, index) => {
-    if (index === 0) {
-      queryBuilder.select(select.column, select.alias)
-    }
-
-    queryBuilder.addSelect(select.column, select.alias)
+    // select must be defined before adding additional selects
+    index === 0
+      ? queryBuilder.select(select.column, select.alias)
+      : queryBuilder.addSelect(select.column, select.alias)
   })
 
-  if (args.includeContent) {
-    queryBuilder.addSelect('library_item.readableContent')
-  }
+  queryBuilder.where('library_item.user_id = :userId', { userId })
 
   if (!args.includePending) {
     queryBuilder.andWhere("library_item.state <> 'PROCESSING'")
@@ -1117,6 +1115,13 @@ export const batchUpdateLibraryItems = async (
   labelIds?: string[] | null,
   args?: unknown
 ) => {
+  if (!searchArgs.query) {
+    throw new Error('Search query is required')
+  }
+
+  const searchQuery = parseSearchQuery(searchArgs.query)
+  const parameters: ObjectLiteral[] = []
+  const queryString = buildQueryString(searchQuery, parameters)
   interface FolderArguments {
     folder: string
   }
@@ -1139,19 +1144,23 @@ export const batchUpdateLibraryItems = async (
 
   const getLibraryItemIds = async (
     userId: string,
-    em: EntityManager
-  ): Promise<{ id: string }[]> => {
+    em: EntityManager,
+    forUpdate = false
+  ): Promise<string[]> => {
     const queryBuilder = getQueryBuilder(userId, em)
-    return queryBuilder.select('library_item.id', 'id').getRawMany()
-  }
 
-  if (!searchArgs.query) {
-    throw new Error('Search query is required')
-  }
+    if (forUpdate) {
+      queryBuilder.setLock('pessimistic_write')
+    }
 
-  const searchQuery = parseSearchQuery(searchArgs.query)
-  const parameters: ObjectLiteral[] = []
-  const queryString = buildQueryString(searchQuery, parameters)
+    const libraryItems = await queryBuilder
+      .select('library_item.id', 'id')
+      .take(searchArgs.size)
+      .skip(searchArgs.from)
+      .getRawMany<{ id: string }>()
+
+    return libraryItems.map((item) => item.id)
+  }
 
   const now = new Date().toISOString()
   // build the script
@@ -1174,27 +1183,27 @@ export const batchUpdateLibraryItems = async (
         throw new Error('Labels are required for this action')
       }
 
-      const libraryItems = await authTrx(
+      const libraryItemIds = await authTrx(
         async (tx) => getLibraryItemIds(userId, tx),
         undefined,
         userId
       )
       // add labels to library items
-      for (const libraryItem of libraryItems) {
-        await addLabelsToLibraryItem(labelIds, libraryItem.id, userId)
+      for (const libraryItemId of libraryItemIds) {
+        await addLabelsToLibraryItem(labelIds, libraryItemId, userId)
       }
 
       return
     }
     case BulkActionType.MarkAsRead: {
-      const libraryItems = await authTrx(
+      const libraryItemIds = await authTrx(
         async (tx) => getLibraryItemIds(userId, tx),
         undefined,
         userId
       )
       // update reading progress for library items
-      for (const libraryItem of libraryItems) {
-        await markItemAsRead(libraryItem.id, userId)
+      for (const libraryItemId of libraryItemIds) {
+        await markItemAsRead(libraryItemId, userId)
       }
 
       return
@@ -1215,8 +1224,10 @@ export const batchUpdateLibraryItems = async (
   }
 
   await authTrx(
-    async (tx) =>
-      getQueryBuilder(userId, tx).update(LibraryItem).set(values).execute(),
+    async (tx) => {
+      const libraryItemIds = await getLibraryItemIds(userId, tx, true)
+      await tx.getRepository(LibraryItem).update(libraryItemIds, values)
+    },
     undefined,
     userId
   )

packages/api/src/utils/highlightGenerator.ts

@@ -49,7 +49,7 @@ type FillNodeResponse = {
 }
 
 function getTextNodesBetween(rootNode: Node, startNode: Node, endNode: Node) {
-  const maxTime = 1000 * 60 * 10 // 10 minutes
+  const maxTime = 1000 * 60 // 60 seconds
   const start = Date.now()
   let textNodeStartingPoint = 0
   let articleText = ''

packages/content-fetch/src/job.ts

@@ -4,9 +4,24 @@ import { redisDataSource } from './redis_data_source'
 const QUEUE_NAME = 'omnivore-backend-queue'
 const JOB_NAME = 'save-page'
 
-interface savePageJob {
+interface SavePageJobData {
   userId: string
-  data: unknown
+  url: string
+  finalUrl: string
+  articleSavingRequestId: string
+  state?: string
+  labels?: string[]
+  source: string
+  folder?: string
+  rssFeedUrl?: string
+  savedAt?: string
+  publishedAt?: string
+  taskId?: string
+}
+
+interface SavePageJob {
+  userId: string
+  data: SavePageJobData
   isRss: boolean
   isImport: boolean
   priority: 'low' | 'high'
@@ -16,7 +31,7 @@ const queue = new Queue(QUEUE_NAME, {
   connection: redisDataSource.queueRedisClient,
 })
 
-const getPriority = (job: savePageJob): number => {
+const getPriority = (job: SavePageJob): number => {
   // we want to prioritized jobs by the expected time to complete
   // lower number means higher priority
   // priority 1: jobs that are expected to finish immediately
@@ -33,7 +48,7 @@ const getPriority = (job: savePageJob): number => {
   return job.priority === 'low' ? 10 : 1
 }
 
-const getAttempts = (job: savePageJob): number => {
+const getAttempts = (job: SavePageJob): number => {
   if (job.isRss || job.isImport) {
     // we don't want to retry rss or import jobs
     return 1
@@ -42,11 +57,11 @@ const getAttempts = (job: savePageJob): number => {
   return 3
 }
 
-const getOpts = (job: savePageJob): BulkJobOptions => {
+const getOpts = (job: SavePageJob): BulkJobOptions => {
   return {
-    // jobId: `${job.userId}-${job.url}`,
-    // removeOnComplete: true,
-    // removeOnFail: true,
+    jobId: `save-page_${job.userId}_${job.data.finalUrl}`, // make sure we don't have duplicate jobs
+    removeOnComplete: true,
+    removeOnFail: true,
     attempts: getAttempts(job),
     priority: getPriority(job),
     backoff: {
@@ -56,7 +71,7 @@ const getOpts = (job: savePageJob): BulkJobOptions => {
   }
 }
 
-export const queueSavePageJob = async (savePageJobs: savePageJob[]) => {
+export const queueSavePageJob = async (savePageJobs: SavePageJob[]) => {
   const jobs = savePageJobs.map((job) => ({
     name: JOB_NAME,
     data: job.data,