diff --git a/packages/api/src/resolvers/integrations/index.ts b/packages/api/src/resolvers/integrations/index.ts index d976fa6a5..a893c3e79 100644 --- a/packages/api/src/resolvers/integrations/index.ts +++ b/packages/api/src/resolvers/integrations/index.ts @@ -245,8 +245,9 @@ export const importFromIntegrationResolver = authorized< authToken, integration.importItemState || ImportItemState.Unarchived ) - // update task name in integration - await updateIntegration(integration.id, { taskName }, uid) + log.info('task created', taskName) + // // update task name in integration + // await updateIntegration(integration.id, { taskName }, uid) analytics.capture({ distinctId: uid, diff --git a/packages/content-fetch/src/worker.ts b/packages/content-fetch/src/worker.ts index 99672497e..0d28d4f5a 100644 --- a/packages/content-fetch/src/worker.ts +++ b/packages/content-fetch/src/worker.ts @@ -40,10 +40,12 @@ export const createWorker = ( { connection: redisDataSource.queueRedisClient, autorun: true, // start processing jobs immediately + // process up to 10 jobs in a second limiter: { - max: 10, // process up to 10 jobs concurrently - duration: 1000, // 1 second + max: 10, + duration: 1000, }, + concurrency: 2, // process up to 2 jobs concurrently } ) diff --git a/packages/import-handler/src/job.ts b/packages/import-handler/src/job.ts index a47fe1dce..c44156cb0 100644 --- a/packages/import-handler/src/job.ts +++ b/packages/import-handler/src/job.ts @@ -1,12 +1,14 @@ import { RedisDataSource } from '@omnivore/utils' import { Queue } from 'bullmq' import { ArticleSavingRequestStatus } from '.' +import crypto from 'crypto' const BACKEND_QUEUE = 'omnivore-backend-queue' const CONTENT_FETCH_QUEUE = 'omnivore-content-fetch-queue' export const SEND_EMAIL_JOB = 'send-email' const FETCH_CONTENT_JOB = 'fetch-content' +const JOB_VERSION = 'v001' interface SendEmailJobData { userId: string @@ -30,6 +32,10 @@ interface FetchContentJobData { publishedAt?: string } +export const stringToHash = (str: string): string => { + return crypto.createHash('md5').update(str).digest('hex') +} + export const queueEmailJob = async ( redisDataSource: RedisDataSource, data: SendEmailJobData @@ -49,7 +55,15 @@ export const enqueueFetchContentJob = async ( connection: redisDataSource.queueRedisClient, }) + // sort the data to make sure the hash is consistent + const sortedData = JSON.stringify(data, Object.keys(data).sort()) + const jobId = `${FETCH_CONTENT_JOB}_${stringToHash( + sortedData + )}_${JOB_VERSION}` const job = await queue.add(FETCH_CONTENT_JOB, data, { + jobId, + removeOnComplete: true, + removeOnFail: true, priority: 100, attempts: 1, })