Merge pull request #4309 from omnivore-app/fix/content-fetch-http

fix: allow content-fetch to still process http requests

packages/api/src/queue-processor.ts

@@ -111,10 +111,10 @@ export const getQueue = async (
         delay: 2000, // 2 seconds
       },
       removeOnComplete: {
-        age: 24 * 3600, // keep up to 24 hours
+        age: 3600, // keep up to 1 hour
       },
       removeOnFail: {
-        age: 7 * 24 * 3600, // keep up to 7 days
+        age: 24 * 3600, // keep up to 1 day
       },
     },
   })

packages/content-fetch/src/app.ts

@@ -2,13 +2,21 @@ import { RedisDataSource } from '@omnivore/utils'
 import { JobType } from 'bullmq'
 import express, { Express } from 'express'
 import asyncHandler from 'express-async-handler'
+import { JobData, processFetchContentJob } from './request_handler'
 import { createWorker, getQueue, QUEUE } from './worker'
 
 const main = () => {
   console.log('Starting worker...')
 
+  if (!process.env.VERIFICATION_TOKEN) {
+    console.error('VERIFICATION_TOKEN is required')
+    process.exit(1)
+  }
+
   const app: Express = express()
-  const port = process.env.PORT || 3002
+
+  app.use(express.json())
+  app.use(express.urlencoded({ extended: true }))
 
   // create redis source
   const redisDataSource = new RedisDataSource({
@@ -75,6 +83,38 @@ const main = () => {
     })
   )
 
+  app.all(
+    '/',
+    asyncHandler(async (req, res) => {
+      console.log('Received http request')
+
+      if (req.method !== 'GET' && req.method !== 'POST') {
+        console.error('request method is not GET or POST')
+        res.sendStatus(405)
+        return
+      }
+
+      if (req.query.token !== process.env.VERIFICATION_TOKEN) {
+        console.error('query does not include valid token')
+        res.sendStatus(403)
+        return
+      }
+
+      try {
+        const data = <JobData>req.body
+        const attempt = parseInt(req.get('X-CloudTasks-TaskRetryCount') || '0')
+        await processFetchContentJob(redisDataSource, data, attempt)
+      } catch (error) {
+        console.error('Error fetching content', { error })
+        res.sendStatus(500)
+        return
+      }
+
+      res.sendStatus(200)
+    })
+  )
+
+  const port = process.env.PORT || 3002
   const server = app.listen(port, () => {
     console.log('Worker started')
   })

packages/content-fetch/src/worker.ts

@@ -16,10 +16,10 @@ export const getQueue = async (
         delay: 2000, // 2 seconds
       },
       removeOnComplete: {
-        age: 24 * 3600, // keep up to 24 hours
+        age: 3600, // keep up to 1 hour
       },
       removeOnFail: {
-        age: 7 * 24 * 3600, // keep up to 7 days
+        age: 24 * 3600, // keep up to 1 day
       },
     },
   })