diff --git a/packages/pdf-handler/package.json b/packages/pdf-handler/package.json index 01ac1887f..12e20bcc3 100644 --- a/packages/pdf-handler/package.json +++ b/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": { diff --git a/packages/pdf-handler/src/index.ts b/packages/pdf-handler/src/index.ts index cf3f3359c..7e4ab43c6 100644 --- a/packages/pdf-handler/src/index.ts +++ b/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 => { - 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() } } diff --git a/packages/pdf-handler/src/job.ts b/packages/pdf-handler/src/job.ts index ca404aa0f..b9489ebf3 100644 --- a/packages/pdf-handler/src/job.ts +++ b/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, diff --git a/packages/pdf-handler/src/redis_data_source.ts b/packages/pdf-handler/src/redis_data_source.ts deleted file mode 100644 index 1c02c2a47..000000000 --- a/packages/pdf-handler/src/redis_data_source.ts +++ /dev/null @@ -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 { - 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) -}) diff --git a/packages/utils/.eslintignore b/packages/utils/.eslintignore new file mode 100644 index 000000000..b38db2f29 --- /dev/null +++ b/packages/utils/.eslintignore @@ -0,0 +1,2 @@ +node_modules/ +build/ diff --git a/packages/utils/.eslintrc b/packages/utils/.eslintrc new file mode 100644 index 000000000..644bb1aec --- /dev/null +++ b/packages/utils/.eslintrc @@ -0,0 +1,6 @@ +{ + "extends": "../../.eslintrc", + "parserOptions": { + "project": "tsconfig.json" + } +} diff --git a/packages/utils/mocha-config.json b/packages/utils/mocha-config.json new file mode 100644 index 000000000..8e24eb08b --- /dev/null +++ b/packages/utils/mocha-config.json @@ -0,0 +1,5 @@ +{ + "extension": ["ts"], + "spec": "test/**/*.test.ts", + "timeout": 10000 + } diff --git a/packages/utils/package.json b/packages/utils/package.json new file mode 100644 index 000000000..46cfe0cca --- /dev/null +++ b/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" + } +} diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts new file mode 100644 index 000000000..72c968fd3 --- /dev/null +++ b/packages/utils/src/index.ts @@ -0,0 +1 @@ +export * from './redis_data_source' diff --git a/packages/utils/src/redis_data_source.ts b/packages/utils/src/redis_data_source.ts new file mode 100644 index 000000000..3c982e90f --- /dev/null +++ b/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 { + 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) +} diff --git a/packages/utils/test/stub.test.ts b/packages/utils/test/stub.test.ts new file mode 100644 index 000000000..24ad25c8f --- /dev/null +++ b/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 + }) +}) diff --git a/packages/utils/tsconfig.json b/packages/utils/tsconfig.json new file mode 100644 index 000000000..fe4d0245b --- /dev/null +++ b/packages/utils/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "./../../tsconfig.json", + "compilerOptions": { + "declaration": true, + "outDir": "build", + }, + "include": ["src", "test"] +}