diff --git a/packages/api/src/jobs/bulk_action.ts b/packages/api/src/jobs/bulk_action.ts index 56a38f282..cea355d6e 100644 --- a/packages/api/src/jobs/bulk_action.ts +++ b/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 } diff --git a/packages/api/src/jobs/integration/export_all_items.ts b/packages/api/src/jobs/integration/export_all_items.ts index f3bf8950d..f642f8e39 100644 --- a/packages/api/src/jobs/integration/export_all_items.ts +++ b/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 diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index 5f0fc6a21..89448b96f 100644 --- a/packages/api/src/queue-processor.ts +++ b/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) } diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts index 58cdafbda..87d12536a 100755 --- a/packages/api/src/server.ts +++ b/packages/api/src/server.ts @@ -178,9 +178,8 @@ const main = async (): Promise => { 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 diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 832aea3fa..47ec7bba8 100644 --- a/packages/api/src/services/library_item.ts +++ b/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 => { 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 ) diff --git a/packages/api/src/utils/highlightGenerator.ts b/packages/api/src/utils/highlightGenerator.ts index bfdd346c7..b3f4448ba 100644 --- a/packages/api/src/utils/highlightGenerator.ts +++ b/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 = '' diff --git a/packages/content-fetch/src/job.ts b/packages/content-fetch/src/job.ts index 8d88472f9..88c263765 100644 --- a/packages/content-fetch/src/job.ts +++ b/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,