fix pdf handler sends update-content-job to the wrong redis

packages/pdf-handler/package.json

@@ -30,12 +30,12 @@
     "@google-cloud/functions-framework": "3.1.2",
     "@google-cloud/pubsub": "^4.0.0",
     "@google-cloud/storage": "^7.0.1",
+    "@omnivore/utils": "1.0.0",
     "@sentry/serverless": "^7.77.0",
     "axios": "^0.27.2",
     "bullmq": "^5.1.4",
     "concurrently": "^7.0.0",
     "dotenv": "^8.2.0",
-    "ioredis": "^5.3.2",
     "pdfjs-dist": "^2.9.359"
   },
   "volta": {

packages/pdf-handler/src/index.ts

@@ -1,7 +1,9 @@
 import { GetSignedUrlConfig, Storage } from '@google-cloud/storage'
+import { RedisDataSource } from '@omnivore/utils'
 import * as Sentry from '@sentry/serverless'
-import { parsePdf } from './pdf'
+import 'dotenv/config'
 import { queueUpdatePageJob, State } from './job'
+import { parsePdf } from './pdf'
 
 Sentry.GCPFunction.init({
   dsn: process.env.SENTRY_DSN,
@@ -49,6 +51,7 @@ const getDocumentUrl = async (
 }
 
 export const updatePageContent = async (
+  redisDataSource: RedisDataSource,
   fileId: string,
   content?: string,
   title?: string,
@@ -56,7 +59,7 @@ export const updatePageContent = async (
   description?: string,
   state?: State
 ): Promise<string | undefined> => {
-  const job = await queueUpdatePageJob({
+  const job = await queueUpdatePageJob(redisDataSource, {
     fileId,
     content,
     title,
@@ -106,6 +109,17 @@ export const pdfHandler = Sentry.GCPFunction.wrapHttpFunction(
         description,
         state: State = 'SUCCEEDED' // Default to succeeded even if we fail to parse
 
+      const redisDataSource = new RedisDataSource({
+        cache: {
+          url: process.env.REDIS_URL,
+          cert: process.env.REDIS_CERT,
+        },
+        mq: {
+          url: process.env.MQ_REDIS_URL,
+          cert: process.env.MQ_REDIS_CERT,
+        },
+      })
+
       try {
         const url = await getDocumentUrl(data)
         console.log('PDF url: ', url)
@@ -130,6 +144,7 @@ export const pdfHandler = Sentry.GCPFunction.wrapHttpFunction(
       } finally {
         // Always update the state, even if we fail to parse
         const result = await updatePageContent(
+          redisDataSource,
           data.name,
           content,
           title,
@@ -147,6 +162,8 @@ export const pdfHandler = Sentry.GCPFunction.wrapHttpFunction(
           'state',
           state
         )
+
+        await redisDataSource.shutdown()
       }
     }
 

packages/pdf-handler/src/job.ts

@@ -1,13 +1,9 @@
+import { RedisDataSource } from '@omnivore/utils'
 import { Queue } from 'bullmq'
-import { redisDataSource } from './redis_data_source'
 
 const QUEUE_NAME = 'omnivore-backend-queue'
 const JOB_NAME = 'update-pdf-content'
 
-const queue = new Queue(QUEUE_NAME, {
-  connection: redisDataSource.queueRedisClient,
-})
-
 export type State = 'SUCCEEDED' | 'FAILED'
 
 type UpdatePageJobData = {
@@ -19,7 +15,14 @@ type UpdatePageJobData = {
   state?: State
 }
 
-export const queueUpdatePageJob = async (data: UpdatePageJobData) => {
+export const queueUpdatePageJob = async (
+  redisDataSource: RedisDataSource,
+  data: UpdatePageJobData
+) => {
+  const queue = new Queue(QUEUE_NAME, {
+    connection: redisDataSource.queueRedisClient,
+  })
+
   return queue.add(JOB_NAME, data, {
     priority: 5,
     attempts: 3,

packages/pdf-handler/src/redis_data_source.ts

@@ -1,89 +0,0 @@
-import Redis, { RedisOptions } from 'ioredis'
-import 'dotenv/config'
-
-export type RedisDataSourceOptions = {
-  REDIS_URL?: string
-  REDIS_CERT?: string
-}
-
-export class RedisDataSource {
-  options: RedisDataSourceOptions
-
-  cacheClient: Redis
-  queueRedisClient: Redis
-
-  constructor(options: RedisDataSourceOptions) {
-    this.options = options
-
-    this.cacheClient = createRedisClient('cache', this.options)
-    this.queueRedisClient = createRedisClient('queue', this.options)
-  }
-
-  setOptions(options: RedisDataSourceOptions): void {
-    this.options = options
-  }
-
-  async shutdown(): Promise<void> {
-    try {
-      await this.queueRedisClient?.quit()
-      await this.cacheClient?.quit()
-    } catch (err) {
-      console.error('error while shutting down redis', err)
-    }
-  }
-}
-
-const createRedisClient = (name: string, options: RedisDataSourceOptions) => {
-  const redisURL = options.REDIS_URL
-  const cert = options.REDIS_CERT?.replace(/\\n/g, '\n') // replace \n with new line
-  if (!redisURL) {
-    throw 'Error: no redisURL supplied'
-  }
-
-  const redisOptions: RedisOptions = {
-    name,
-    connectTimeout: 10000, // 10 seconds
-    tls: cert
-      ? {
-          cert,
-          rejectUnauthorized: false, // for self-signed certs
-        }
-      : undefined,
-    maxRetriesPerRequest: null,
-    offlineQueue: false,
-  }
-
-  const redis = new Redis(redisURL, redisOptions)
-
-  redis.on('connect', () => {
-    console.log('Redis connected', name)
-  })
-
-  redis.on('error', (err) => {
-    console.error('Redis error', err, name)
-  })
-
-  redis.on('close', () => {
-    console.log('Redis closed', name)
-  })
-
-  return redis
-}
-
-export const redisDataSource = new RedisDataSource({
-  REDIS_URL: process.env.REDIS_URL,
-  REDIS_CERT: process.env.REDIS_CERT,
-})
-
-// eslint-disable-next-line @typescript-eslint/no-misused-promises
-process.on('SIGINT', async () => {
-  console.log('SIGINT signal received.')
-
-  try {
-    await redisDataSource.shutdown()
-  } catch (error) {
-    console.error('error while shutting down redis', error)
-  }
-
-  process.exit(0)
-})

packages/utils/.eslintignore

@@ -0,0 +1,2 @@
+node_modules/
+build/

packages/utils/.eslintrc

@@ -0,0 +1,6 @@
+{
+  "extends": "../../.eslintrc",
+  "parserOptions": {
+    "project": "tsconfig.json"
+  }
+}

packages/utils/mocha-config.json

@@ -0,0 +1,5 @@
+{
+    "extension": ["ts"],
+    "spec": "test/**/*.test.ts",
+    "timeout": 10000
+  }

packages/utils/package.json

@@ -0,0 +1,25 @@
+{
+  "name": "@omnivore/utils",
+  "version": "1.0.0",
+  "description": "Utility functions for Omnivore packages.",
+  "main": "./build/src/index.js",
+  "types": "./build/src/index.d.ts",
+  "scripts": {
+    "test": "yarn mocha -r ts-node/register --config mocha-config.json",
+    "lint": "eslint src --ext ts,js,tsx,jsx",
+    "build": "tsc"
+  },
+  "devDependencies": {
+    "@types/chai": "^4.3.6",
+    "@types/mocha": "^10.0.0",
+    "chai": "^4.3.6",
+    "eslint-plugin-prettier": "^4.0.0",
+    "mocha": "^10.0.0"
+  },
+  "dependencies": {
+    "ioredis": "^5.3.2"
+  },
+  "volta": {
+    "extends": "../../package.json"
+  }
+}

packages/utils/src/index.ts

@@ -0,0 +1 @@
+export * from './redis_data_source'

packages/utils/src/redis_data_source.ts

@@ -0,0 +1,69 @@
+import Redis, { RedisOptions } from 'ioredis'
+
+type RedisClientType = 'cache' | 'mq'
+type RedisDataSourceOption = {
+  url?: string
+  cert?: string
+}
+export type RedisDataSourceOptions = {
+  [key in RedisClientType]: RedisDataSourceOption
+}
+
+export class RedisDataSource {
+  options: RedisDataSourceOptions
+
+  cacheClient: Redis
+  queueRedisClient: Redis
+
+  constructor(options: RedisDataSourceOptions) {
+    this.options = options
+
+    const cacheClient = createIORedisClient('cache', this.options)
+    if (!cacheClient) throw 'Error initializing cache redis client'
+
+    this.cacheClient = cacheClient
+    this.queueRedisClient =
+      createIORedisClient('mq', this.options) || this.cacheClient // if mq is not defined, use cache
+  }
+
+  async shutdown(): Promise<void> {
+    try {
+      await this.queueRedisClient?.quit()
+      await this.cacheClient?.quit()
+
+      console.log('redis shutdown complete')
+    } catch (err) {
+      console.error('error while shutting down redis', err)
+    }
+  }
+}
+
+const createIORedisClient = (
+  name: RedisClientType,
+  options: RedisDataSourceOptions
+): Redis | undefined => {
+  const option = options[name]
+  const redisURL = option.url
+  if (!redisURL) {
+    console.log(`no redisURL supplied: ${name}`)
+    return undefined
+  }
+
+  const redisCert = option.cert
+  const tls =
+    redisURL.startsWith('rediss://') && redisCert
+      ? {
+          ca: redisCert,
+          rejectUnauthorized: false,
+        }
+      : undefined
+
+  const redisOptions: RedisOptions = {
+    tls,
+    name,
+    connectTimeout: 10000,
+    maxRetriesPerRequest: null,
+    offlineQueue: false,
+  }
+  return new Redis(redisURL, redisOptions)
+}

packages/utils/test/stub.test.ts

@@ -0,0 +1,8 @@
+import 'mocha'
+import { expect } from 'chai'
+
+describe('stub test', () => {
+  it('should pass', () => {
+    expect(true).to.be.true
+  })
+})

packages/utils/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "./../../tsconfig.json",
+  "compilerOptions": {
+    "declaration": true,
+    "outDir": "build",
+  },
+  "include": ["src", "test"]
+}